Repository: Shopify/sarama
Pull request #1574 (2020-01-15 15:50:17)

Enables zstd (for real this time)
Zstd support was initially added to sarama before
https://issues.apache.org/jira/browse/KAFKA-4514
https://cwiki.apache.org/confluence/display/KAFKA/KIP-110%3A+Add+Codec+for+ZStandard+Compression
was done.

The final release added some changes like bumping the produce and fetch
requests to only allow new clients to use zstd.

This PR tries to do that, however, there are some other protocol changes
that are not addressed on this PR, and I'm not sure what would be the
effect of bumping the produce and fetch requests without filling the
protocol gaps.

Tries to solve https://github.com/Shopify/sarama/issues/1252

TODO:
- [x] Test consumers via https://github.com/Shopify/sarama/pull/1582

## Update 2020-01-15 13:00 ET

- it seems it's working fine after my latests commits 🎉 

Now I need to figure it out what to do with the new producer response field, or just leave it like that?


## Update 2020-01-15 11:00 ET
- It still doesn't work :(~

It's however and improvement from the current state, this is the new error message:
```
producer/broker/1 state change to [closing] because kafka: error decoding packet: invalid length
```

current master result: (from https://github.com/Shopify/sarama/issues/1252)
```
SARAMA: producer/broker/1 starting up
SARAMA: producer/broker/1 state change to [open] on sarama-test/0
SARAMA: Connected to broker at localhost:9092 (registered as #1)
SARAMA: producer/broker/1 state change to [closing] because EOF
SARAMA: Closed connection to broker localhost:9092
TEST:   sending message failed: EOF
SARAMA: Producer shutting down.
SARAMA: Closing Client
SARAMA: Closed connection to broker localhost:9092
SARAMA: producer/broker/1 shut down
```

Proposed change: (`make testzstd`)
```
╭─[10:44:05] diegoalvarez@d1egoaz-MBP/ ~/src/github.com/Shopify/sarama
( diego_zstd-support-kafka-2_1_0_0: ✔
╰─✘ make testzstd
go test -run TestSaramaZSTD
>>> kafka version 2.1.0
Initializing new client
client/metadata fetching metadata for all topics from broker localhost:9092
Connected to broker at localhost:9092 (unregistered)
>>> encoder_decoder.go on versionedDecode, realDecoder, version: 5
>>> encoder_decoder.go helper.off != len(buf), off: 163, len: 163, version: 5
client/brokers registered new broker #1 at localhost:9092
Successfully initialized new client
producer/broker/1 starting up
producer/broker/1 state change to [open] on sarama-test/0
>>> produce_set.go buildRequest() set req.Version = 7 with compression zstd
Connected to broker at localhost:9092 (registered as #1)
>>> produce_request.go requiredVersion r.Version 7
>>> record_batch.go on encode
>>> record_batch.go/encodeRecords() compressing message on encodeRecords
>>> compressing with zstd
>>> record_batch.go/encodeRecords() NO err compressing message on encodeRecords
>>> record_batch.go put raw bytes on encode
>>> record_batch.go pe.pop on encode
>>> record_batch.go NO err pe.pop on encode
>>> record_batch.go on encode
>>> record_batch.go put raw bytes on encode
>>> record_batch.go pe.pop on encode
>>> record_batch.go NO err pe.pop on encode
>>> encoder_decoder.go on versionedDecode, realDecoder, version: 7
>>> encoder_decoder.go helper.off != len(buf), off: 47, len: 55, version: 7
>>> encoder_decoder.go on versionedDecode, version 7
producer/broker/1 state change to [closing] because kafka: error decoding packet: invalid length
Closed connection to broker localhost:9092
Producer shutting down.
Closing Client
Closed connection to broker localhost:9092
producer/broker/1 input chan closed
producer/broker/1 shut down
--- FAIL: TestSaramaZSTD (0.03s)
    zstd_test.go:45: TEST:   sending message failed: kafka: error decoding packet: invalid length
FAIL
exit status 1
FAIL    github.com/Shopify/sarama       5.422s
```

Code is failing here:
```
>>> encoder_decoder.go helper.off != len(buf), off: 47, len: 55, version: 7
```
I'm not sure what I'm missing, I'm currently 👀 the java source code.

Any help would be appreciated.

d1egoaz: ping @bai @bobrik 🙏 
d1egoaz: a little bit of more info, I think the message got produced, the error is on handling the produce response.
I'm using kafkacat to consume from the topic, and I'm seeing messages being added to the topic.

I'm also running kafkacat with debug mode, and I _think_ it's consuming messages from a topic with zstd compression according to:

```
%7|1579106515.545|FETCH|rdkafka#consumer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/1: Topic sarama-test [0] MessageSet size 3441, error "Success", MaxOffset 37, LSO 37, Ver 2/2
%7|1579106515.545|CONSUME|rdkafka#consumer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/1: Enqueue 37 message(s) (444 bytes, 37 ops) on sarama-test [0] fetch queue (qlen 37, v2, last_offset 36, 0 ctrl msgs, zstd)
%7|1579106515.545|FETCH|rdkafka#consumer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/1: Fetch topic sarama-test [0] at offset 37 (v2)
%7|1579106515.545|FETCH|rdkafka#consumer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/1: Fetch 1/1/1 toppar(s)
%7|1579106515.545|SEND|rdkafka#consumer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/1: Sent FetchRequest (v4, 75 bytes @ 0, CorrId 5)
hello world!
hello world!
hello world!
hello world!

```
d1egoaz: Error is being fired by:

https://github.com/Shopify/sarama/blob/diego_zstd-support-kafka-2_1_0_0/broker.go#L735-L751
<img src="https://screenshot.click/2020-01-15_11-56-32_rm96p.png" width="50%" />

Wich confirms my data that the message is actually being sent.
d1egoaz: It looks like we haven't read all bytes from the response, according to the error message and https://github.com/Shopify/sarama/blob/diego_zstd-support-kafka-2_1_0_0/real_decoder.go
`helper.off != len(buf), off: 47, len: 55`
So, it's expecting that we should read another `8 bytes`, wondering if that's coming from a protocol change (the gaps I mentioned above)
d1egoaz: found the missing bytes! 🎉 
<img src="https://screenshot.click/2020-01-15_12-41-15_enzd8.png" width="50%" />

Will send a commit soon, as I need to add some tests
d1egoaz: I noticed you were interested on this @lizthegrey might want help me to test it and report back, thanks
lizthegrey: > I noticed you were interested on this @lizthegrey might want help me to test it and report back, thanks

Oh thank fuck. Yes, we'd be delighted to test this. Let me figure out how to feature flag this... cc @ianwilkes @tredman 
d1egoaz: it looks like consumer is still not working: `EOF`
```
2020/01/15 15:52:51 error: &sarama.ConsumerError{Topic:"my-zstd-topic", Partition:0, Err:(*errors.errorString)(0xc00006e040)}
2020/01/15 15:52:51 error.err: &errors.errorString{s:"EOF"}
```

might be similar to the producer part, protocol changes between latest sarama protocol support and `request.Version = 10`
d1egoaz: not looking good for the consumer side :(
We need to investigate what of these fields needs to be added to the fetch request/response
https://github.com/apache/kafka/blob/99a4068c5ca61951d70b9e647ead3b08a2af4309/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java#L120-L197
d1egoaz: Wondering if we should merge this as it's right now and leave a note like:
![image](https://user-images.githubusercontent.com/950087/72531402-61047a80-383f-11ea-9854-503a997e1650.png)

looks like the consumer part has more protocol gaps that needs to be fixed, I can try to work on that on the next weeks in a new PR.

What do you think @bai , we also need to wait a little bit for more confirmation that producing works.

I'm still not sure how ruby-kafka added zstd support with only this minor changes https://github.com/zendesk/ruby-kafka/pull/724/files

I'd like to test it to see if that even works 🤷‍♂ 
dnwe: @d1egoaz can you add this patch to your branch (`git am -i file.patch`) and see if it fixes your consumer issues?

```patch
From 0993ba789864a005b56d9d828bb60771370d87d1 Mon Sep 17 00:00:00 2001
From: Dominic Evans <dominic.evans@uk.ibm.com>
Date: Thu, 16 Jan 2020 23:23:36 +0000
Subject: [PATCH] fix: fill in the FetchRequest protocol gaps

In order to consume zstd-compressed records the consumer needs to send
version 10 FetchRequests, but they need to do so in a well-formed manner

Signed-off-by: Dominic Evans <dominic.evans@uk.ibm.com>
---
 fetch_request.go      | 78 +++++++++++++++++++++++++++++++++++++++++--
 fetch_request_test.go | 44 +++++++++++++++---------
 request.go            |  2 +-
 3 files changed, 104 insertions(+), 20 deletions(-)

diff --git a/fetch_request.go b/fetch_request.go
index f2a4643..2b83d06 100644
--- a/fetch_request.go
+++ b/fetch_request.go
@@ -1,20 +1,39 @@
 package sarama
 
 type fetchRequestBlock struct {
-	fetchOffset int64
-	maxBytes    int32
+	Version            int16
+	currentLeaderEpoch int32
+	fetchOffset        int64
+	logStartOffset     int64
+	maxBytes           int32
 }
 
 func (b *fetchRequestBlock) encode(pe packetEncoder) error {
+	if b.Version >= 9 {
+		pe.putInt32(b.currentLeaderEpoch)
+	}
 	pe.putInt64(b.fetchOffset)
+	if b.Version >= 5 {
+		pe.putInt64(b.logStartOffset)
+	}
 	pe.putInt32(b.maxBytes)
 	return nil
 }
 
 func (b *fetchRequestBlock) decode(pd packetDecoder) (err error) {
+	if b.Version >= 9 {
+		if b.currentLeaderEpoch, err = pd.getInt32(); err != nil {
+			return err
+		}
+	}
 	if b.fetchOffset, err = pd.getInt64(); err != nil {
 		return err
 	}
+	if b.Version >= 5 {
+		if b.logStartOffset, err = pd.getInt64(); err != nil {
+			return err
+		}
+	}
 	if b.maxBytes, err = pd.getInt32(); err != nil {
 		return err
 	}
@@ -31,6 +50,7 @@ type FetchRequest struct {
 	Version     int16
 	Isolation   IsolationLevel
 	blocks      map[string]map[int32]*fetchRequestBlock
+	forgotten   map[string][]int32
 }
 
 type IsolationLevel int8
@@ -71,6 +91,26 @@ func (r *FetchRequest) encode(pe packetEncoder) (err error) {
 			}
 		}
 	}
+	if r.Version >= 7 {
+		err = pe.putArrayLength(len(r.forgotten))
+		if err != nil {
+			return err
+		}
+		for topic, partitions := range r.forgotten {
+			err = pe.putString(topic)
+			if err != nil {
+				return err
+			}
+			err = pe.putArrayLength(len(partitions))
+			if err != nil {
+				return err
+			}
+			for _, partition := range partitions {
+				pe.putInt32(partition)
+			}
+		}
+	}
+
 	return nil
 }
 
@@ -120,13 +160,44 @@ func (r *FetchRequest) decode(pd packetDecoder, version int16) (err error) {
 			if err != nil {
 				return err
 			}
-			fetchBlock := &fetchRequestBlock{}
+			fetchBlock := &fetchRequestBlock{Version: r.Version}
 			if err = fetchBlock.decode(pd); err != nil {
 				return err
 			}
 			r.blocks[topic][partition] = fetchBlock
 		}
 	}
+
+	if r.Version >= 7 {
+		forgottenCount, err := pd.getArrayLength()
+		if err != nil {
+			return err
+		}
+		if forgottenCount == 0 {
+			return nil
+		}
+		r.forgotten = make(map[string][]int32)
+		for i := 0; i < forgottenCount; i++ {
+			topic, err := pd.getString()
+			if err != nil {
+				return err
+			}
+			partitionCount, err := pd.getArrayLength()
+			if err != nil {
+				return err
+			}
+			r.forgotten[topic] = make([]int32, partitionCount)
+
+			for j := 0; j < partitionCount; j++ {
+				partition, err := pd.getInt32()
+				if err != nil {
+					return err
+				}
+				r.forgotten[topic][j] = partition
+			}
+		}
+	}
+
 	return nil
 }
 
@@ -165,6 +236,7 @@ func (r *FetchRequest) AddBlock(topic string, partitionID int32, fetchOffset int
 	}
 
 	tmp := new(fetchRequestBlock)
+	tmp.Version = r.Version
 	tmp.maxBytes = maxBytes
 	tmp.fetchOffset = fetchOffset
 
diff --git a/fetch_request_test.go b/fetch_request_test.go
index 1a94c2d..2fdd905 100644
--- a/fetch_request_test.go
+++ b/fetch_request_test.go
@@ -29,20 +29,32 @@ var (
 )
 
 func TestFetchRequest(t *testing.T) {
-	request := new(FetchRequest)
-	testRequest(t, "no blocks", request, fetchRequestNoBlocks)
-
-	request.MaxWaitTime = 0x20
-	request.MinBytes = 0xEF
-	testRequest(t, "with properties", request, fetchRequestWithProperties)
-
-	request.MaxWaitTime = 0
-	request.MinBytes = 0
-	request.AddBlock("topic", 0x12, 0x34, 0x56)
-	testRequest(t, "one block", request, fetchRequestOneBlock)
-
-	request.Version = 4
-	request.MaxBytes = 0xFF
-	request.Isolation = ReadCommitted
-	testRequest(t, "one block v4", request, fetchRequestOneBlockV4)
+	t.Run("no blocks", func(t *testing.T) {
+		request := new(FetchRequest)
+		testRequest(t, "no blocks", request, fetchRequestNoBlocks)
+	})
+
+	t.Run("with properties", func(t *testing.T) {
+		request := new(FetchRequest)
+		request.MaxWaitTime = 0x20
+		request.MinBytes = 0xEF
+		testRequest(t, "with properties", request, fetchRequestWithProperties)
+	})
+
+	t.Run("one block", func(t *testing.T) {
+		request := new(FetchRequest)
+		request.MaxWaitTime = 0
+		request.MinBytes = 0
+		request.AddBlock("topic", 0x12, 0x34, 0x56)
+		testRequest(t, "one block", request, fetchRequestOneBlock)
+	})
+
+	t.Run("one block v4", func(t *testing.T) {
+		request := new(FetchRequest)
+		request.Version = 4
+		request.MaxBytes = 0xFF
+		request.Isolation = ReadCommitted
+		request.AddBlock("topic", 0x12, 0x34, 0x56)
+		testRequest(t, "one block v4", request, fetchRequestOneBlockV4)
+	})
 }
diff --git a/request.go b/request.go
index 97437d6..6e4ad87 100644
--- a/request.go
+++ b/request.go
@@ -105,7 +105,7 @@ func allocateBody(key, version int16) protocolBody {
 	case 0:
 		return &ProduceRequest{}
 	case 1:
-		return &FetchRequest{}
+		return &FetchRequest{Version: version}
 	case 2:
 		return &OffsetRequest{Version: version}
 	case 3:
-- 
2.24.0
```
dnwe: (obviously this patch is only the FetchRequests — we might need the FetchResponse filled in properly as-well)
dnwe: I've pushed the above patch up under a separate PR here https://github.com/Shopify/sarama/pull/1582
d1egoaz: Great @dnwe I'll try to give it a go, I'm currently checking the producer as I've got some other `error decoding packet: invalid length` messages in some local tests, going to see what other protocol gaps are also needed for the producer
dnwe: @d1egoaz hmm I checked the ProduceRequest and it seemed there weren't any new fields required. V4 added a new error type, V5 and V6 were identical to V4 and V7 just meant "supports zstd"
d1egoaz: @dnwe see my latest commit, I added the new field in the wrong section 🤦‍♂ 
Now it's working fine! 🎉 
https://github.com/Shopify/sarama/pull/1574/commits/7e314f9cbc412d757e5a00464fd01548548a200c
d1egoaz: @lizthegrey  my tests found and error that was fixed, on my local env and some extended tests this has been working great!

@dnwe also added support for the consumer and it worked great for me and for CI.


Please let us know if you/your team were able to test this
thanks
d1egoaz: @dnwe do you want me to merge this with your commit? or would you prefer to merge that part in https://github.com/Shopify/sarama/pull/1582

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 844f481..5017e58 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,13 @@
 # Changelog
 
+
+#### Unreleased
+
+Improvements:
+- Enable zstd compression
+  ([1574](https://github.com/Shopify/sarama/pull/1574),
+  [1582](https://github.com/Shopify/sarama/pull/1582))
+
 #### Version 1.25.0 (2020-01-13)
 
 New Features:
diff --git a/config.go b/config.go
index 69c7161..e0e02f2 100644
--- a/config.go
+++ b/config.go
@@ -629,6 +629,10 @@ func (c *Config) Validate() error {
 		}
 	}
 
+	if c.Producer.Compression == CompressionZSTD && !c.Version.IsAtLeast(V2_1_0_0) {
+		return ConfigurationError("zstd compression requires Version >= V2_1_0_0")
+	}
+
 	if c.Producer.Idempotent {
 		if !c.Version.IsAtLeast(V0_11_0_0) {
 			return ConfigurationError("Idempotent producer requires Version >= V0_11_0_0")
diff --git a/consumer.go b/consumer.go
index 72c4d7c..ff44ade 100644
--- a/consumer.go
+++ b/consumer.go
@@ -888,6 +888,10 @@ func (bc *brokerConsumer) fetchNewMessages() (*FetchResponse, error) {
 		request.Isolation = bc.consumer.conf.Consumer.IsolationLevel
 	}
 
+	if bc.consumer.conf.Version.IsAtLeast(V2_1_0_0) {
+		request.Version = 10
+	}
+
 	for child := range bc.subscriptions {
 		request.AddBlock(child.topic, child.partition, child.offset, child.fetchSize)
 	}
diff --git a/produce_request.go b/produce_request.go
index 0c755d0..178972a 100644
--- a/produce_request.go
+++ b/produce_request.go
@@ -214,6 +214,8 @@ func (r *ProduceRequest) requiredVersion() KafkaVersion {
 		return V0_10_0_0
 	case 3:
 		return V0_11_0_0
+	case 7:
+		return V2_1_0_0
 	default:
 		return MinVersion
 	}
diff --git a/produce_response.go b/produce_response.go
index 4c5cd35..e4f19a7 100644
--- a/produce_response.go
+++ b/produce_response.go
@@ -5,11 +5,27 @@ import (
 	"time"
 )
 
+// Protocol, http://kafka.apache.org/protocol.html
+// v1
+// v2 = v3 = v4
+// v5 = v6 = v7
+// Produce Response (Version: 7) => [responses] throttle_time_ms
+//   responses => topic [partition_responses]
+//     topic => STRING
+//     partition_responses => partition error_code base_offset log_append_time log_start_offset
+//       partition => INT32
+//       error_code => INT16
+//       base_offset => INT64
+//       log_append_time => INT64
+//       log_start_offset => INT64
+//   throttle_time_ms => INT32
+
+// partition_responses in protocol
 type ProduceResponseBlock struct {
-	Err    KError
-	Offset int64
-	// only provided if Version >= 2 and the broker is configured with `LogAppendTime`
-	Timestamp time.Time
+	Err         KError    // v0, error_code
+	Offset      int64     // v0, base_offset
+	Timestamp   time.Time // v2, log_append_time, and the broker is configured with `LogAppendTime`
+	StartOffset int64     // v5, log_start_offset
 }
 
 func (b *ProduceResponseBlock) decode(pd packetDecoder, version int16) (err error) {
@@ -32,6 +48,13 @@ func (b *ProduceResponseBlock) decode(pd packetDecoder, version int16) (err erro
 		}
 	}
 
+	if version >= 5 {
+		b.StartOffset, err = pd.getInt64()
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -49,13 +72,17 @@ func (b *ProduceResponseBlock) encode(pe packetEncoder, version int16) (err erro
 		pe.putInt64(timestamp)
 	}
 
+	if version >= 5 {
+		pe.putInt64(b.StartOffset)
+	}
+
 	return nil
 }
 
 type ProduceResponse struct {
-	Blocks       map[string]map[int32]*ProduceResponseBlock
+	Blocks       map[string]map[int32]*ProduceResponseBlock // v0, responses
 	Version      int16
-	ThrottleTime time.Duration // only provided if Version >= 1
+	ThrottleTime time.Duration // v1, throttle_time_ms
 }
 
 func (r *ProduceResponse) decode(pd packetDecoder, version int16) (err error) {
@@ -129,6 +156,7 @@ func (r *ProduceResponse) encode(pe packetEncoder) error {
 			}
 		}
 	}
+
 	if r.Version >= 1 {
 		pe.putInt32(int32(r.ThrottleTime / time.Millisecond))
 	}
@@ -143,19 +171,6 @@ func (r *ProduceResponse) version() int16 {
 	return r.Version
 }
 
-func (r *ProduceResponse) requiredVersion() KafkaVersion {
-	switch r.Version {
-	case 1:
-		return V0_9_0_0
-	case 2:
-		return V0_10_0_0
-	case 3:
-		return V0_11_0_0
-	default:
-		return MinVersion
-	}
-}
-
 func (r *ProduceResponse) GetBlock(topic string, partition int32) *ProduceResponseBlock {
 	if r.Blocks == nil {
 		return nil
diff --git a/produce_set.go b/produce_set.go
index b684aa4..36c43c6 100644
--- a/produce_set.go
+++ b/produce_set.go
@@ -129,6 +129,10 @@ func (ps *produceSet) buildRequest() *ProduceRequest {
 		req.Version = 3
 	}
 
+	if ps.parent.conf.Producer.Compression == CompressionZSTD && ps.parent.conf.Version.IsAtLeast(V2_1_0_0) {
+		req.Version = 7
+	}
+
 	for topic, partitionSets := range ps.msgs {
 		for partition, set := range partitionSets {
 			if req.Version >= 3 {


Test patch:
diff --git a/config_test.go b/config_test.go
index 3f7a7a7..df41d5d 100644
--- a/config_test.go
+++ b/config_test.go
@@ -405,6 +405,18 @@ func TestLZ4ConfigValidation(t *testing.T) {
 	}
 }
 
+func TestZstdConfigValidation(t *testing.T) {
+	config := NewConfig()
+	config.Producer.Compression = CompressionZSTD
+	if err := config.Validate(); string(err.(ConfigurationError)) != "zstd compression requires Version >= V2_1_0_0" {
+		t.Error("Expected invalid zstd/kafka version error, got ", err)
+	}
+	config.Version = V2_1_0_0
+	if err := config.Validate(); err != nil {
+		t.Error("Expected zstd to work, got ", err)
+	}
+}
+
 // This example shows how to integrate with an existing registry as well as publishing metrics
 // on the standard output
 func ExampleConfig_metrics() {
diff --git a/functional_consumer_test.go b/functional_consumer_test.go
index 135ee79..8b31b45 100644
--- a/functional_consumer_test.go
+++ b/functional_consumer_test.go
@@ -106,6 +106,21 @@ func TestVersionMatrixLZ4(t *testing.T) {
 	consumeMsgs(t, testVersions, producedMessages)
 }
 
+// Support for zstd codec was introduced in v2.1.0.0
+func TestVersionMatrixZstd(t *testing.T) {
+	setupFunctionalTest(t)
+	defer teardownFunctionalTest(t)
+
+	// Produce lot's of message with all possible combinations of supported
+	// protocol versions starting with v2.1.0.0 (first where zstd was supported)
+	testVersions := versionRange(V2_1_0_0)
+	allCodecs := []CompressionCodec{CompressionZSTD}
+	producedMessages := produceMsgs(t, testVersions, allCodecs, 17, 100, false)
+
+	// When/Then
+	consumeMsgs(t, testVersions, producedMessages)
+}
+
 func TestVersionMatrixIdempotent(t *testing.T) {
 	setupFunctionalTest(t)
 	defer teardownFunctionalTest(t)
diff --git a/produce_response_test.go b/produce_response_test.go
index 197c7fb..0dfac79 100644
--- a/produce_response_test.go
+++ b/produce_response_test.go
@@ -10,8 +10,8 @@ var (
 	produceResponseNoBlocksV0 = []byte{
 		0x00, 0x00, 0x00, 0x00}
 
-	produceResponseManyBlocksVersions = [][]byte{
-		{
+	produceResponseManyBlocksVersions = map[int][]byte{
+		0: {
 			0x00, 0x00, 0x00, 0x01,
 
 			0x00, 0x03, 'f', 'o', 'o',
@@ -20,7 +20,9 @@ var (
 			0x00, 0x00, 0x00, 0x01, // Partition 1
 			0x00, 0x02, // ErrInvalidMessage
 			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xFF, // Offset 255
-		}, {
+		},
+
+		1: {
 			0x00, 0x00, 0x00, 0x01,
 
 			0x00, 0x03, 'f', 'o', 'o',
@@ -31,7 +33,8 @@ var (
 			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xFF, // Offset 255
 
 			0x00, 0x00, 0x00, 0x64, // 100 ms throttle time
-		}, {
+		},
+		2: {
 			0x00, 0x00, 0x00, 0x01,
 
 			0x00, 0x03, 'f', 'o', 'o',
@@ -42,6 +45,20 @@ var (
 			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xFF, // Offset 255
 			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, 0xE8, // Timestamp January 1st 0001 at 00:00:01,000 UTC (LogAppendTime was used)
 
+			0x00, 0x00, 0x00, 0x64, // 100 ms throttle time
+		},
+		7: { // version 7 adds StartOffset
+			0x00, 0x00, 0x00, 0x01,
+
+			0x00, 0x03, 'f', 'o', 'o',
+			0x00, 0x00, 0x00, 0x01,
+
+			0x00, 0x00, 0x00, 0x01, // Partition 1
+			0x00, 0x02, // ErrInvalidMessage
+			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xFF, // Offset 255
+			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, 0xE8, // Timestamp January 1st 0001 at 00:00:01,000 UTC (LogAppendTime was used)
+			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x32, // StartOffset 50
+
 			0x00, 0x00, 0x00, 0x64, // 100 ms throttle time
 		},
 	}
@@ -69,14 +86,19 @@ func TestProduceResponseDecode(t *testing.T) {
 			t.Error("Decoding did not produce a block for foo/1")
 		} else {
 			if block.Err != ErrInvalidMessage {
-				t.Error("Decoding failed for foo/2/Err, got:", int16(block.Err))
+				t.Error("Decoding failed for foo/1/Err, got:", int16(block.Err))
 			}
 			if block.Offset != 255 {
 				t.Error("Decoding failed for foo/1/Offset, got:", block.Offset)
 			}
 			if v >= 2 {
 				if block.Timestamp != time.Unix(1, 0) {
-					t.Error("Decoding failed for foo/2/Timestamp, got:", block.Timestamp)
+					t.Error("Decoding failed for foo/1/Timestamp, got:", block.Timestamp)
+				}
+			}
+			if v >= 7 {
+				if block.StartOffset != 50 {
+					t.Error("Decoding failed for foo/1/StartOffset, got:", block.StartOffset)
 				}
 			}
 		}
@@ -95,9 +117,10 @@ func TestProduceResponseEncode(t *testing.T) {
 
 	response.Blocks["foo"] = make(map[int32]*ProduceResponseBlock)
 	response.Blocks["foo"][1] = &ProduceResponseBlock{
-		Err:       ErrInvalidMessage,
-		Offset:    255,
-		Timestamp: time.Unix(1, 0),
+		Err:         ErrInvalidMessage,
+		Offset:      255,
+		Timestamp:   time.Unix(1, 0),
+		StartOffset: 50,
 	}
 	response.ThrottleTime = 100 * time.Millisecond
 	for v, produceResponseManyBlocks := range produceResponseManyBlocksVersions {