Re: [PR] Move offset validation logic to consumer classes [pinot]

["Jackie-Jiang (via GitHub)" <gi...@apache.org>]

Jackie-Jiang commented on code in PR #13015:
URL: https://github.com/apache/pinot/pull/13015#discussion_r1609243748


##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaMessageBatch.java:
##########
@@ -43,12 +44,13 @@ public class KafkaMessageBatch implements MessageBatch<byte[]> {
    *                            delay when a batch has all messages filtered.
    */
   public KafkaMessageBatch(List<BytesStreamMessage> messages, int unfilteredMessageCount, long offsetOfNextBatch,
-      long firstOffset, @Nullable StreamMessageMetadata lastMessageMetadata) {
+      long firstOffset, @Nullable StreamMessageMetadata lastMessageMetadata, boolean isMissingMessages) {

Review Comment:
   ```suggestion
         long firstOffset, @Nullable StreamMessageMetadata lastMessageMetadata, boolean hasDataLoss) {
   ```



##########
pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisMessageBatch.java:
##########
@@ -31,12 +31,22 @@ public class KinesisMessageBatch implements MessageBatch<byte[]> {
   private final List<BytesStreamMessage> _messages;
   private final KinesisPartitionGroupOffset _offsetOfNextBatch;
   private final boolean _endOfShard;
+  private final boolean _isMissingOffsets;
 
   public KinesisMessageBatch(List<BytesStreamMessage> messages, KinesisPartitionGroupOffset offsetOfNextBatch,
       boolean endOfShard) {
     _messages = messages;
     _offsetOfNextBatch = offsetOfNextBatch;
     _endOfShard = endOfShard;
+    _isMissingOffsets = false;
+  }
+
+  public KinesisMessageBatch(List<BytesStreamMessage> messages, KinesisPartitionGroupOffset offsetOfNextBatch,

Review Comment:
   No need to add this new constructor. We can add it when data loss can be detected



##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupConsumer.java:
##########
@@ -84,4 +84,9 @@ default MessageBatch fetchMessages(StreamPartitionMsgOffset startOffset, @Nullab
   default StreamPartitionMsgOffset checkpoint(StreamPartitionMsgOffset lastOffset) {
     return lastOffset;
   }
+
+  default boolean isOffsetMismatchAcceptable(StreamPartitionMsgOffset startOffset,

Review Comment:
   Do we need this API? I don't see how is it used



##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java:
##########
@@ -96,6 +96,10 @@ default boolean isEndOfPartitionGroup() {
     return false;
   }
 
+  default boolean hasMissingOffsets() {

Review Comment:
   Shall we call it `hasDataLoss()` which IMO is more intuitive. Also please add some javadoc



##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumer.java:
##########
@@ -84,7 +86,10 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
         lastMessageMetadata = messageMetadata;
       }
     }
-    return new KafkaMessageBatch(filteredRecords, records.size(), offsetOfNextBatch, firstOffset, lastMessageMetadata);
+
+    isMissingMessages = firstOffset > startOffset;
+    return new KafkaMessageBatch(filteredRecords, records.size(), offsetOfNextBatch, firstOffset, lastMessageMetadata,
+        isMissingMessages);

Review Comment:
   (nit, more consice)
   ```suggestion
   
       isMissingMessages = firstOffset > startOffset;
       return new KafkaMessageBatch(filteredRecords, records.size(), offsetOfNextBatch, firstOffset, lastMessageMetadata,
          firstOffset > startOffset);
   ```



##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -926,14 +926,14 @@ public Map<String, PartitionLagState> getPartitionToLagState(
    * batchFirstOffset should be less than or equal to startOffset.
    * If batchFirstOffset is greater, then some messages were not received.
    *
-   * @param startOffset The offset of the first message desired, inclusive.
-   * @param batchFirstOffset The offset of the first message in the batch.
+   * @param messageBatch Message batch to validate
    */
-  private void validateStartOffset(StreamPartitionMsgOffset startOffset, StreamPartitionMsgOffset batchFirstOffset) {
-    if (batchFirstOffset.compareTo(startOffset) > 0) {
+  private void validateStartOffset(MessageBatch messageBatch) {

Review Comment:
   Suggest renaming it to `reportDataLoss`



##########
pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisMessageBatch.java:
##########
@@ -58,4 +68,9 @@ public StreamPartitionMsgOffset getOffsetOfNextBatch() {
   public boolean isEndOfPartitionGroup() {
     return _endOfShard;
   }
+
+  @Override
+  public boolean hasMissingOffsets() {
+    return MessageBatch.super.hasMissingOffsets();

Review Comment:
   Let's also add a TODO to support detecting data loss for Kinesis
   ```suggestion
       return false;
   ```



##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -470,7 +470,7 @@ protected boolean consumeLoop()
 
       StreamPartitionMsgOffset batchFirstOffset = messageBatch.getFirstMessageOffset();
       if (batchFirstOffset != null) {
-        validateStartOffset(_currentOffset, batchFirstOffset);
+        validateStartOffset(messageBatch);

Review Comment:
   No need to fetch the first message offset. We can directly check data loss from the message batch



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org