fix: handle nullable metadata in OffsetFetchResponse

The Kafka protocol defines OffsetFetchResponse.Partitions.Metadata as
nullable (versions 0-7), but sarama decoded it with getString() which
rejects a -1 length. Use getNullableString()/putNullableString() instead,
mapping nil to empty string to preserve backwards compatibility.

Fixes #3444

Signed-off-by: Dominic Evans <dominic.evans@uk.ibm.com>

Author: dnwe

offset_fetch_response.go

@@ -24,10 +24,13 @@ func (b *OffsetFetchResponseBlock) decode(pd packetDecoder, version int16) (err
 		b.LeaderEpoch = -1
 	}
 
-	b.Metadata, err = pd.getString()
+	metadata, err := pd.getNullableString()
 	if err != nil {
 		return err
 	}
+	if metadata != nil {
+		b.Metadata = *metadata
+	}
 
 	b.Err, err = pd.getKError()
 	if err != nil {
@@ -44,7 +47,7 @@ func (b *OffsetFetchResponseBlock) encode(pe packetEncoder, version int16) (err
 	if version >= 5 {
 		pe.putInt32(b.LeaderEpoch)
 	}
-	err = pe.putString(b.Metadata)
+	err = pe.putNullableString(&b.Metadata)
 	if err != nil {
 		return err
 	}

offset_fetch_response_test.go

@@ -5,9 +5,24 @@ package sarama
 import (
 	"fmt"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 var (
+	// version 0 response with one topic "t", one partition 0, offset 0,
+	// null metadata (length -1), and error code 7 (ErrRequestTimedOut)
+	offsetFetchResponseV0NullMetadata = []byte{
+		0x00, 0x00, 0x00, 0x01, // num topics = 1
+		0x00, 0x01, 't', // topic name = "t"
+		0x00, 0x00, 0x00, 0x01, // num partitions = 1
+		0x00, 0x00, 0x00, 0x00, // partition = 0
+		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // offset = 0
+		0xFF, 0xFF, // metadata = null (length -1)
+		0x00, 0x07, // error code = 7
+	}
+
 	emptyOffsetFetchResponse = []byte{
 		0x00, 0x00, 0x00, 0x00,
 	}
@@ -39,6 +54,17 @@ func TestEmptyOffsetFetchResponse(t *testing.T) {
 	}
 }
 
+func TestOffsetFetchResponseNullMetadata(t *testing.T) {
+	response := &OffsetFetchResponse{}
+	err := versionedDecode(offsetFetchResponseV0NullMetadata, response, 0, nil)
+	require.NoError(t, err)
+
+	block := response.GetBlock("t", 0)
+	require.NotNil(t, block)
+	assert.Empty(t, block.Metadata)
+	assert.Equal(t, ErrRequestTimedOut, block.Err)
+}
+
 func TestNormalOffsetFetchResponse(t *testing.T) {
 	// The response encoded form cannot be checked for it varies due to
 	// unpredictable map traversal order.