Fix compression timestamps

produce_set.go

@@ -90,11 +90,29 @@ func (ps *produceSet) buildRequest() *ProduceRequest {
 					Logger.Println(err) // if this happens, it's basically our fault.
 					panic(err)
 				}
-				req.AddMessage(topic, partition, &Message{
+				compMsg := &Message{
 					Codec: ps.parent.conf.Producer.Compression,
 					Key:   nil,
 					Value: payload,
-				})
+				}
+				if ps.parent.conf.Version.IsAtLeast(V0_10_0_0) {
+					// Compressed messages must use a protocol version
+					// that is newer than the inner messages version.
+					// Due to a lack of better timestamp notation copy the oldest
+					// (earliest) timestamp to message.
+					for _, msgBlock := range set.setToSend.Messages {
+						msg := msgBlock.Msg
+						if msg.Version > compMsg.Version {
+							compMsg.Version = msg.Version
+						}
+						if !msg.Timestamp.IsZero() &&
+							(compMsg.Timestamp.IsZero() ||
+								compMsg.Timestamp.After(msg.Timestamp)) {
+							compMsg.Timestamp = msg.Timestamp
+						}
+					}
+				}
+				req.AddMessage(topic, partition, compMsg)
 			}
 		}
 	}

produce_set_test.go

@@ -141,3 +141,45 @@ func TestProduceSetRequestBuilding(t *testing.T) {
 		t.Error("Wrong number of topics in request")
 	}
 }
+
+func TestProduceSetCompressedRequestBuilding(t *testing.T) {
+	parent, ps := makeProduceSet()
+	parent.conf.Producer.RequiredAcks = WaitForAll
+	parent.conf.Producer.Timeout = 10 * time.Second
+	parent.conf.Producer.Compression = CompressionGZIP
+	parent.conf.Version = V0_10_0_0
+
+	msg := &ProducerMessage{
+		Topic:     "t1",
+		Partition: 0,
+		Key:       StringEncoder(TestMessage),
+		Value:     StringEncoder(TestMessage),
+		Timestamp: time.Now(),
+	}
+	for i := 0; i < 10; i++ {
+		safeAddMessage(t, ps, msg)
+	}
+
+	req := ps.buildRequest()
+
+	if req.Version != 2 {
+		t.Error("Wrong request version")
+	}
+
+	for _, msgBlock := range req.msgSets["t1"][0].Messages {
+		msg := msgBlock.Msg
+		err := msg.decodeSet()
+		if err != nil {
+			t.Error("Failed to decode set from payload")
+		}
+		for _, compMsgBlock := range msg.Set.Messages {
+			compMsg := compMsgBlock.Msg
+			if compMsg.Version != 1 {
+				t.Error("Wrong compressed message version")
+			}
+		}
+		if msg.Version != 1 {
+			t.Error("Wrong compressed parent message version")
+		}
+	}
+}