Error decoding v5 CreateTopics message

[chris-giblin]

Versions

Sarama	Kafka	Go
v1.27.2	2.12-2.4.1	go1.14.2 linux/amd64

Configuration

Test program to decode a specific version 5 CreateTopics message obtained "off the wire" from within a proxy.

Logs

Test program output:

--- FAIL: TestBasicCreateTopics (7.30s)
    topicmgr_test.go:27: 
        	Error Trace:	topicmgr_test.go:27
        	Error:      	Received unexpected error:
        	            	kafka: insufficient data to decode packet, more bytes expected
        	Test:       	TestBasicCreateTopics
        	Messages:   	Error decoding CreateTopics message

Problem Description

Receive error when decoding a valid version 5 CreateTopics message.

Specifically, sarama calls binary.BigEndian.Uint32 at a point where, I believe, a variable length is specified by the Kafka protocol. Calling Uint32 assumes a fixed length of 4 bytes. Naturally the returned length is very wrong (in this case a very large number, far larger than the message's length), leading to the above error message.

Excerpts from a test program which decodes the CreateTopics request are given below, interspersed with comments.

To produce the message, a request to create a topic was issued using the Kafka command console:

$KAFKA_HOME/bin/kafka-topics.sh --create --topic junk-1550  --bootstrap-server $BROKER_LIST --command-config command.properties"

I have captured the resulting protocol message and hard-coded it the test program:

var (
	createTopicsMsg = []byte{
		0x00, 0x00, 0x00, 0x32, // msg len
		0x00, 0x13, // apikey
		0x00, 0x05, // version
		0x00, 0x00, 0x00, 0x05, // correlation id
		0x00, 0x0d, 0x61, 0x64, 0x6d, 0x69, 0x6e, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x2d, 0x31, // client id
		0x00, 0x02, 0x0a, 0x6a, 0x75, 0x6e, 0x6b, 0x2d, 0x31, 0x35, 0x35, 0x30, // topic name
		0xff, 0xff, 0xff, 0xff, // num partitions
		0xff, 0xff, // replication factor
		0x01, 0x01, 0x00, 0x00, 0x01, 0xd0, 0xf0, 0x00, 0x00, // partitionid/broker? validate
	}
)

I then pass the message to sarama.decodeRequest() :

	req, _, err := decodeRequest(bytes.NewReader(createTopicsMsg))
	if err != nil {
		return nil, err
	}

This returns an error. The error occurs while create_topics_request.go 's decode() function calls pd.getArrayLength():

func (c *CreateTopicsRequest) decode(pd packetDecoder, version int16) (err error) {
	n, err := pd.getArrayLength()
	if err != nil {
		return err
	}

getArrayLength() calls binary.BigEndian.Uint32(). At this point, pd.offset is 23, pointing to the first byte of the row in createTopicsMsg labelled with ` // topic name``, above:

		0x00, 0x02, 0x0a, 0x6a, 0x75, 0x6e, 0x6b, 0x2d, 0x31, 0x35, 0x35, 0x30, // topic name

By my reading, there are three length bytes (0x00, 0x02, 0x0a) in this field, yet binary.BigEndian.Uint32 will process four, additionally picking up the first byte of the topic name (0x6a). The returned length is wildly incorrect and the getArrayLength() returns an error after satisfying this condition:

	if tmp > rd.remaining() {
		rd.off = len(rd.raw)
		return -1, ErrInsufficientData
	}

Version 5 of CreateTopics uses a COMPACT_STRING for the topic name. COMPACT_STRING uses variable length bytes (var_int). If I am right, then create_topics_request.go 's decode() function must consider the message version (5) and call a different length function.

Same problem for DeleteTopics message as well.

Workaround: I switched my Kafka client to version kafka_2.11-2.3.1

https://kafka.apache.org/protocol#The_Messages_CreateTopics

[github-actions[bot]]

Thank you for taking the time to raise this issue. However, it has not had any activity on it in the past 90 days and will be closed in 30 days if no updates occur. Please check if the main branch has already resolved the issue since it was raised. If you believe the issue is still valid and you would like input from the maintainers then please comment to ask for it to be reviewed.

[dnwe]

Yes, currently we only have a small number of flexible protocol versions supported within Sarama. We are actively working on that though

[github-actions[bot]]

Thank you for taking the time to raise this issue. However, it has not had any activity on it in the past 90 days and will be closed in 30 days if no updates occur. Please check if the main branch has already resolved the issue since it was raised. If you believe the issue is still valid and you would like input from the maintainers then please comment to ask for it to be reviewed.