raghavyadav01 commented on code in PR #20020:
URL: https://github.com/apache/pinot/pull/20020#discussion_r4504455536
##########
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:
If the range has aged out of retention, what's the operator's recovery path
after this change? Would an opt-in "allow partial replay" flag be worth having,
or is the expectation that they drop the segment manually?
##########
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:
When several losses are suppressed in one window, only the last location
survives. Should we keep the first one as well, since the earliest gap shows
where the loss started?
##########
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:
Could this reuse `ThrottledLogger` (already used by `KinesisConsumer` for
the same purpose)? If the recover/close tail reporting is the reason not to,
would it make sense to add that to `ThrottledLogger` so both consumers get it?
--
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]
