xiangfu0 commented on code in PR #20020:
URL: https://github.com/apache/pinot/pull/20020#discussion_r4368859018
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java:
##########
@@ -298,7 +302,12 @@ public void run() {
}
}
- _currentOffset = messageBatch.getOffsetOfNextBatch();
+ StreamPartitionMsgOffset nextBatchOffset =
messageBatch.getOffsetOfNextBatch();
+ if (messageBatch.getUnfilteredMessageCount() == 0 &&
nextBatchOffset.compareTo(_currentOffset) > 0) {
+ throw new IllegalStateException("Cannot verify empty stream offset
advance from " + _currentOffset + " to "
Review Comment:
Confirmed, and fixed in 784ea00a04. A poll that returns only commit/abort
markers while the position moves is routine on a transactional topic, and that
position can land below, at, or past the end offset, so failing the replay on
it was wrong.
The writer now mirrors `RealtimeSegmentDataManager`. On a batch without
unfiltered messages it keeps `_currentOffset` and polls again, and the consumer
keeps its own progress (no re-seek on a repeated offset after an empty fetch).
Every batch is checked for reported data loss before anything from it is
indexed, and loss fails the replay without publishing progress. The gap from
the retained offset to the first record of the next non-empty batch is exactly
what the consumer's loss check validates, so a marker-only tail below, at, or
past the end completes once a record at or after the end is delivered.
I also dropped the "first record after an empty advance must be below the
end" rule from the intermediate revision. The next data poll's loss check
validates the same gap either way, so that rule added no guarantee and only
failed marker-only tails. There is no extra idle timeout in the writer:
`ReingestionResource` already bounds consumption with its configurable timeout
and then stops the writer.
`StatelessRealtimeSegmentWriterTest` now covers markers below, at and past
the end, filtered batches, and loss reported on an empty batch or on a data
batch (not indexed, offset unchanged). The re-ingestion resource tests pass
unchanged.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -165,13 +170,17 @@ public synchronized KafkaMessageBatch
fetchMessages(StreamPartitionMsgOffset sta
}
}
long offsetOfNextBatch = _nextReadOffset;
- // A gap between the requested startOffset and the first returned offset
does NOT by itself
- // imply data loss. Transactional producers write commit/abort control
records that occupy
- // offsets but are never delivered to the consumer (even under
read_uncommitted), so a
- // contiguous stream of user records legitimately has offset gaps. Real
data loss only
- // happens when the requested startOffset is below the log's start offset,
i.e. the broker
- // has already deleted (via retention or truncation) records at or after
startOffset. For
- // read_committed we never flag loss because aborted-record gaps are
always expected.
+ // A gap between the requested startOffset and the first returned offset
does NOT by itself imply data loss.
+ // Transactional producers write commit/abort control records that occupy
offsets but are never delivered to
+ // the consumer (under either isolation level), so a contiguous stream of
user records legitimately has offset
+ // gaps. On a low-traffic transactional topic every transaction produces
one: after record N the caller is
+ // parked at the marker offset N + 1 and the next record lands at N + 2.
The Kafka client gives no way to tell
+ // a skipped control record from an out-of-range reset (both just move the
position), so the only reliable
+ // signal is the broker's log start offset: real data loss means records
at/after startOffset were deleted
+ // (retention or truncation), i.e. the log start offset is now above
startOffset. That lookup is one round
+ // trip per gap, never on the contiguous hot path. For read_committed we
never flag loss because aborted-record
+ // gaps are always expected, and the caller's startOffset can lag far
behind the consumer's position through
+ // an aborted region.
boolean hasDataLoss = false;
if (!_isReadCommitted && firstOffset > startOffset) {
Review Comment:
In this PR the loss is not missed, only reported one poll later, and that is
deliberate here.
No caller adopts an empty batch's offset. `RealtimeSegmentDataManager` keeps
its offset on an empty batch, and after 784ea00a04 so does the re-ingestion
writer. So after a reset surfaces on an empty poll, the caller asks again from
the same offset and the consumer does not re-seek. The next poll that returns
records then has its first offset past the requested one, and the log-start
lookup sees the log starting above it and flags the loss. The new
`testResetSeenOnEmptyPollIsReportedOnNextDataPoll` pins this in both plugins
(24c86ab4fc). If the stream delivers nothing after the reset, nothing more is
consumed until a record arrives, and that record triggers the report.
Validating the move on the empty poll itself would add a second broker
lookup per transaction on exactly the low-traffic topics this PR targets.
#20024, where the data manager does adopt the empty batch's offset, tracks a
validated position and checks the move on the empty poll there. Same answer for
kafka-4.0.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -188,12 +197,30 @@ private long getLogStartOffset(int timeoutMs) {
return _consumer.beginningOffsets(List.of(_topicPartition),
Duration.ofMillis(timeoutMs))
.getOrDefault(_topicPartition, Long.MIN_VALUE);
} catch (Exception e) {
- LOGGER.warn("Failed to read log start offset for {}; treating the offset
gap as no data loss "
- + "(this can mask genuine data loss if it persists)",
_topicPartition, e);
+ logLogStartOffsetLookupFailure(e);
return Long.MIN_VALUE;
}
}
+ /// Logs a failed log-start-offset lookup at WARN at most once per
+ /// [#LOG_START_OFFSET_LOOKUP_FAILURE_LOG_INTERVAL_MS], reporting how many
failures were suppressed since the
+ /// previous line. Suppressed failures are still logged at DEBUG.
+ private void logLogStartOffsetLookupFailure(Exception e) {
+ long nowMs = System.currentTimeMillis();
+ if (nowMs - _lastLogStartOffsetLookupFailureLogMs <
LOG_START_OFFSET_LOOKUP_FAILURE_LOG_INTERVAL_MS) {
+ _suppressedLogStartOffsetLookupFailures++;
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Failed to read log start offset for {} (WARN
suppressed)", _topicPartition, e);
+ }
+ return;
+ }
+ LOGGER.warn("Failed to read log start offset for {} ({} similar failures
suppressed since the last report); "
Review Comment:
Fixed in 24c86ab4fc for both kafka-3.0 and kafka-4.0. The suppressed count
is now reported at WARN, with the count, as soon as a lookup succeeds again and
when the consumer closes, so a failure storm that ends for good still reports
its tail. The counter is an `AtomicInteger` because `close()` can run outside
`fetchMessages()`. `testSuppressedLookupFailuresAreReportedOnRecoveryAndClose`
covers both paths.
--
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]
