add list consumer groups under a consumer group command

Author: ys-achinta

cmd/consumergroup.go

@@ -0,0 +1,22 @@
+package cmd
+
+import (
+	"github.com/gojek/kat/cmd/list"
+	"github.com/gojek/kat/logger"
+	"github.com/spf13/cobra"
+)
+
+var consumerGroupCmd = &cobra.Command{
+	Use:   "consumergroup",
+	Short: "Admin commands on consumergroups",
+}
+
+func init() {
+	consumerGroupCmd.PersistentFlags().StringP("broker-list", "b", "", "Comma separated list of broker ips")
+	consumerGroupCmd.PersistentFlags().StringP("topic", "t", "", "Specify topic")
+	if err := consumerGroupCmd.MarkPersistentFlagRequired("broker-list"); err != nil {
+		logger.Fatal(err)
+	}
+
+	consumerGroupCmd.AddCommand(list.ListConsumerGroupsCmd)
+}

cmd/list/listconsumergroup.go

@@ -1,10 +1,10 @@
 package list
 
 import (
+	"sort"
 	"strings"
 
 	"github.com/gojek/kat/cmd/base"
-	"github.com/gojek/kat/logger"
 	"github.com/gojek/kat/pkg/client"
 	"github.com/spf13/cobra"
 )
@@ -28,20 +28,21 @@ var ListConsumerGroupsCmd = &cobra.Command{
 	},
 }
 
-func init() {
-	ListConsumerGroupsCmd.PersistentFlags().StringP("broker-list", "b", "", "Comma separated list of broker ips")
-	ListConsumerGroupsCmd.PersistentFlags().StringP("topic", "t", "", "Specify topic")
-	if err := ListConsumerGroupsCmd.MarkPersistentFlagRequired("broker-list"); err != nil {
-		logger.Fatal(err)
-	}
-}
-
 func (c *consumerGroupAdmin) ListGroups(topic string) error {
 	consumerGroupsMap, err := c.saramaClient.ListConsumerGroups()
 	if err != nil {
 		return err
 	}
 
+	consumerGroupList := make([]string, len(consumerGroupsMap))
+	for group := range consumerGroupsMap {
+		consumerGroupList = append(consumerGroupList, group)
+	}
+
+	sort.Slice(consumerGroupList, func(i int, j int) bool {
+		return consumerGroupList[i] < consumerGroupList[j]
+	})
+
 	var consumerGroups []string
 
 	for consumerGroupID := range consumerGroupsMap {

cmd/root.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"os"
 
-	"github.com/gojek/kat/cmd/list"
 	"github.com/gojek/kat/cmd/mirror"
 
 	"github.com/gojek/kat/logger"
@@ -21,7 +20,7 @@ func init() {
 	cobra.OnInitialize()
 	cliCmd.AddCommand(topicCmd)
 	cliCmd.AddCommand(mirror.MirrorCmd)
-	cliCmd.AddCommand(list.ListConsumerGroupsCmd)
+	cliCmd.AddCommand(consumerGroupCmd)
 }
 
 func Execute() {

go.mod

@@ -11,12 +11,13 @@ require (
 	github.com/mattn/go-runewidth v0.0.5 // indirect
 	github.com/mitchellh/go-homedir v1.1.0
 	github.com/olekukonko/tablewriter v0.0.1
-	github.com/pkg/errors v0.9.1 // indirect
+	github.com/pkg/errors v0.9.1
 	github.com/r3labs/diff v0.0.0-20191018104334-e3ae93f4edbb
 	github.com/sirupsen/logrus v1.4.2
 	github.com/spf13/cobra v0.0.3
 	github.com/spf13/pflag v1.0.3 // indirect
 	github.com/stretchr/testify v1.3.0
 	golang.org/x/crypto v0.0.0-20191029031824-8986dd9e96cf
 	golang.org/x/net v0.0.0-20191028085509-fe3aa8a45271 // indirect
+	gotest.tools v2.2.0+incompatible
 )

go.sum

@@ -48,6 +48,8 @@ github.com/olekukonko/tablewriter v0.0.1 h1:b3iUnf1v+ppJiOfNX4yxxqfWKMQPZR5yoh8u
 github.com/olekukonko/tablewriter v0.0.1/go.mod h1:vsDQFd/mU46D+Z4whnwzcISnGGzXWMclvtLoiIKAKIo=
 github.com/pierrec/lz4 v2.2.6+incompatible h1:6aCX4/YZ9v8q69hTyiR7dNLnTA3fgtKHVVW5BCd5Znw=
 github.com/pierrec/lz4 v2.2.6+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/r3labs/diff v0.0.0-20191018104334-e3ae93f4edbb h1:kaV32NbiIn7ESdHB4PEW2VTKhB0odk9wo4/yW2acmoo=
@@ -95,3 +97,5 @@ gopkg.in/jcmturner/gokrb5.v7 v7.2.3 h1:hHMV/yKPwMnJhPuPx7pH2Uw/3Qyf+thJYlisUc440
 gopkg.in/jcmturner/gokrb5.v7 v7.2.3/go.mod h1:l8VISx+WGYp+Fp7KRbsiUuXTTOnxIc3Tuvyavf11/WM=
 gopkg.in/jcmturner/rpc.v1 v1.1.0 h1:QHIUxTX1ISuAv9dD2wJ9HWQVuWDX/Zc0PfeC2tjc4rU=
 gopkg.in/jcmturner/rpc.v1 v1.1.0/go.mod h1:YIdkC4XfD6GXbzje11McwsDuOlZQSb9W4vfLvuNnlv8=
+gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo=
+gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=

pkg/client/saramaclient.go

@@ -80,6 +80,17 @@ func (s *SaramaClient) GetConsumerGroupsForTopic(groups []string, topic string)
 			groupDescription, _ := s.admin.DescribeConsumerGroups([]string{groups[i]})
 			for _, memberDesc := range groupDescription[0].Members {
 				ma, _ := memberDesc.GetMemberAssignment()
+				fmt.Println("----version----")
+				fmt.Println(ma.Version)
+				fmt.Println("----version----")
+
+				fmt.Println("----userdata----")
+				fmt.Println(ma.UserData)
+				fmt.Println("----userdata----")
+
+				fmt.Println("----topics----")
+				fmt.Println(ma.Topics)
+				fmt.Println("----topics----")
 				for topicName, _ := range ma.Topics {
 					if topicName == topic {
 						consumerGroupsChannel <- groupDescription[0].GroupId

pkg/client/saramaclient_test.go

@@ -270,36 +270,33 @@ func TestSaramaClient_CreatePartitionsFailure(t *testing.T) {
 	assert.Error(t, err)
 	admin.AssertExpectations(t)
 }
-
-type mockGroupDescription interface {
-	GetMemberAssignment() (*sarama.ConsumerGroupMemberAssignment, error)
-}
-
-type mockMemberAssignment struct{}
-
-func (m *mockMemberAssignment) GetMemberAssignment() (*sarama.ConsumerGroupMemberAssignment, error) {
-	return &sarama.ConsumerGroupMemberAssignment{
-		Version:  1,
-		Topics:   map[string][]int32{"t1": {0, 1, 2}},
-		UserData: []byte("ss"),
-	}, nil
-}
-
-type mockGroupDescriptionStruct struct {
-	GroupId string
-	Members map[string]mockMemberAssignment
-}
-
 func TestSaramaClient_GetConsumerGroupsForTopic(t *testing.T) {
 	admin := &MockClusterAdmin{}
 	client := SaramaClient{admin: admin}
 
-	admin.On("DescribeConsumerGroups", []string{"c1"}).Return([]*mockGroupDescriptionStruct{{
-		GroupId: "c1",
-		Members: map[string]mockMemberAssignment{"k1": {}},
-	}}, nil)
+	groupDesciption := []*sarama.GroupDescription{{
+		GroupId: "test-group-id",
+		Members: map[string]*sarama.GroupMemberDescription{
+			"instance-id-0": {
+				ClientId:         "instance-id-0",
+				MemberAssignment: []byte{0x04, 0x05, 0x06},
+			},
+
+			"instance-id-1": {
+				ClientId:         "instance-id-1",
+				MemberAssignment: []byte{0x04, 0x05, 0x06},
+			},
+
+			"instance-id-2": {
+				ClientId:         "instance-id-2",
+				MemberAssignment: []byte{0x04, 0x05, 0x06},
+			},
+		},
+	}}
+
+	admin.On("DescribeConsumerGroups", []string{"test-group-id"}).Return(groupDesciption, nil)
 
-	consumerGroupChannel, err := client.GetConsumerGroupsForTopic([]string{"c1"}, "t1")
+	consumerGroupChannel, err := client.GetConsumerGroupsForTopic([]string{"test-group-id"}, "test-topic")
 
 	require.NoError(t, err)
 	assert.Equal(t, 1, len(consumerGroupChannel))