xiangfu0 commented on code in PR #20020:
URL: https://github.com/apache/pinot/pull/20020#discussion_r4450555903
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -1164,16 +1174,49 @@ 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.
+ ///
/// @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) {
Review Comment:
Done in af886d3c7f. `consumeLoop()` now flushes any suppressed report in a
`finally` when it exits, whether that's the end criteria, a stop or an
exception. So a loss found at t=30s on a segment that commits at t=40s is
logged and recorded in the segment error cache.
`testSuppressedDataLossIsReportedWhenConsumptionStops` runs the real loop to
its end criteria and commit inside the window, and checks that the second loss
is reported exactly once.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java:
##########
@@ -272,6 +281,12 @@ 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:
Agreed, and it's intended. When a replay's start offset has aged out of
retention, it now fails with "Stream data loss at offset X while replaying to
Y". Before, it committed a segment that silently missed the deleted records,
and that segment replaced the errored one as if it were complete. There's no
opt-out. I've rewritten the description with a "Behavior change" section, which
also notes that this only happens under `read_uncommitted` because loss is
never flagged under `read_committed`. I also added the `release-notes` label.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/writer/StatelessRealtimeSegmentWriter.java:
##########
@@ -272,6 +281,12 @@ 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",
+ _currentOffset, _endOffset);
+ if (messageBatch.getUnfilteredMessageCount() == 0) {
Review Comment:
Fixed by taking the split you suggested in the consumer thread (dbe385d190).
The writer's keep-offset change is gone from this PR, so on an empty batch the
writer adopts the next offset as it does on master, and the only writer change
left here is the loss check. The writer change that adopts validated offsets
now lives in #20024, next to the data manager change that commits segments
ending one past the last marker, so the two ship together.
_🤖 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]
