Sarama library improvement prove of concept to avoid kafka broker re-compression

[lintang0]

Versions

• Sarama Version: f7e3024
• Kafka Version: 0.10.1.0
• Go Version: 1.8

Sarama Configuration

config.Version = sarama.V0_10_1_0
config.Producer.Compression = sarama.CompressionGZIP

Kafka Configuration in server.properties

 compression.type=producer

Kafka server.log

[2017-06-28 17:34:06,452] INFO BBMS : messageAndOffset.offset:expectedInnerOffset : 0:0 (kafka.message.ByteBufferMessageSet)
[2017-06-28 17:34:06,453] INFO BBMS : messageAndOffset.offset:expectedInnerOffset : 0:1 (kafka.message.ByteBufferMessageSet)
[2017-06-28 17:34:06,453] INFO BBMS : messageAndOffset.offset:expectedInnerOffset : 0:2 (kafka.message.ByteBufferMessageSet)
[2017-06-28 17:34:06,453] INFO BBMS : messageAndOffset.offset:expectedInnerOffset : 0:3 (kafka.message.ByteBufferMessageSet)
[2017-06-28 17:34:06,454] INFO BBMS : messageAndOffset.offset:expectedInnerOffset : 0:4 (kafka.message.ByteBufferMessageSet)
[2017-06-28 17:34:06,454] INFO BBMS : Recompressing Message  (kafka.message.ByteBufferMessageSet)

Problem Description

Kafka broker re-compresses messages when receiving compressed messages due to the inner offset of zero shown in the above kafka server.log. There are discussion (https://cwiki.apache.org/confluence/display/KAFKA/KIP-31+-+Move+to+relative+offsets+in+compressed+message+sets) regarding the inner offset.

Here is a proof of concept in sarama message_set.go to demonstrate the improvement of broker without re-compression. Broker's CPU usage reduced 30% in the test environment.

diff --git a/message_set.go b/message_set.go

@@ -51,7 +51,16 @@ type MessageSet struct {
 }

 func (ms *MessageSet) encode(pe packetEncoder) error {
+
+       var correction int
+       if len(ms.Messages) == 1 && ms.Messages[0].Msg.Set != nil {
+               correction = len(ms.Messages[0].Msg.Set.Messages)
+       } else {
+               correction = len(ms.Messages)
+       }
+
    for i := range ms.Messages {
+           ms.Messages[i].Offset = int64(i + correction - 1)
            err := ms.Messages[i].encode(pe)
            if err != nil {
                    return err

With the above update, broker doesn't need to re-compress the messages since the inner offset set to the 0, 1, 2... by sarama library, leading to 30% lower cpu usage in the test setup. here is the broker log with 5 messages produced.

[2017-06-28 17:34:33,989] INFO BBMS : messageAndOffset.offset:expectedInnerOffset : 0:0 (kafka.message.ByteBufferMessageSet)
[2017-06-28 17:34:33,989] INFO BBMS : messageAndOffset.offset:expectedInnerOffset : 1:1 (kafka.message.ByteBufferMessageSet)
[2017-06-28 17:34:33,989] INFO BBMS : messageAndOffset.offset:expectedInnerOffset : 2:2 (kafka.message.ByteBufferMessageSet)
[2017-06-28 17:34:33,990] INFO BBMS : messageAndOffset.offset:expectedInnerOffset : 3:3 (kafka.message.ByteBufferMessageSet)
[2017-06-28 17:34:33,990] INFO BBMS : messageAndOffset.offset:expectedInnerOffset : 4:4 (kafka.message.ByteBufferMessageSet)
[2017-06-28 17:34:33,990] INFO BBMS : Original Message Retained :  (kafka.message.ByteBufferMessageSet)

The following patch was used to produce the debug log based on the source code https://github.com/apache/kafka/archive/0.10.1.0.tar.gz

@@ -448,14 +448,17 @@
       // Validate the timestamp
       validateTimestamp(message, now, messageTimestampType, messageTimestampDiffMaxMs)
       // Check if we need to overwrite offset
-          if (messageAndOffset.offset != expectedInnerOffset.getAndIncrement()) 
+          var innerOffsetIncrement = expectedInnerOffset.getAndIncrement() 
+          if (messageAndOffset.offset != innerOffsetIncrement)             inPlaceAssignment = false           if (message.timestamp > maxTimestamp) {             maxTimestamp = message.timestamp             offsetOfMaxTimestamp = offsetCounter.value + expectedInnerOffset.value - 1           } 
+          info("BBMS : messageAndOffset.offset:expectedInnerOffset : " + messageAndOffset.offset + ":" + innerOffsetIncrement)         } 

if (sourceCodec != NoCompressionCodec && message.compressionCodec != NoCompressionCodec)           throw new InvalidMessageException("Compressed outer message should not have an inner message with a " +             s"compression attribute set: $message") 

@@ -478,6 +481,7 @@
         (Some(now), {if (targetCodec == NoCompressionCodec) offsetCounter.value else offsetCounter.value + validatedMessages.length - 1})         } 
+        info("BBMS : Recompressing Message ")         ValidationAndOffsetAssignResult(validatedMessages = new ByteBufferMessageSet(compressionCodec = targetCodec,                                                                                      offsetCounter = offsetCounter,                                                                                      wrapperMessageTimestamp = largestTimestampOfMessageSet, 

@@ -487,6 +491,7 @@
                                     offsetOfMaxTimestamp = offsetOfMaxTimestampInMessageSet,                                         messageSizeMaybeChanged = true)       } else { 
+          info("BBMS : Original Message Retained : " )
     // Do not do re-compression but simply update the offset, timestamp and attributes field of the wrapper message.         buffer.putLong(0, offsetCounter.addAndGet(validatedMessages.size) - 1)
     // validate the messages

[eapache]

Makes sense to me, I'd be happy to consider this as a Pull Request.

My only concern would be if we need to do something to maintain backwards-compatibility for older brokers which don't know how to deal with relative inner offsets.

[eapache]

Closing as stale. Please re-open as a pull request if you are still interested.