raghavyadav01 commented on code in PR #20020:
URL: https://github.com/apache/pinot/pull/20020#discussion_r4448541516
##########
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:
This changes behavior for re-ingestion. A segment whose start offset has
already aged out of retention used to finish as a partial segment, and now the
replay fails outright with no way to opt out. That may be the right call, but
the description says there's no behavior change. Can we call it out in the
description and release notes?
##########
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:
A suppressed loss only gets logged when `reportDataLoss` runs again after
the window closes, and nothing flushes it when the consume loop exits. Say a
loss is logged at t=0, a different one is found at t=30s, and the segment
commits at t=40s. The second loss never reaches the log or the segment error
cache, and the gauge is removed when the segment closes. Could we emit any
pending `_unreportedDataLossLocation` when consumption stops?
##########
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:
With this, the writer only finishes once a record at or after `endOffset`
comes in. #20024 commits segments that end one past the last marker and
switches the writer to adopting the empty batch's offset for that reason.
During a rollout, a server with only this change could re-ingest one of those
segments on an idle transactional partition and spin until the re-ingestion
timeout. Could this writer change land together with #20024, or at least ship
in the same release?
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -139,18 +151,15 @@ public synchronized KafkaMessageBatch
fetchMessages(StreamPartitionMsgOffset sta
}
lastMessageMetadata = messageMetadata;
}
- } else if (_isReadCommitted) {
- // No records returned. With read_committed isolation, the underlying
KafkaConsumer's
- // internal position may still have advanced past offsets filtered out
as aborted
- // transactional records, so snap _nextReadOffset to the consumer's
actual position.
- // RealtimeSegmentDataManager doesn't advance _currentOffset on empty
batches and will
- // call us again with the same startOffset; without this update the
seek-check would
- // re-seek to startOffset and undo the consumer's progress through the
aborted region.
- //
- // For read_uncommitted (the default), empty polls can never advance the
internal
- // position past records that didn't exist, so we skip the position()
call in the
- // hot path. The call is bounded by the same timeout as poll() so a sick
broker
- // can't stall this loop unbounded.
+ } else {
Review Comment:
In this PR nothing reads an empty batch's next offset: the data manager only
adopts it when there are unfiltered messages, and the writer now ignores it. So
this change doesn't do anything yet, and the offset it reports isn't validated
against a reset. #20024 adds that validation. Would you consider keeping this
PR to the two rate limits plus the writer's loss check, and moving the
empty-poll position change and the keep-offset logic into #20024?
--
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]
