xiangfu0 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4460691522
##########
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:
Agreed, fixed in 9bc4bff864. Catch-up now records when it capped a position
at the final offset or settled on it in front of a message, and on KEEP such a
segment is built against the committed ZK metadata so the existing CRC guard
runs, as on the ONLINE transition. A match retains it, a mismatch goes to
DISCARDED so the ONLINE transition downloads, and a pauseless table or an unset
CRC builds locally as on the ONLINE path. A position adopted exactly at the
final offset still keeps without a metadata fetch, so the extra ZK read only
happens where equality was synthesized.
I kept settling for every stream rather than limiting it to streams with
real gaps, because Kafka read_committed is the main case it exists for and it
never reports loss either. This is also the narrow form raghavyadav01 suggested
earlier: a check only where offset equality is synthesized, and DISCARDED
rather than ERROR when reuse is refused. Covered by the capped and settled rows
of `testKeepRetainsSegmentAtReachedPosition`.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
--
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]
