APIE-844 - Confluent CLI support for kstreams

cmd/lint/en_US.dic

@@ -1,4 +1,4 @@
-48977
+48979
 0/nm
 0th/pt
 1/n1
@@ -43627,6 +43627,7 @@ subterfuge/SM
 subterranean
 subtext/SM
 subtitle/DSMG
+subtopology
 subtle/TR
 subtlety/SM
 subtly
@@ -47761,6 +47762,7 @@ warmness/M
 warmonger/SMG
 warmongering/M
 warmth/M
+warmup
 warn/JDGS
 warning/M
 warp/MDGS

go.mod

@@ -35,7 +35,7 @@ require (
 	github.com/confluentinc/ccloud-sdk-go-v2/iam-ip-filtering v0.5.0
 	github.com/confluentinc/ccloud-sdk-go-v2/identity-provider v0.3.0
 	github.com/confluentinc/ccloud-sdk-go-v2/kafka-quotas v0.4.0
-	github.com/confluentinc/ccloud-sdk-go-v2/kafkarest v0.0.0-20250909043602-f80bee0eb280
+	github.com/confluentinc/ccloud-sdk-go-v2/kafkarest v0.25.0
 	github.com/confluentinc/ccloud-sdk-go-v2/ksql v0.2.0
 	github.com/confluentinc/ccloud-sdk-go-v2/mds v0.4.0
 	github.com/confluentinc/ccloud-sdk-go-v2/metrics v0.2.0

go.sum

@@ -230,6 +230,8 @@ github.com/confluentinc/ccloud-sdk-go-v2/kafka-quotas v0.4.0 h1:T9e7lNj/VjxE89+t
 github.com/confluentinc/ccloud-sdk-go-v2/kafka-quotas v0.4.0/go.mod h1:7gqwWFIyj2MAGpL/kf6SGXm/pi2Z6qpMJIjKlgEEhhg=
 github.com/confluentinc/ccloud-sdk-go-v2/kafkarest v0.0.0-20250909043602-f80bee0eb280 h1:GFVI3pGckhpP66Xb05usB8txzubnnoigZHp292ax5Rg=
 github.com/confluentinc/ccloud-sdk-go-v2/kafkarest v0.0.0-20250909043602-f80bee0eb280/go.mod h1:b8v8EIBtpQDx0zAxCpGxhuSWBRAwh/+PRFNtaBR5P7c=
+github.com/confluentinc/ccloud-sdk-go-v2/kafkarest v0.25.0 h1:sV+PXI3jyEbqzyL8dS+v8DprMdwRD87vICegjgo1WBU=
+github.com/confluentinc/ccloud-sdk-go-v2/kafkarest v0.25.0/go.mod h1:KZa40uDMwQlwKJQtj44cjTuvwLnE/wgcPfYPFq+tD9A=
 github.com/confluentinc/ccloud-sdk-go-v2/ksql v0.2.0 h1:g6OHa1iW3HO3N/YiTAL9Q6Y7rdjMBAjOPYK37akTt0M=
 github.com/confluentinc/ccloud-sdk-go-v2/ksql v0.2.0/go.mod h1:0LAvd4VqlaRwKU4yvDEkVCtV43yNezt56+hBe9Lmg7Q=
 github.com/confluentinc/ccloud-sdk-go-v2/mds v0.4.0 h1:jIXXhGi+Xn+XYFCErnMvd035QijbYXla1Bo8W7V7lFM=

internal/kafka/command.go

@@ -25,6 +25,7 @@ func New(cfg *config.Config, prerunner pcmd.PreRunner) *cobra.Command {
 	cmd.AddCommand(newRegionCommand(prerunner))
 	cmd.AddCommand(newReplicaCommand(prerunner))
 	cmd.AddCommand(newShareGroupCommand(prerunner))
+	cmd.AddCommand(newStreamsGroupCommand(prerunner))
 	cmd.AddCommand(newTopicCommand(cfg, prerunner))
 
 	return cmd

internal/kafka/command_consumer_group.go

@@ -14,6 +14,7 @@ type consumerGroupOut struct {
 	IsSimple          bool   `human:"Simple" serialized:"is_simple"`
 	PartitionAssignor string `human:"Partition Assignor" serialized:"partition_assignor"`
 	State             string `human:"State" serialized:"state"`
+	ProtocolType      string `human:"Type,omitempty" serialized:"type,omitempty"`
 }
 
 func (c *consumerCommand) newGroupCommand(cfg *config.Config) *cobra.Command {

internal/kafka/command_consumer_group_describe.go

@@ -46,6 +46,7 @@ func (c *consumerCommand) groupDescribe(cmd *cobra.Command, args []string) error
 		IsSimple:          group.GetIsSimple(),
 		PartitionAssignor: group.GetPartitionAssignor(),
 		State:             group.GetState(),
+		ProtocolType:      group.GetType(),
 	})
 	return table.Print()
 }

internal/kafka/command_consumer_group_list.go

@@ -44,6 +44,7 @@ func (c *consumerCommand) groupList(cmd *cobra.Command, _ []string) error {
 			IsSimple:          group.GetIsSimple(),
 			PartitionAssignor: group.GetPartitionAssignor(),
 			State:             group.GetState(),
+			ProtocolType:      group.GetType(),
 		})
 	}
 	return list.Print()

internal/kafka/command_kafka_streams_group.go

@@ -0,0 +1,57 @@
+package kafka
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+)
+
+type streamsGroupCommand struct {
+	*pcmd.AuthenticatedCLICommand
+}
+
+type streamsGroupOut struct {
+	ClusterId             string `human:"Cluster Id" serialized:"cluster_id"`
+	GroupId               string `human:"Group Id" serialized:"group_id"`
+	State                 string `human:"State" serialized:"state"`
+	MemberCount           int32  `human:"Member Count" serialized:"member_count"`
+	SubtopologyCount      int32  `human:"Subtopology Count" serialized:"subtopology_count"`
+	GroupEpoch            int32  `human:"Group Epoch" serialized:"group_epoch"`
+	TopologyEpoch         int32  `human:"Topology Epoch" serialized:"topology_epoch"`
+	TargetAssignmentEpoch int32  `human:"Target Assignment Epoch" serialized:"target_assignment_epoch"`
+	Members               string `human:"Members" serialized:"members"`
+	Subtopologies         string `human:"Subtopologies" serialized:"subtopologies"`
+}
+
+func newStreamsGroupCommand(prerunner pcmd.PreRunner) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "streams-group",
+		Short:       "Manage Kafka streams groups.",
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogin},
+	}
+
+	c := &streamsGroupCommand{pcmd.NewAuthenticatedCLICommand(cmd, prerunner)}
+
+	cmd.AddCommand(c.newStreamsGroupDescribeCommand())
+	cmd.AddCommand(c.newStreamsGroupListCommand())
+	cmd.AddCommand(c.newStreamsGroupMemberAssignmentCommand())
+	cmd.AddCommand(c.newStreamsGroupMemberCommand())
+	cmd.AddCommand(c.newStreamsGroupMemberTargetAssignmentCommand())
+	cmd.AddCommand(c.newStreamsGroupMemberTargetAssignmentTaskPartitionsCommand())
+	cmd.AddCommand(c.newStreamsGroupMemberTaskPartitionsCommand())
+	cmd.AddCommand(c.newStreamsGroupSubtopologyCommand())
+
+	return cmd
+}
+
+func (c *streamsGroupCommand) validStreamsGroupArgs(cmd *cobra.Command, args []string) []string {
+	if len(args) > 0 {
+		return nil
+	}
+
+	if err := c.PersistentPreRunE(cmd, args); err != nil {
+		return nil
+	}
+
+	return pcmd.AutocompleteStreamsGroups(cmd, c.AuthenticatedCLICommand)
+}

internal/kafka/command_kafka_streams_group_describe.go

@@ -0,0 +1,56 @@
+package kafka
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/output"
+)
+
+func (c *streamsGroupCommand) newStreamsGroupDescribeCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:               "describe <group>",
+		Short:             "Describe a Kafka streams group.",
+		Args:              cobra.ExactArgs(1),
+		ValidArgsFunction: pcmd.NewValidArgsFunction(c.validStreamsGroupArgs),
+		RunE:              c.streamsGroupDescribe,
+	}
+
+	pcmd.AddEndpointFlag(cmd, c.AuthenticatedCLICommand)
+	pcmd.AddClusterFlag(cmd, c.AuthenticatedCLICommand)
+	pcmd.AddContextFlag(cmd, c.CLICommand)
+	pcmd.AddEnvironmentFlag(cmd, c.AuthenticatedCLICommand)
+	pcmd.AddOutputFlag(cmd)
+
+	return cmd
+}
+
+func (c *streamsGroupCommand) streamsGroupDescribe(cmd *cobra.Command, args []string) error {
+	groupId := args[0]
+
+	kafkaREST, err := c.GetKafkaREST(cmd)
+	if err != nil {
+		return err
+	}
+
+	streamsGroup, err := kafkaREST.CloudClient.GetKafkaStreamsGroup(groupId)
+	if err != nil {
+		return err
+	}
+
+	table := output.NewTable(cmd)
+	table.Add(&streamsGroupOut{
+		ClusterId:             streamsGroup.GetClusterId(),
+		GroupId:               streamsGroup.GetGroupId(),
+		State:                 streamsGroup.GetState(),
+		MemberCount:           streamsGroup.GetMemberCount(),
+		SubtopologyCount:      streamsGroup.GetSubtopologyCount(),
+		GroupEpoch:            streamsGroup.GetGroupEpoch(),
+		TopologyEpoch:         streamsGroup.GetTopologyEpoch(),
+		TargetAssignmentEpoch: streamsGroup.GetTargetAssignmentEpoch(),
+		Members:               streamsGroup.Members.GetRelated(),
+		Subtopologies:         streamsGroup.Subtopologies.GetRelated(),
+	})
+
+	return table.Print()
+}

internal/kafka/command_kafka_streams_group_list.go

@@ -0,0 +1,65 @@
+package kafka
+
+import (
+	"github.com/spf13/cobra"
+
+	kafkarestv3 "github.com/confluentinc/ccloud-sdk-go-v2/kafkarest/v3"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/output"
+)
+
+func (c *streamsGroupCommand) newStreamsGroupListCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "list",
+		Short: "List Kafka streams groups.",
+		Args:  cobra.NoArgs,
+		RunE:  c.listStreamsGroups,
+	}
+
+	pcmd.AddEndpointFlag(cmd, c.AuthenticatedCLICommand)
+	pcmd.AddClusterFlag(cmd, c.AuthenticatedCLICommand)
+	pcmd.AddContextFlag(cmd, c.CLICommand)
+	pcmd.AddEnvironmentFlag(cmd, c.AuthenticatedCLICommand)
+	pcmd.AddOutputFlag(cmd)
+
+	return cmd
+}
+
+func (c *streamsGroupCommand) listStreamsGroups(cmd *cobra.Command, _ []string) error {
+	groups, err := c.getStreamsGroups(cmd)
+	if err != nil {
+		return err
+	}
+
+	list := output.NewList(cmd)
+	for _, streamsGroup := range groups {
+		list.Add(&streamsGroupOut{
+			ClusterId:             streamsGroup.GetClusterId(),
+			GroupId:               streamsGroup.GetGroupId(),
+			State:                 streamsGroup.GetState(),
+			MemberCount:           streamsGroup.GetMemberCount(),
+			SubtopologyCount:      streamsGroup.GetSubtopologyCount(),
+			GroupEpoch:            streamsGroup.GetGroupEpoch(),
+			TopologyEpoch:         streamsGroup.GetTopologyEpoch(),
+			TargetAssignmentEpoch: streamsGroup.GetTargetAssignmentEpoch(),
+			Members:               streamsGroup.Members.GetRelated(),
+			Subtopologies:         streamsGroup.Subtopologies.GetRelated(),
+		})
+	}
+	return list.Print()
+}
+
+func (c *streamsGroupCommand) getStreamsGroups(cmd *cobra.Command) ([]kafkarestv3.StreamsGroupData, error) {
+	kafkaREST, err := c.GetKafkaREST(cmd)
+	if err != nil {
+		return nil, err
+	}
+
+	groups, err := kafkaREST.CloudClient.ListKafkaStreamsGroups()
+	if err != nil {
+		return nil, err
+	}
+
+	return groups.Data, nil
+}

internal/kafka/command_kafka_streams_group_member.go

@@ -0,0 +1,32 @@
+package kafka
+
+import (
+	"github.com/spf13/cobra"
+)
+
+type streamsGroupMemberOut struct {
+	Kind          string `human:"Kind" serialized:"kind"`
+	ClusterId     string `human:"Cluster Id" serialized:"cluster_id"`
+	GroupId       string `human:"Group Id" serialized:"group_id"`
+	MemberId      string `human:"Member Id" serialized:"member_id"`
+	ProcessId     string `human:"Process Id" serialized:"process_id"`
+	ClientId      string `human:"Client Id" serialized:"client_id"`
+	InstanceId    string `human:"Instance Id" serialized:"instance_id"`
+	MemberEpoch   int32  `human:"Member Epoch" serialized:"member_epoch"`
+	TopologyEpoch int32  `human:"Topology Epoch" serialized:"topology_epoch"`
+	IsClassic     bool   `human:"Is Classic" serialized:"is_classic"`
+	Assignments   string `human:"Assignments" serialized:"assignments"`
+	TargetAssign  string `human:"Target Assignment" serialized:"target_assignment"`
+}
+
+func (c *streamsGroupCommand) newStreamsGroupMemberCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "member",
+		Short: "Manage Kafka streams group members.",
+	}
+
+	cmd.AddCommand(c.newStreamsGroupMemberDescribeCommand())
+	cmd.AddCommand(c.newStreamsGroupMemberListCommand())
+
+	return cmd
+}

internal/kafka/command_kafka_streams_group_member_assignment.go

@@ -0,0 +1,28 @@
+package kafka
+
+import (
+	"github.com/spf13/cobra"
+)
+
+type streamsGroupMemberAssignmentOut struct {
+	Kind         string `human:"Kind" serialized:"kind"`
+	ClusterId    string `human:"Cluster Id" serialized:"cluster_id"`
+	GroupId      string `human:"Group Id" serialized:"group_id"`
+	MemberId     string `human:"Member Id" serialized:"member_id"`
+	ActiveTasks  string `human:"Active Tasks" serialized:"active_tasks"`
+	StandbyTasks string `human:"Standby Tasks" serialized:"standby_tasks"`
+	WarmupTasks  string `human:"Warmup Tasks" serialized:"warmup_tasks"`
+}
+
+func (c *streamsGroupCommand) newStreamsGroupMemberAssignmentCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:     "member-assignment",
+		Aliases: []string{"ma"},
+		Short:   "Manage Kafka streams group member assignments.",
+	}
+
+	cmd.AddCommand(c.newStreamsGroupMemberAssignmentDescribeCommand())
+	cmd.AddCommand(c.newStreamsGroupMemberAssignmentListCommand())
+
+	return cmd
+}

internal/kafka/command_kafka_streams_group_member_assignment_describe.go

@@ -0,0 +1,61 @@
+package kafka
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/output"
+)
+
+func (c *streamsGroupCommand) newStreamsGroupMemberAssignmentDescribeCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "describe <member>",
+		Short: "Describe a Kafka streams group member assignment.",
+		Args:  cobra.ExactArgs(1),
+		RunE:  c.streamsGroupMemberAssignmentDescribe,
+	}
+
+	cmd.Flags().String("group", "", "Group Id.")
+
+	pcmd.AddEndpointFlag(cmd, c.AuthenticatedCLICommand)
+	pcmd.AddClusterFlag(cmd, c.AuthenticatedCLICommand)
+	pcmd.AddContextFlag(cmd, c.CLICommand)
+	pcmd.AddEnvironmentFlag(cmd, c.AuthenticatedCLICommand)
+	pcmd.AddOutputFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("group"))
+
+	return cmd
+}
+
+func (c *streamsGroupCommand) streamsGroupMemberAssignmentDescribe(cmd *cobra.Command, args []string) error {
+	groupId, err := cmd.Flags().GetString("group")
+	if err != nil {
+		return err
+	}
+
+	memberId := args[0]
+
+	kafkaREST, err := c.GetKafkaREST(cmd)
+	if err != nil {
+		return err
+	}
+
+	assignment, err := kafkaREST.CloudClient.GetKafkaStreamsGroupMemberAssignment(groupId, memberId)
+	if err != nil {
+		return err
+	}
+
+	table := output.NewTable(cmd)
+	table.Add(&streamsGroupMemberAssignmentOut{
+		Kind:         assignment.GetKind(),
+		ClusterId:    assignment.GetClusterId(),
+		GroupId:      assignment.GetGroupId(),
+		MemberId:     assignment.GetMemberId(),
+		ActiveTasks:  assignment.ActiveTasks.GetRelated(),
+		StandbyTasks: assignment.StandbyTasks.GetRelated(),
+		WarmupTasks:  assignment.WarmupTasks.GetRelated(),
+	})
+
+	return table.Print()
+}