xiangfu0 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4450664068
##########
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:
Fixed in 40421c69a4: the flag is gone. On an idle transactional partition
KEEP retains the segment as before, without a metadata fetch.
_🤖 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;
+ _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()
Review Comment:
Fixed in 40421c69a4. Protocol CATCHING_UP now settles on the final offset in
front of a no-loss message that starts at or past it, as suggested, so the
active-topic case posts the target instead of throwing "Past max offset". The
test also requires the message's offset to be below its next offset.
Kinesis-style inclusive offsets, where a message past the final offset means
the committer's last record is missing, and straddling messages keep the old
overshoot handling. A batch of only tombstones past the target is now capped
the same way.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java:
##########
@@ -272,6 +283,16 @@ public void run() {
// Fetch messages
MessageBatch messageBatch = _consumer.fetchMessages(_currentOffset,
_fetchTimeoutMs);
+ Preconditions.checkState(!messageBatch.hasDataLoss(), "Stream data
loss at offset %s while replaying to %s",
Review Comment:
Intended. The fail-closed check comes from #20020, and this PR extends it to
the empty batches it now adopts. Re-ingestion replaces a segment whose end
offset is already committed, so replaying across a reported loss would publish
a segment silently missing rows. I noted the operator-intervention consequence
in the description's `StatelessRealtimeSegmentWriter` section.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
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:
Fixed in 40421c69a4: the flag is gone. Adopted, capped and settled offsets
all go through the existing CRC guard. A pauseless replica, and a committer
whose commit end failed, build locally again as before.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
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:
Fixed in 40421c69a4. With the flag gone, KEEP builds and retains as before,
without a metadata fetch. That holds whether the replica adopted marker+1 while
consuming, adopted or capped it during catch-up, or settled on it
(`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]
