raghavyadav01 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4448542474
##########
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:
During a rolling upgrade, old replicas can't consume up to a marker+1 end
offset. On idle partitions each of them waits out the 31s catch-up budget and
then downloads. Old-version re-ingestion of those segments also times out on an
idle stream. Could adoption be behind a config for a release, or is
release-note coverage the plan?
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1412,12 +1533,21 @@ protected boolean buildSegmentAndReplace()
/// ordering); swapping in a divergent copy corrupts upsert metadata and
leaves replicas inconsistent otherwise. So
/// when `committedSegmentZKMetadata` carries a CRC, a mismatch discards the
local build (returns `false`) and the
/// caller downloads the committed segment. Skipped for pauseless tables and
segments whose CRC is not yet set (the
- /// COMMITTING window), so we never download a not-yet-uploaded segment (see
PR #17885).
+ /// COMMITTING window), so we never download a not-yet-uploaded segment (see
PR #17885). After an unverified offset
+ /// advance, those exceptions cannot establish row equality: local reuse
requires an enabled, matching committed
+ /// CRC check. Otherwise the caller must use the existing download/recovery
path.
///
- /// @param committedSegmentZKMetadata committed ZK metadata carrying the
CRC, or `null` to skip the check
+ /// @param committedSegmentZKMetadata committed ZK metadata carrying the
CRC, or `null` for legacy unflagged reuse
/// @return `true` if built and replaced locally; `false` if the build
failed or was rejected on a CRC mismatch
protected boolean buildSegmentAndReplace(@Nullable SegmentZKMetadata
committedSegmentZKMetadata)
throws Exception {
+ boolean crcCheckEnabled = committedSegmentZKMetadata != null &&
committedSegmentZKMetadata.getCrc() >= 0
Review Comment:
For pauseless tables `crcCheckEnabled` is always false, so once the flag is
set the segment can never be built locally. On a transactional topic,
non-committers will then always wait on the COMMITTING download. And a
committer whose commit-end failed (back to HOLDING when the ONLINE transition
arrives) will try to download a segment only it can upload. Could pauseless
keep the old local-build behavior here, or could the flag stay unset for
validated adoptions?
##########
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:
An INITIAL_CONSUMING replica that adopts the same position and posts it can
also be elected committer, so I don't see what the CATCHING_UP exclusion
protects. The cost: if the committer at marker+1 dies or keeps failing on an
idle partition, the followers can never reach the target or re-post. The
partition then only recovers once a new record makes them overshoot. Could
CATCHING_UP adopt capped, no-loss positions like INITIAL does, or at least have
a time bound?
##########
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:
The flag gets set for every validated marker skip, but a compaction gap
skipped inside a data poll never sets it. So whether a segment loses local
reuse depends on fetch timing, not on whether anything was actually uncertain.
Would it work to set it only where offset equality is synthesized (the
CONSUMING_TO_ONLINE cap and settle) or when the batch reports loss?
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1412,12 +1533,21 @@ protected boolean buildSegmentAndReplace()
/// ordering); swapping in a divergent copy corrupts upsert metadata and
leaves replicas inconsistent otherwise. So
/// when `committedSegmentZKMetadata` carries a CRC, a mismatch discards the
local build (returns `false`) and the
/// caller downloads the committed segment. Skipped for pauseless tables and
segments whose CRC is not yet set (the
- /// COMMITTING window), so we never download a not-yet-uploaded segment (see
PR #17885).
+ /// COMMITTING window), so we never download a not-yet-uploaded segment (see
PR #17885). After an unverified offset
+ /// advance, those exceptions cannot establish row equality: local reuse
requires an enabled, matching committed
+ /// CRC check. Otherwise the caller must use the existing download/recovery
path.
///
- /// @param committedSegmentZKMetadata committed ZK metadata carrying the
CRC, or `null` to skip the check
+ /// @param committedSegmentZKMetadata committed ZK metadata carrying the
CRC, or `null` for legacy unflagged reuse
/// @return `true` if built and replaced locally; `false` if the build
failed or was rejected on a CRC mismatch
protected boolean buildSegmentAndReplace(@Nullable SegmentZKMetadata
committedSegmentZKMetadata)
throws Exception {
+ boolean crcCheckEnabled = committedSegmentZKMetadata != null &&
committedSegmentZKMetadata.getCrc() >= 0
+ && !PauselessConsumptionUtils.isPauselessEnabled(_tableConfig);
+ if (_hasUnverifiedOffsetAdvance && !crcCheckEnabled) {
Review Comment:
The KEEP response calls `buildSegmentAndReplace()` with null metadata, so a
flagged segment returns false there. The replica goes to ERROR and logs "Could
not build segment" at ERROR level. On an idle transactional partition all
replicas end at the same marker+1 offset, so every non-committer would take
this path. Could KEEP fetch the committed ZK metadata so the CRC check can run,
and go to DISCARDED instead of ERROR when reuse is refused?
--
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]
