When compressing messages, wrap them in a message-set

producer.go

@@ -104,6 +104,15 @@ func (p *Producer) safeSendMessage(key, value Encoder, retry bool) error {
 		return err
 	}
 
+	if p.config.Compression != CompressionNone {
+		set := new(MessageSet)
+		set.addMessage(&Message{Codec: CompressionNone, Key: keyBytes, Value: valBytes})
+		valBytes, err = encode(set)
+		if err != nil {
+			return err
+		}
+	}
+
 	request := &ProduceRequest{RequiredAcks: p.config.RequiredAcks, Timeout: p.config.Timeout}
 	request.AddMessage(p.topic, partition, &Message{Codec: p.config.Compression, Key: keyBytes, Value: valBytes})