Addressing review comments

icefury71 · icefury71 · commit 19008bece027 · 2022-06-22T22:00:06.000-07:00
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarMessageBatch.java b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarMessageBatch.java
@@ -35,6 +35,10 @@
 
 /**
  * A {@link MessageBatch} for collecting messages from pulsar topic
+ *
+ * When 'enableKeyValueStitch' flag is enabled, existing {@link org.apache.pinot.spi.stream.StreamMessageDecoder}
+ * plugins will not work. A custom decoder will be needed to unpack key and value byte arrays and decode
+ * them independently.
  */
 public class PulsarMessageBatch implements MessageBatch<byte[]> {
   private static final Logger LOGGER = LoggerFactory.getLogger(PulsarMessageBatch.class);
@@ -58,7 +62,7 @@ public byte[] getMessageAtIndex(int index) {
     if (_enableKeyValueStitch) {
       return stitchKeyValue(msg.getKeyBytes(), msg.getData());
     }
-    return _messageList.get(index).getData();
+    return msg.getData();
   }
 
   @Override
