Add support to track if offset needs to honoured or not in Kinesis #13112

KKcorps · 2024-05-08T19:16:48Z

This will allow us to resume from LATEST or EARLIEST instead of always starting from a sequence number

KKcorps · 2024-05-08T19:17:01Z

codecov-commenter · 2024-05-09T03:17:31Z

Codecov Report

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

Project coverage is 46.61%. Comparing base (59551e4) to head (3198ee4).
Report is 488 commits behind head on master.

Files	Patch %	Lines
...not/common/metadata/segment/SegmentZKMetadata.java	33.33%	3 Missing and 1 partial ⚠️
...e/pinot/spi/config/table/SegmentZKPropsConfig.java	0.00%	3 Missing ⚠️
...egment/spi/index/metadata/SegmentMetadataImpl.java	33.33%	2 Missing ⚠️
...a/manager/realtime/RealtimeSegmentDataManager.java	75.00%	1 Missing ⚠️
...ment/creator/impl/SegmentColumnarIndexCreator.java	0.00%	1 Missing ⚠️
...pache/pinot/spi/stream/PartitionGroupConsumer.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #13112       +/-   ##
=============================================
- Coverage     61.75%   46.61%   -15.15%     
- Complexity      207     1103      +896     
=============================================
  Files          2436     1928      -508     
  Lines        133233   102058    -31175     
  Branches      20636    16459     -4177     
=============================================
- Hits          82274    47571    -34703     
- Misses        44911    50975     +6064     
+ Partials       6048     3512     -2536

Flag	Coverage Δ
custom-integration1	`?`
integration	`?`
integration1	`?`
integration2	`?`
java-11	`46.61% <33.33%> (-15.10%)`	⬇️
java-21	`?`
skip-bytebuffers-false	`46.61% <33.33%> (-15.14%)`	⬇️
skip-bytebuffers-true	`?`
temurin	`46.61% <33.33%> (-15.15%)`	⬇️
unittests	`46.61% <33.33%> (-15.14%)`	⬇️
unittests1	`46.61% <33.33%> (-0.28%)`	⬇️
unittests2	`?`

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.

Jackie-Jiang · 2024-05-11T06:11:22Z

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

+ */
+package org.apache.pinot.plugin.stream.kinesis;
+
+public enum OffsetStartStatus {


Basically we want to have a flag to mark the first segment of a partition. I think adding a flag in the ZK metadata is more general. Without the flag, we can treat the segment with sequence id 0 as the first segment. The reason why we still want to have a flag:

When consumption failed on segment 0, we might recreate segment 1 which is also the start of the partition

We can check if the flag exist to determine whether we are migrating from the inclusive offset to exclusive offset for existing tables

Not really getting this point. If this is persisted in ZK, it serves the same purpose right?

Jackie-Jiang · 2024-05-11T06:13:39Z

...inesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisPartitionGroupOffset.java

@@ -40,10 +40,20 @@
 public class KinesisPartitionGroupOffset implements StreamPartitionMsgOffset {
 private final String _shardId;
 private final String _sequenceNumber;
+ private final OffsetStartStatus _startStatus;


Seems it is not persistent to the offset string. Is this intentional?
We need to think about how to migrate the current table to this new format

This was a miss. Made the change.
I have ensured backward compatibility so for old table we don't need to worry. the start status would always be RESUME for them which is the current behavior as well.

KKcorps · 2024-05-16T06:19:09Z

@Jackie-Jiang please do another review of this

Jackie-Jiang · 2024-05-18T23:05:48Z

pinot-common/src/main/java/org/apache/pinot/common/metadata/segment/SegmentZKMetadata.java

@@ -323,6 +324,15 @@ public void setEndOffset(String endOffset) {
 setValue(Segment.Realtime.END_OFFSET, endOffset);
 }

+ public StreamContinuationMode getContinuationMode() {
+ return _znRecord.getEnumField(Segment.Realtime.CONTINUATION_MODE, StreamContinuationMode.class,
+ StreamContinuationMode.RESUME);


Do you see we might add another mode in the future? I feel a boolean field of whether it is the first segment of a streaming partition is good enough.

Jackie-Jiang · 2024-05-18T23:12:49Z

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

@@ -73,11 +74,12 @@ public KinesisConsumer(KinesisConfig config, KinesisClient kinesisClient) {
 * Fetch records from the Kinesis stream between the start and end KinesisCheckpoint
 */
 @Override
- public KinesisMessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset, int timeoutMs) {
+ public KinesisMessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset,


Can we get #12806 in first if it works well?

Makes sense. I already tested it, merging it now.

Jackie-Jiang · 2024-05-18T23:13:51Z

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

+ requestBuilder = requestBuilder.shardIteratorType(_config.getShardIteratorType());
+ break;
+ }
+ default: //


This is bad practice to leave unexpected mode unhandled. Throw exception instead

Jackie-Jiang · 2024-05-18T23:14:40Z

...inesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisPartitionGroupOffset.java

@@ -41,18 +41,25 @@ public class KinesisPartitionGroupOffset implements StreamPartitionMsgOffset {
 private final String _shardId;
 private final String _sequenceNumber;

+ public static final String STATUS_SEPARATOR = "::";


Revert the changes in this file?

Jackie-Jiang · 2024-05-18T23:16:34Z

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

@@ -55,15 +56,15 @@ default void start(StreamPartitionMsgOffset startOffset) {
 * @throws TimeoutException If the operation could not be completed within timeout
 * @return A batch of messages from the stream partition group
 */
- default MessageBatch fetchMessages(StreamPartitionMsgOffset startOffset, int timeoutMs)
+ default MessageBatch fetchMessages(StreamPartitionMsgOffset startOffset, CommonConstants.Segment.Realtime.StreamContinuationMode continuationMode, int timeoutMs)


I feel a boolean is easier to understand, and I don't see other possible mode

Please also update the javadoc

(MAJOR) This is backward incompatible (this is a public facing interface). We need to add default impl for the new added API and not changing existing signature

Jackie-Jiang · 2024-05-18T23:22:20Z

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

@@ -430,7 +431,8 @@ protected boolean consumeLoop()
 // Update _currentOffset upon return from this method
 MessageBatch messageBatch;
 try {
- messageBatch = _partitionGroupConsumer.fetchMessages(_currentOffset, _streamConfig.getFetchTimeoutMillis());
+ messageBatch = _partitionGroupConsumer.fetchMessages(_currentOffset, _segmentZKMetadata.getContinuationMode(),


(MAJOR) Only the first batch should be count as ingested from new stream.
Within the fetchMessages() API, we should just a boolean inclusive to mark whether we should consume the current offset

Jackie-Jiang · 2024-05-18T23:24:24Z

@swaminathanmanish

…w partition or not

KKcorps requested a review from Jackie-Jiang May 8, 2024 19:17

Jackie-Jiang added ingestion real-time enhancement labels May 11, 2024

Jackie-Jiang reviewed May 11, 2024

View reviewed changes

KKcorps mentioned this pull request May 14, 2024

Refactor kinesis config and change default shard iterator type #13113

Open

Jackie-Jiang reviewed May 18, 2024

View reviewed changes

KKcorps force-pushed the offset_start_tracking branch 3 times, most recently from 3ef926c to ebb7eda Compare May 20, 2024 16:38

KKcorps and others added 7 commits May 23, 2024 23:59

Add support to track if offset needs to honoured or not in Kinesis

a4f757f

Lint fixes

cc2f77e

Store offsets in ZK

4525c41

Store a new field in segment zk metadata to indicate if this was a ne…

1da3e9e

…w partition or not

Make changes in kinesis and pulsar

20c1612

Rename the new config to startOffsetInclusive

39068d0

lint fixes

f43e9a0

KKcorps force-pushed the offset_start_tracking branch from ebb7eda to 2ba2375 Compare May 23, 2024 18:33

Fix compilation error

3198ee4

KKcorps force-pushed the offset_start_tracking branch from 2ba2375 to 3198ee4 Compare May 23, 2024 18:39

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add support to track if offset needs to honoured or not in Kinesis #13112

Add support to track if offset needs to honoured or not in Kinesis #13112

KKcorps commented May 8, 2024

KKcorps commented May 8, 2024

codecov-commenter commented May 9, 2024 •

edited

Jackie-Jiang May 11, 2024

KKcorps May 14, 2024

Jackie-Jiang May 11, 2024

KKcorps May 14, 2024

KKcorps commented May 16, 2024

Jackie-Jiang May 18, 2024

Jackie-Jiang May 18, 2024

KKcorps May 20, 2024 •

edited

Jackie-Jiang May 18, 2024

Jackie-Jiang May 18, 2024

Jackie-Jiang May 18, 2024

Jackie-Jiang May 18, 2024

Jackie-Jiang May 18, 2024

Jackie-Jiang May 18, 2024

Jackie-Jiang commented May 18, 2024

Add support to track if offset needs to honoured or not in Kinesis #13112

Are you sure you want to change the base?

Add support to track if offset needs to honoured or not in Kinesis #13112

Conversation

KKcorps commented May 8, 2024

KKcorps commented May 8, 2024

codecov-commenter commented May 9, 2024 • edited

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

KKcorps commented May 16, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

KKcorps May 20, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jackie-Jiang commented May 18, 2024

codecov-commenter commented May 9, 2024 •

edited

KKcorps May 20, 2024 •

edited