Move offset validation logic to consumer classes #13015

KKcorps · 2024-04-26T12:03:47Z

Depends on - #12806

Jackie-Jiang

There is no guarantee we can always detect gap with start offset and batch first offset, especially after we introduce the exclusive start offset concept.
Seems like we still want to report gap within the general handling flow. In that case it should be easier to add an API hasGap() into MessageBatch, which can be filled by the stream consumer

KKcorps · 2024-04-29T12:54:22Z

There is no guarantee we can always detect gap with start offset and batch first offset, especially after we introduce the exclusive start offset concept. Seems like we still want to report gap within the general handling flow. In that case it should be easier to add an API hasGap() into MessageBatch, which can be filled by the stream consumer

That makes sense since we already have the information in the consumer from which offset did we actually wanted to consume the data vs from which offset we actually did OR any other criteria which consumer wants to implement

pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java

pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupConsumer.java

.../pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisMessageBatch.java

.../pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaMessageBatch.java

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

...ka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumer.java

Jackie-Jiang

LGTM otherwise

Jackie-Jiang · 2024-05-22T05:08:15Z

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

   */
-  private void validateStartOffset(StreamPartitionMsgOffset startOffset, StreamPartitionMsgOffset batchFirstOffset) {
-    if (batchFirstOffset.compareTo(startOffset) > 0) {
+  private void reportDataLoss(MessageBatch messageBatch) {


Please update the java doc for this method

Jackie-Jiang · 2024-05-22T05:09:02Z

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

      _serverMetrics.addMeteredTableValue(_tableStreamName, ServerMeter.STREAM_DATA_LOSS, 1L);
      String message =
-          "startOffset(" + startOffset + ") is older than topic's beginning offset(" + batchFirstOffset + ")";
+          "Message loss detected in stream partition: " + _partitionGroupId + " for table: " + _tableNameWithType


Let's use String.format() for readability

Jackie-Jiang · 2024-05-22T05:09:25Z

...tion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisConsumer.java

@@ -66,7 +66,7 @@ public synchronized KinesisMessageBatch fetchMessages(StreamPartitionMsgOffset s
    KinesisPartitionGroupOffset startOffset = (KinesisPartitionGroupOffset) startMsgOffset;
    String shardId = startOffset.getShardId();
    String startSequenceNumber = startOffset.getSequenceNumber();
-
+    boolean isMissingMessages = false;


Revert this

Jackie-Jiang · 2024-05-22T05:09:35Z

pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java

@@ -96,6 +96,10 @@ default boolean isEndOfPartitionGroup() {
    return false;
  }

+  default boolean hasDataLoss() {


Add some javadoc for this method

codecov-commenter · 2024-05-22T05:30:09Z

Codecov Report

Attention: Patch coverage is 50.00000% with 5 lines in your changes are missing coverage. Please review.

Project coverage is 62.20%. Comparing base (59551e4) to head (41da47c).
Report is 485 commits behind head on master.

Files	Patch %	Lines
...a/manager/realtime/RealtimeSegmentDataManager.java	20.00%	3 Missing and 1 partial ⚠️
...pinot/plugin/stream/kafka20/KafkaMessageBatch.java	66.66%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #13015      +/-   ##
============================================
+ Coverage     61.75%   62.20%   +0.45%     
+ Complexity      207      198       -9     
============================================
  Files          2436     2531      +95     
  Lines        133233   138570    +5337     
  Branches      20636    21448     +812     
============================================
+ Hits          82274    86197    +3923     
- Misses        44911    45926    +1015     
- Partials       6048     6447     +399

Flag	Coverage Δ
custom-integration1	`<0.01% <0.00%> (-0.01%)`	⬇️
integration	`<0.01% <0.00%> (-0.01%)`	⬇️
integration1	`<0.01% <0.00%> (-0.01%)`	⬇️
integration2	`0.00% <0.00%> (ø)`
java-11	`62.18% <50.00%> (+0.47%)`	⬆️
java-21	`35.11% <20.00%> (-26.52%)`	⬇️
skip-bytebuffers-false	`62.20% <50.00%> (+0.45%)`	⬆️
skip-bytebuffers-true	`<0.01% <0.00%> (-27.73%)`	⬇️
temurin	`62.20% <50.00%> (+0.45%)`	⬆️
unittests	`62.20% <50.00%> (+0.45%)`	⬆️
unittests1	`46.67% <33.33%> (-0.22%)`	⬇️
unittests2	`27.90% <30.00%> (+0.17%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

* Enhance Kinesis consumer * Simplify the handling * Address comments * Move offset validation logic to consumer classes * Add missing message interface to message batch * fix linting * remove unused interface * Cleanup and refactoring * lint fixes --------- Co-authored-by: Xiaotian (Jackie) Jiang <jackie.jxt@gmail.com> Co-authored-by: Kartik Khare <kharekartik@kartiks-macbook-pro.tail8a064.ts.net> Co-authored-by: Kartik Khare <kharekartik@Kartiks-MacBook-Pro.local>

Jackie-Jiang reviewed Apr 26, 2024

View reviewed changes

KKcorps force-pushed the validate_offset_patch branch from f51ee5b to ffe8014 Compare April 30, 2024 11:37

KKcorps marked this pull request as ready for review May 22, 2024 02:58

Jackie-Jiang and others added 5 commits May 22, 2024 09:27

Enhance Kinesis consumer

12f6555

Simplify the handling

bc137f4

Address comments

05c748b

Move offset validation logic to consumer classes

2c28b7a

Add missing message interface to message batch

7918286

KKcorps force-pushed the validate_offset_patch branch from ffe8014 to 7918286 Compare May 22, 2024 03:58

KKcorps changed the title ~~[WIP] Move offset validation logic to consumer classes"~~ Move offset validation logic to consumer classes" May 22, 2024

Jackie-Jiang changed the title ~~Move offset validation logic to consumer classes"~~ Move offset validation logic to consumer classes May 22, 2024

fix linting

236dee9

KKcorps force-pushed the validate_offset_patch branch from 553cdb5 to 236dee9 Compare May 22, 2024 04:28

Jackie-Jiang reviewed May 22, 2024

View reviewed changes

Kartik Khare added 2 commits May 22, 2024 10:18

remove unused interface

1f9362d

Cleanup and refactoring

49f73b6

Jackie-Jiang approved these changes May 22, 2024

View reviewed changes

lint fixes

41da47c

KKcorps merged commit 29c560f into apache:master May 23, 2024
19 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Move offset validation logic to consumer classes #13015

Move offset validation logic to consumer classes #13015

KKcorps commented Apr 26, 2024

Jackie-Jiang left a comment

KKcorps commented Apr 29, 2024

Jackie-Jiang left a comment

Jackie-Jiang May 22, 2024

Jackie-Jiang May 22, 2024

Jackie-Jiang May 22, 2024

Jackie-Jiang May 22, 2024

codecov-commenter commented May 22, 2024 •

edited

Loading

Move offset validation logic to consumer classes #13015

Move offset validation logic to consumer classes #13015

Conversation

KKcorps commented Apr 26, 2024

Jackie-Jiang left a comment

Choose a reason for hiding this comment

KKcorps commented Apr 29, 2024

Jackie-Jiang left a comment

Choose a reason for hiding this comment

Jackie-Jiang May 22, 2024

Choose a reason for hiding this comment

Jackie-Jiang May 22, 2024

Choose a reason for hiding this comment

Jackie-Jiang May 22, 2024

Choose a reason for hiding this comment

Jackie-Jiang May 22, 2024

Choose a reason for hiding this comment

codecov-commenter commented May 22, 2024 • edited Loading

Codecov Report

codecov-commenter commented May 22, 2024 •

edited

Loading