xiangfu0 commented on code in PR #20020:
URL: https://github.com/apache/pinot/pull/20020#discussion_r4368368546
##########
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:
Minor/for-awareness: genuine loss discovered via an *empty* poll is never
flagged, since this gate needs `firstOffset > startOffset` and `firstOffset` is
-1 on an empty batch. If retention deletes records at `startOffset` and the
poll returns zero records, no loss is reported. Looks like an accepted
limitation given the design — just flagging it.
(Same applies to the identical kafka-4.0 consumer.)
##########
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:
**MAJOR (worth confirming before merge).** This fails the whole reingestion
on any pure-empty batch that advanced its position. On a transactional Kafka
topic a `poll()` can legitimately return only commit/abort markers (zero
unfiltered records) while `position()` snaps forward — and this can land
*strictly between* `_currentOffset` and `_endOffset` when the surrounding user
records fall into different polls (`max.poll.records`, fetch boundaries).
That's not data loss.
The live `RealtimeSegmentDataManager` handles the identical batch
gracefully: on `getUnfilteredMessageCount()==0` it does **not** advance
`_currentOffset`, it just re-polls (the no-re-seek rule preserves the
consumer's progress). So the writer is strictly less tolerant than the
production consumer for the same stream, and a topic whose tail batches markers
alone can get its reingestion wedged (no internal retry here).
Could the writer mirror RSDM instead — on a pure-empty advance, keep
`_currentOffset` and re-poll under a bounded idle timeout — rather than
aborting? (Note: gating on `hasDataLoss()` wouldn't work, since an empty batch
reports `firstOffset=-1` and never flags loss.)
##########
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:
Minor: the suppressed-failure count is only flushed on the *next* WARN. If a
failure storm stops for good, the trailing
`_suppressedLogStartOffsetLookupFailures` is never surfaced (DEBUG only). Small
observability gap; the no-loss fallback decision itself is unaffected.
(Same applies to the identical kafka-4.0 consumer.)
--
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]
