Merge pull request #1002 from emfree/assign_relative_offsets

Assign relative offsets in compressed message sets

produce_set.go

@@ -132,6 +132,17 @@ func (ps *produceSet) buildRequest() *ProduceRequest {
 				// and sent as the payload of a single fake "message" with the appropriate codec
 				// set and no key. When the server sees a message with a compression codec, it
 				// decompresses the payload and treats the result as its message set.
+
+				if ps.parent.conf.Version.IsAtLeast(V0_10_0_0) {
+					// If our version is 0.10 or later, assign relative offsets
+					// to the inner messages. This lets the broker avoid
+					// recompressing the message set.
+					// (See https://cwiki.apache.org/confluence/display/KAFKA/KIP-31+-+Move+to+relative+offsets+in+compressed+message+sets
+					// for details on relative offsets.)
+					for i, msg := range set.recordsToSend.msgSet.Messages {
+						msg.Offset = int64(i)
+					}
+				}
 				payload, err := encode(set.recordsToSend.msgSet, ps.parent.conf.MetricRegistry)
 				if err != nil {
 					Logger.Println(err) // if this happens, it's basically our fault.

produce_set_test.go

@@ -173,11 +173,14 @@ func TestProduceSetCompressedRequestBuilding(t *testing.T) {
 		if err != nil {
 			t.Error("Failed to decode set from payload")
 		}
-		for _, compMsgBlock := range msg.Set.Messages {
+		for i, compMsgBlock := range msg.Set.Messages {
 			compMsg := compMsgBlock.Msg
 			if compMsg.Version != 1 {
 				t.Error("Wrong compressed message version")
 			}
+			if compMsgBlock.Offset != int64(i) {
+				t.Error("Wrong relative inner offset")
+			}
 		}
 		if msg.Version != 1 {
 			t.Error("Wrong compressed parent message version")