xiangfu0 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4486574351
##########
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:
Correction: my reply above is out of date since 9bc4bff864, which is
70607cb657 after the restack. KEEP no longer skips the metadata fetch in every
case. When catch-up capped a position at the final offset or settled on it in
front of a message, KEEP fetches the committed ZK metadata and builds against
it, so the existing CRC guard runs as on the ONLINE transition. A match retains
the segment, and a mismatch goes to DISCARDED rather than ERROR, so the ONLINE
transition downloads. A pauseless table or an unset CRC still builds locally,
and a position adopted exactly at the final offset still keeps without a fetch.
The reason is that a cap or a settle is only as good as the stream's loss
reporting, and Pulsar and Kafka read_committed never report loss. A replica
that skipped the committer's last messages, for example because retention
passed it during catch-up, would cap or settle too. So this ended up as the
fetch-and-DISCARDED approach you suggested here. It is covered by the capped
and settled rows of `testKeepRetainsSegmentAtReachedPosition`.
##########
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:
Correction: "removed the flag" is out of date since 9bc4bff864, which is
70607cb657 after the restack. There is a narrow flag again, set only where
offset equality is synthesized, as you suggested: the catch-up cap at the final
offset and the settle in front of a message. Plain adoptions, validated marker
skips included, and gaps inside a data poll don't set it, so fetch timing no
longer decides reuse. The flag only matters on KEEP, where a flagged segment is
checked against the committed CRC (details on the KEEP thread).
I brought it back because the cap and the settle rely on the stream
reporting loss, and Pulsar and Kafka read_committed never do, so a replica that
skipped the committer's last messages would reach the final offset the same
way. The overshoot-and-download path for a reported loss, a straddling message
and inclusive offsets is unchanged.
--
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]
