xiangfu0 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4366260608
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -656,12 +734,16 @@ private boolean processStreamEvents(MessageBatch
messageBatch, long idlePipeSlee
throw new RuntimeException("Realtime segment full");
}
- // Decode message
StreamMessage<?> streamMessage = messageBatch.getStreamMessage(index);
- StreamDataDecoderResult decodedRow =
_streamDataDecoder.decode(streamMessage);
StreamMessageMetadata metadata = streamMessage.getMetadata();
StreamPartitionMsgOffset offset = metadata.getOffset();
StreamPartitionMsgOffset nextOffset = metadata.getNextOffset();
+ if (reachedFinalOffsetBeforeMessage(messageBatch, nextOffset)) {
Review Comment:
Intended, and the same as the existing mid-batch exits. When
`endCriteriaReached()` fires inside a batch today (row or time limit), the loop
also breaks before the remaining messages and this block still reports
`getLastMessageMetadata()`, so the semantics of the post-loop metrics have
always been "the latest message the consumer fetched", not "the latest message
this segment indexed". Settling on `_finalOffset` adds one more way to stop
inside a batch and inherits that.
Why I left it alone for the catch-up case: the ingestion delay tracker is
keyed by partition, not by segment, and ages the delay at read time (`now -
ingestionTimeMs`). The skipped message is by construction the first message of
the successor segment (it starts at exactly `_finalOffset`), so the only effect
is that, between this stop and the successor reporting its first batch, the
partition metric is based on a message one position ahead of what this replica
indexed: offset lag understated by one and the delay by the timestamp gap
between the two messages. The handoff interval itself dominates that error
since the tracker keeps aging the last reported timestamp meanwhile. Making the
post-loop block use the last processed message instead would be a reasonable
tightening, but it changes the existing premature-exit path too, so I would
rather do it separately than inside this change.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
