[9.0](backport #44576) [metricbeat][kafka] fix panic when fetching consumergroup member assignments (#44872)

* [metricbeat][kafka] fix panic when fetching consumergroup member assignments (#44576)

* [metricbeat][kafka] fix panic when fetching consumergroup member assignments

* add changelog entry

* mage fmt

* fix linter

* update new changelog

* fix behaviour in case of sarama error in MemberDescription conversion

* fix linter

* move changelog entry to appropriate version section

* remove entries from  docs/release-notes/index.md

---------

Co-authored-by: Ishleen Kaur <102962586+ishleenk17@users.noreply.github.com>
(cherry picked from commit 91de405)

# Conflicts:
#	metricbeat/module/kafka/broker_test.go

* resolve merge conflicts

* remove unrelated changelog entries

---------

Co-authored-by: stefans-elastic <stefan.stas@elastic.co>

Author: mergify[bot]

CHANGELOG.next.asciidoc

@@ -155,6 +155,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Fix Azure Monitor metric timespan to restore Storage Account PT1H metrics {issue}40376[40376] {pull}40367[40367]
 - Remove excessive info-level logs in cgroups setup {pull}40491[40491]
 - Add missing ECS Cloud fields in GCP `metrics` metricset when using `exclude_labels: true` {issue}40437[40437] {pull}40467[40467]
+- Fix panic in kafka consumergroup member assignment fetching when there are 0 members in consumer group. {pull}44576[44576]
 - Upgrade `go.mongodb.org/mongo-driver` from `v1.14.0` to `v1.17.4` to fix connection leaks in MongoDB module {pull}44769[44769]
 
 *Osquerybeat*

metricbeat/module/kafka/broker.go

@@ -138,7 +138,7 @@ func (b *Broker) Connect() error {
 	other := finder.findBroker(brokerAddress(b.broker), meta.Brokers)
 	if other == nil { // no broker found
 		closeBroker(b.broker)
-		return fmt.Errorf("No advertised broker with address %v found", b.Addr())
+		return fmt.Errorf("no advertised broker with address %v found", b.Addr())
 	}
 
 	debugf("found matching broker %v with id %v", other.Addr(), other.ID())
@@ -244,21 +244,11 @@ func (b *Broker) DescribeGroups(
 
 		members := map[string]MemberDescription{}
 		for memberID, memberDescr := range descr.Members {
-			assignment, err := memberDescr.GetMemberAssignment()
+			memberDescription, err := fromSaramaGroupMemberDescription(memberDescr)
 			if err != nil {
-				members[memberID] = MemberDescription{
-					ClientID:   memberDescr.ClientId,
-					ClientHost: memberDescr.ClientHost,
-					Err:        err,
-				}
 				continue
 			}
-
-			members[memberID] = MemberDescription{
-				ClientID:   memberDescr.ClientId,
-				ClientHost: memberDescr.ClientHost,
-				Topics:     assignment.Topics,
-			}
+			members[memberID] = memberDescription
 		}
 		groups[descr.GroupId] = GroupDescription{Members: members}
 	}
@@ -538,6 +528,31 @@ func (m *brokerFinder) lookupHosts(ips []net.IP) []string {
 	return hosts
 }
 
+func fromSaramaGroupMemberDescription(memberDescr *sarama.GroupMemberDescription) (MemberDescription, error) {
+	if memberDescr == nil {
+		return MemberDescription{}, errors.New("nil GroupMemberDescription")
+	}
+
+	assignment, err := memberDescr.GetMemberAssignment()
+	if err != nil {
+		return MemberDescription{ //nolint:nilerr // in this case we should return no error and the error is reported in MemberDescription
+			ClientID:   memberDescr.ClientId,
+			ClientHost: memberDescr.ClientHost,
+			Err:        err,
+		}, nil
+	}
+
+	assignmentTopics := make(map[string][]int32)
+	if assignment != nil {
+		assignmentTopics = assignment.Topics
+	}
+	return MemberDescription{
+		ClientID:   memberDescr.ClientId,
+		ClientHost: memberDescr.ClientHost,
+		Topics:     assignmentTopics,
+	}, nil
+}
+
 func anyIPsMatch(as, bs []net.IP) bool {
 	for _, a := range as {
 		for _, b := range bs {

metricbeat/module/kafka/broker_test.go

@@ -18,12 +18,14 @@
 package kafka
 
 import (
+	"errors"
 	"net"
 	"testing"
 
-	"errors"
-
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/sarama"
 )
 
 type dummyNet struct{}
@@ -158,3 +160,84 @@ func TestFindMatchingAddress(t *testing.T) {
 		})
 	}
 }
+
+func Test_getMember(t *testing.T) {
+	tests := []struct {
+		name             string
+		inputMemberDescr *sarama.GroupMemberDescription
+
+		expectedErrMsg string
+		expectedResult MemberDescription
+	}{
+		{
+			name: "success",
+			inputMemberDescr: &sarama.GroupMemberDescription{
+				ClientId:         "test-client",
+				ClientHost:       "test-host",
+				MemberAssignment: []byte{0, 0, 0, 0, 0, 1, 0, 10, 116, 101, 115, 116, 45, 116, 111, 112, 105, 99, 0, 0, 0, 1, 0, 0, 0, 0, 255, 255, 255, 255},
+			},
+
+			expectedErrMsg: "",
+			expectedResult: MemberDescription{
+				Err:        nil,
+				ClientID:   "test-client",
+				ClientHost: "test-host",
+				Topics: map[string][]int32{
+					"test-topic": {0},
+				},
+			},
+		},
+		{
+			name:             "nil sarama GroupMemberDescription",
+			inputMemberDescr: nil,
+
+			expectedErrMsg: "nil GroupMemberDescription",
+			expectedResult: MemberDescription{},
+		},
+		{
+			name: "0 members in the group",
+			inputMemberDescr: &sarama.GroupMemberDescription{
+				ClientId:         "test-client",
+				ClientHost:       "test-host",
+				MemberAssignment: nil,
+			},
+
+			expectedErrMsg: "",
+			expectedResult: MemberDescription{
+				Err:        nil,
+				ClientID:   "test-client",
+				ClientHost: "test-host",
+				Topics:     map[string][]int32{},
+			},
+		},
+		{
+			name: "ignore sarama error",
+			inputMemberDescr: &sarama.GroupMemberDescription{
+				ClientId:         "test-client",
+				ClientHost:       "test-host",
+				MemberAssignment: []byte{1, 2, 3},
+			},
+
+			expectedErrMsg: "",
+			expectedResult: MemberDescription{
+				Err:        errors.New("kafka: insufficient data to decode packet, more bytes expected"),
+				ClientID:   "test-client",
+				ClientHost: "test-host",
+				Topics:     nil,
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := fromSaramaGroupMemberDescription(tt.inputMemberDescr)
+			if tt.expectedErrMsg == "" {
+				require.NoError(t, err)
+
+				assert.Equal(t, tt.expectedResult, result)
+			} else {
+				assert.Error(t, err, tt.expectedErrMsg)
+			}
+		})
+	}
+}