xiangfu0 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4450664927
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -605,6 +620,74 @@ 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 consumer position for INITIAL
consumption or ONLINE catch-up.
+ /// Transaction markers and aborted records can move a stream's position
without delivering rows, so adoption
+ /// keeps position and lag current. But a position, including a batch with
`hasDataLoss() == false`, does not prove
+ /// the skipped interval contained no rows: validation may be unavailable,
or the topic may have been compacted.
+ /// Remember that uncertainty for local-reuse CRC validation even when the
new position is below the final offset.
+ ///
+ /// Protocol CATCHING_UP can subsequently become the committer, so it must
not synthesize progress from an empty
+ /// batch. ONLINE catch-up can cap the position at the committed final
offset, because any local reuse is guarded
+ /// by the committed CRC. A reported loss is not capped; ordinary overshoot
recovery still applies.
+ ///
+ /// A follower in protocol CATCHING_UP to a final offset the stream only
reaches by skipping (a committer that
+ /// adopted the position past a commit marker) therefore cannot reach it by
consuming. It keeps polling without
+ /// advancing, one idle sleep per empty batch, until the ONLINE transition
that follows the commit. That transition
+ /// is the intended way out: its CONSUMING_TO_ONLINE catch-up adopts the
position, and the local segment is reused
+ /// only on a matching committed CRC, otherwise downloaded.
+ ///
+ /// @return `true` if `_currentOffset` was advanced
+ private boolean adoptNextBatchOffsetOfEmptyBatch(MessageBatch messageBatch) {
+ if (_state == State.CATCHING_UP) {
Review Comment:
Agreed, the exclusion is gone in 40421c69a4. Protocol catch-up adopts like
initial consumption, caps a no-loss position at the target (also for a batch of
only tombstones), and settles in front of a message that starts at or past it.
Followers reach a marker+1 target and can post it and be elected committer, so
no time bound is needed. A reported loss keeps the old overshoot handling.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -605,6 +620,74 @@ 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 consumer position for INITIAL
consumption or ONLINE catch-up.
+ /// Transaction markers and aborted records can move a stream's position
without delivering rows, so adoption
+ /// keeps position and lag current. But a position, including a batch with
`hasDataLoss() == false`, does not prove
+ /// the skipped interval contained no rows: validation may be unavailable,
or the topic may have been compacted.
+ /// Remember that uncertainty for local-reuse CRC validation even when the
new position is below the final offset.
+ ///
+ /// Protocol CATCHING_UP can subsequently become the committer, so it must
not synthesize progress from an empty
+ /// batch. ONLINE catch-up can cap the position at the committed final
offset, because any local reuse is guarded
+ /// by the committed CRC. A reported loss is not capped; ordinary overshoot
recovery still applies.
+ ///
+ /// A follower in protocol CATCHING_UP to a final offset the stream only
reaches by skipping (a committer that
+ /// adopted the position past a commit marker) therefore cannot reach it by
consuming. It keeps polling without
+ /// advancing, one idle sleep per empty batch, until the ONLINE transition
that follows the commit. That transition
+ /// is the intended way out: its CONSUMING_TO_ONLINE catch-up adopts the
position, and the local segment is reused
+ /// only on a matching committed CRC, otherwise downloaded.
+ ///
+ /// @return `true` if `_currentOffset` was advanced
+ private boolean adoptNextBatchOffsetOfEmptyBatch(MessageBatch messageBatch) {
+ if (_state == State.CATCHING_UP) {
+ return false;
+ }
+ StreamPartitionMsgOffset nextBatchOffset =
messageBatch.getOffsetOfNextBatch();
+ if (nextBatchOffset == null || nextBatchOffset.compareTo(_currentOffset)
<= 0) {
+ return false;
+ }
+ if (_state == State.CONSUMING_TO_ONLINE && _finalOffset != null &&
!messageBatch.hasDataLoss()
+ && nextBatchOffset.compareTo(_finalOffset) > 0) {
+ nextBatchOffset = _finalOffset;
+ }
+ if (_segmentLogger.isDebugEnabled()) {
+ _segmentLogger.debug("Empty batch moved the stream position. Advancing
from {} to {}", _currentOffset,
+ nextBatchOffset);
+ }
+ _hasUnverifiedOffsetAdvance = true;
Review Comment:
Went one step further in 40421c69a4 and removed the flag. The cap and the
settle only claim offsets inside a gap no replica indexes rows from. That is
the same evidence as an uncapped adoption or a gap inside a data poll, so the
existing CRC guard is enough. Where that evidence is missing, the old
overshoot-and-download path stays instead of a flag: a batch reporting loss, a
message straddling the final offset, and an inclusive offset. Kinesis messages
have `offset == nextOffset`, so there a message past the final offset means the
committer's last record is missing.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -577,6 +585,13 @@ protected boolean consumeLoop()
}
_currentOffset = nextOffset;
lastUpdatedOffset =
_streamPartitionMsgOffsetFactory.create(nextOffset);
+ } else if (adoptNextBatchOffsetOfEmptyBatch(messageBatch)) {
Review Comment:
The plan is release-note coverage rather than a config. The mixed window
preserves correctness: old replicas end in a download of the committed segment,
with no loss or duplicates. It is also limited to transactional partitions
while old and new servers mix. A gate would either leave idle lag at 1 by
default or need a later flip. I added the release-note text to the Rolling
upgrade section of the description.
_🤖 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]
