xiangfu0 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4431158167
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -605,6 +616,65 @@ 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.
+ ///
+ /// @return `true` if `_currentOffset` was advanced
+ private boolean adoptNextBatchOffsetOfEmptyBatch(MessageBatch messageBatch) {
+ if (_state == State.CATCHING_UP) {
Review Comment:
It cannot stall the replica for good, and the ONLINE transition is indeed
the intended way out. I added a note saying so on
`adoptNextBatchOffsetOfEmptyBatch` and a test for that exact path in d6ebb13f63.
What the follower does while it waits: it stays in CATCHING_UP (no time
limit there), and each empty batch costs one 100 ms idle sleep. The consumer
reports the same validated position on every poll, so after the first move is
validated there is no further `beginningOffsets` call until the idle timeout
recreates the consumer (three minutes by default), which costs one more. Once
the committer commits, `goOnlineFromConsuming` stops the loop within one poll.
`catchupToFinalOffset` then switches to CONSUMING_TO_ONLINE, adopts the
position, and `buildSegmentAndReplace` reuses the local segment only on a
matching committed CRC, otherwise downloads it.
If a record arrives before the commit, the follower indexes it and
overshoots ("Past max offset"), which ends in a download on the ONLINE
transition: the same recovery as any overshoot today. If the committer fails
instead, this follower never posts `segmentConsumed` again, so it cannot take
over the commit. Recovery then goes through the replicas that do post (or the
validation manager), as it does for any follower that cannot reach its catch-up
target. That is the cost of not letting an empty position make a protocol
follower commit-eligible.
`testProtocolCatchupToSkippedOffsetEscapesThroughOnlineTransition` drives
the whole path: three empty polls in protocol catch-up leave the offset where
it was, then the ONLINE transition reaches the target in one poll and either
reuses the segment (CRC match) or downloads it (mismatch).
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -605,6 +616,65 @@ 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.
+ ///
+ /// @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;
+ _currentOffset = nextBatchOffset;
+ return true;
+ }
+
+ /// ONLINE catch-up can stop before a message ending past the committed
final offset. This only establishes
+ /// offset equality, not row equality, so record the unverified advance
before publishing the final offset.
+ /// Protocol CATCHING_UP must retain ordinary record advancement and
overshoot handling instead of manufacturing
+ /// a target match that could make this replica the committer. Reported loss
also retains overshoot recovery.
+ ///
+ /// @return `true` if consumption must stop before the message
+ private boolean reachedFinalOffsetBeforeMessage(MessageBatch messageBatch,
StreamPartitionMsgOffset nextOffset) {
+ if (_state != State.CONSUMING_TO_ONLINE || _finalOffset == null ||
messageBatch.hasDataLoss()
+ || _currentOffset.compareTo(_finalOffset) >= 0 ||
nextOffset.compareTo(_finalOffset) <= 0) {
Review Comment:
Agreed, it is the flag that makes this safe. I added a comment at the settle
point in d6ebb13f63: the message is left unindexed only because
`_hasUnverifiedOffsetAdvance` is set there, so `buildSegmentAndReplace()`
reuses the local segment only on a matching committed CRC and downloads it
otherwise. A message the committer did index, a straddling one (offset < final
< next) included, therefore cannot leave a divergent copy. For Kafka the
straddling case cannot occur, since next = offset + 1. What happens there is a
message at exactly the final offset after a skipped marker, and that message
belongs to the next segment anyway.
--
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]
