xiangfu0 commented on code in PR #20020:
URL: https://github.com/apache/pinot/pull/20020#discussion_r4431202606
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1164,16 +1170,31 @@ 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.
+ ///
/// @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()) {
+ return;
+ }
+ _serverMetrics.setValueOfTableGauge(_clientId,
ServerGauge.STREAM_DATA_LOSS, 1L);
+ _dataLossBatchesSinceLastReport++;
+ long now = now();
+ if (_lastDataLossReportTimeMs >= 0 && now - _lastDataLossReportTimeMs <
DATA_LOSS_REPORT_INTERVAL_MS) {
Review Comment:
Good catch. The fix was small, so I made it rather than defer it
(865928b5fd).
`reportDataLoss` now remembers where the most recent suppressed loss was
detected. The first batch processed after the one-minute interval reports it,
even when that batch is healthy. The method runs for every fetched batch,
including empty ones, so the deferred report follows the end of the interval
within one fetch. A distinct loss that the stream recovers from inside the
window now reaches the log and the segment error cache with its own offsets and
the batch count. The deferred report also restarts the interval, so a loss
storm still yields at most one report per minute.
The one gap left is a segment that stops consuming within that minute: the
suppressed event then shows only through the gauge, which is set for every
lossy batch. `testSuppressedDataLossIsReportedAfterRecovery` covers the
deferred report, the restarted interval, and that nothing is reported twice.
--
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]
