Make topic name accessible in the MessageSplitterImpl (#102)

linkedin · Jan 2, 2019 · 54c5009 · 54c5009
1 parent 756b70f
commit 54c5009
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/...ka-clients/src/main/java/com/linkedin/kafka/clients/largemessage/MessageSplitterImpl.java b/...ka-clients/src/main/java/com/linkedin/kafka/clients/largemessage/MessageSplitterImpl.java
@@ -94,8 +94,8 @@ public List<ProducerRecord<byte[], byte[]>> split(String topic,
       LargeMessageSegment segment = new LargeMessageSegment(segmentMessageId, seq,
           numberOfSegments, messageSizeInBytes, payload);
 
-      // NOTE: we have to use null topic here to serialize because the segment should be topic independent.
-      byte[] segmentValue = _segmentSerializer.serialize(null, segment);
+      // NOTE: Even though we are passing topic here to serialize, the segment itself should be topic independent.
+      byte[] segmentValue = _segmentSerializer.serialize(topic, segment);
       ProducerRecord<byte[], byte[]> segmentProducerRecord =
           new ProducerRecord<>(topic, partition, timestamp, segmentKey, segmentValue);
       segments.add(segmentProducerRecord);