xiangfu0 commented on code in PR #20020:
URL: https://github.com/apache/pinot/pull/20020#discussion_r4504489484
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java:
##########
@@ -272,6 +276,8 @@ 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:
Nothing gets committed and the segment stays in ERROR. Each validation run
retries re-ingestion and fails the same way, so the operator either restores
the range from another source or deletes the segment to accept the loss (on a
pauseless table the next segment is already consuming from this one's end
offset). The error message now spells that out (604c40a9eb). I'd rather not add
an opt-in partial replay here, since it would bring back the silent gap that
the delete makes an explicit choice, but a per-request flag on the re-ingestion
endpoint could be a follow-up if you think it's worth having.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java:
##########
@@ -185,14 +197,57 @@ public synchronized KafkaMessageBatch
fetchMessages(StreamPartitionMsgOffset sta
/// timeout as [#poll]. Returns [Long#MIN_VALUE] when it cannot be
determined so the caller
/// treats an offset gap as expected (no data loss) rather than raising a
false positive.
private long getLogStartOffset(int timeoutMs) {
+ long logStartOffset;
try {
- return _consumer.beginningOffsets(List.of(_topicPartition),
Duration.ofMillis(timeoutMs))
+ logStartOffset = _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;
}
+ reportSuppressedLogStartOffsetLookupFailures("the lookup succeeded again");
+ return logStartOffset;
+ }
+
+ /// 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, and their
count is also reported when lookups
+ /// succeed again or the consumer is closed (see
[#reportSuppressedLogStartOffsetLookupFailures]), so the tail of a
+ /// failure storm is not lost when no further failure comes to trigger the
next WARN.
+ private void logLogStartOffsetLookupFailure(Exception e) {
Review Comment:
I left this one as is. `ThrottledLogger` lives in pinot-common, which the
Kafka plugin doesn't depend on (it builds against pinot-spi and
pinot-kafka-base, while Kinesis declares pinot-common itself), and it isn't
thread-safe, while `close()` here can drain the count from another thread.
Moving `ThrottledLogger` into pinot-spi with a flush for the dropped counts,
and switching both consumers to it, would make a good follow-up, and I'm happy
to send it.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1165,16 +1185,61 @@ public Map<String, ConsumerPartitionState>
getConsumerPartitionState(
/// Checks and reports if the consumer is going through data loss.
///
+ /// The [ServerGauge#STREAM_DATA_LOSS] gauge is set for every batch that
reports loss. The ERROR log line and the
+ /// segment error entry are emitted at most once per
[#DATA_LOSS_REPORT_INTERVAL_MS] per segment, carrying the
+ /// number of affected batches since the previous report, so a stream that
keeps reporting loss on every poll
+ /// does not flood the log or churn the segment error cache. A batch with
loss whose report was suppressed is
+ /// reported by the first batch processed after the interval has elapsed,
even when that batch and every one before
+ /// it is healthy, so a loss event that the stream recovers from within the
interval still reaches the log and the
+ /// segment error cache. This method runs for every fetched batch, including
empty ones, so the deferred report
+ /// follows the end of the interval within one fetch. When the consume loop
exits before that,
+ /// [#flushDataLossReport] reports it instead.
+ ///
/// @param messageBatch Message batch to validate
- private void reportDataLoss(MessageBatch messageBatch) {
- if (messageBatch.hasDataLoss()) {
- _serverMetrics.setValueOfTableGauge(_clientId,
ServerGauge.STREAM_DATA_LOSS, 1L);
- String message = "Message loss detected in stream partition: " +
_partitionGroupId + " for table: "
- + _tableNameWithType + " startOffset: " + _currentOffset + "
batchFirstOffset: "
- + messageBatch.getFirstMessageOffset();
- _segmentLogger.error(message);
- _realtimeTableDataManager.addSegmentError(_segmentNameStr, new
SegmentErrorInfo(now(), message, null));
+ @VisibleForTesting
+ void reportDataLoss(MessageBatch messageBatch) {
+ if (!messageBatch.hasDataLoss()) {
+ if (_unreportedDataLossLocation != null) {
+ long now = now();
+ if (now - _lastDataLossReportTimeMs >= DATA_LOSS_REPORT_INTERVAL_MS) {
+ logDataLoss(now, _unreportedDataLossLocation);
+ }
+ }
+ return;
}
+ _serverMetrics.setValueOfTableGauge(_clientId,
ServerGauge.STREAM_DATA_LOSS, 1L);
+ _dataLossBatchesSinceLastReport++;
+ String location = "startOffset: " + _currentOffset + " batchFirstOffset: "
+ messageBatch.getFirstMessageOffset();
+ long now = now();
+ if (_lastDataLossReportTimeMs >= 0 && now - _lastDataLossReportTimeMs <
DATA_LOSS_REPORT_INTERVAL_MS) {
+ _unreportedDataLossLocation = location;
Review Comment:
Done in 604c40a9eb. A report covering several batches now names both the
earliest and the latest location (`earliest startOffset: ... batchFirstOffset:
..., latest ...`), and collapses to one when they're the same.
`testDataLossReportNamesEarliestAndLatestLocation` covers both the
next-lossy-batch and the healthy-batch paths.
_🤖 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]
