xiangfu0 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4459898158
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -605,6 +627,78 @@ protected boolean consumeLoop()
return true;
}
+ // TODO Issue 5359 Need to find a way to bump metrics without getting actual
offset value.
+ private void updateHighestStreamOffsetConsumedGauge() {
+ if (_currentOffset instanceof LongMsgOffset) {
+ // TODO: only LongMsgOffset supplies long offset value.
+ _serverMetrics.setValueOfTableGauge(_clientId,
ServerGauge.HIGHEST_STREAM_OFFSET_CONSUMED,
+ ((LongMsgOffset) _currentOffset).getOffset());
+ }
+ }
+
+ /// Adopts an empty batch's advanced stream position. Transaction markers
and aborted records move a stream's
+ /// position without delivering rows, so adoption keeps the position and lag
current. The skipped offsets are treated
+ /// like a gap between two delivered messages, which needs no further proof
either: the stream reports a loss
+ /// through `hasDataLoss()`.
+ ///
+ /// @return `true` if `_currentOffset` was advanced
+ private boolean adoptNextBatchOffsetOfEmptyBatch(MessageBatch messageBatch) {
+ StreamPartitionMsgOffset nextBatchOffset =
messageBatch.getOffsetOfNextBatch();
+ if (nextBatchOffset == null || nextBatchOffset.compareTo(_currentOffset)
<= 0) {
+ return false;
+ }
+ nextBatchOffset = capAtFinalOffset(messageBatch, nextBatchOffset);
+ if (_segmentLogger.isDebugEnabled()) {
+ _segmentLogger.debug("Empty batch moved the stream position. Advancing
from {} to {}", _currentOffset,
+ nextBatchOffset);
+ }
+ _currentOffset = nextBatchOffset;
+ return true;
+ }
+
+ /// While catching up to a final offset, caps at it the position of a batch
that indexes no row (no messages, or
+ /// only filtered ones such as tombstones) when that position lies past the
final offset. The final offset then lies
+ /// inside an interval no replica indexes rows from, so this replica holds
exactly the rows below it. A committer can
+ /// stop at such an offset, e.g. one past a Kafka commit marker it skipped
on an empty poll. A batch reporting loss is
+ /// not capped and keeps the existing overshoot handling.
+ @Nullable
+ private StreamPartitionMsgOffset capAtFinalOffset(MessageBatch messageBatch,
+ @Nullable StreamPartitionMsgOffset nextBatchOffset) {
+ if (nextBatchOffset != null && isCatchingUpToFinalOffset() &&
!messageBatch.hasDataLoss()
+ && nextBatchOffset.compareTo(_finalOffset) > 0) {
+ return _finalOffset;
+ }
+ return nextBatchOffset;
+ }
+
+ /// While catching up to a final offset, stops before a message that starts
at or past it and settles on the final
+ /// offset. This happens when the final offset is one past offsets the
stream never delivers, such as a Kafka commit
+ /// marker the committer skipped on an empty poll. The offsets below the
message are a gap like any other inside a
+ /// data poll, so this replica then holds exactly the rows below the final
offset.
+ ///
+ /// The existing overshoot handling, which ends in a download, is kept for a
batch reporting loss, for a message
+ /// straddling the final offset (starting below it), and for a message whose
offset equals its next offset (inclusive
+ /// offsets such as Kinesis sequence numbers): with those, a message past
the final offset means this replica's view
+ /// misses one the committer indexed.
+ ///
+ /// @return `true` if consumption must stop before the message
+ private boolean reachedFinalOffsetBeforeMessage(MessageBatch messageBatch,
StreamPartitionMsgOffset offset,
+ StreamPartitionMsgOffset nextOffset) {
+ if (!isCatchingUpToFinalOffset() || messageBatch.hasDataLoss() ||
_currentOffset.compareTo(_finalOffset) >= 0
Review Comment:
Non-blocking residual risk: Pulsar's `nextOffset` is a synthetic (ledger,
entry+1), so it has no legitimate gaps. A follower in CATCHING_UP can only see
a message at or past the final offset by skipping one, and Pulsar never reports
loss. The same holds for Kafka read_committed, which never flags loss. Before
this PR such a follower overshot and downloaded. Now it settles, and KEEP
retains it without a CRC check (the `settled` case of
`testKeepRetainsSegmentAtReachedPosition`). It's rare (retention passing a
follower mid-catch-up), but the committed CRC is in ZK by KEEP time. Verifying
it after a settle or cap, or limiting settles to streams with real gaps, would
keep that safety net.
--
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]
