From 0fe66adda4e0b3e21a203c6cdaf83c3ffbd5d25b Mon Sep 17 00:00:00 2001 From: swaminathanmanish <126024920+swaminathanmanish@users.noreply.github.com> Date: Wed, 5 Aug 2026 21:36:50 +0530 Subject: [PATCH 1/3] Fix false StreamDataLoss on transactional Kafka topics An offset gap between the requested startOffset and the first returned record was treated as data loss under read_uncommitted. Transactional producers write commit/abort control records that occupy offsets but are never delivered to the consumer, so a healthy contiguous stream legitimately has gaps, raising false StreamDataLoss alerts. Only flag data loss when the requested startOffset is below the broker's log start offset (beginningOffsets), i.e. records at/after startOffset were actually deleted via retention or truncation. When the log start offset cannot be determined, default to no data loss to avoid false positives. Applied to both kafka-3.0 and kafka-4.0, with unit tests covering transactional gaps, real truncation, contiguous batches, read_committed, and lookup failure. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../kafka30/KafkaPartitionLevelConsumer.java | 31 +++- ...fkaPartitionLevelConsumerDataLossTest.java | 175 ++++++++++++++++++ .../kafka40/KafkaPartitionLevelConsumer.java | 31 +++- ...fkaPartitionLevelConsumerDataLossTest.java | 175 ++++++++++++++++++ 4 files changed, 402 insertions(+), 10 deletions(-) create mode 100644 pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerDataLossTest.java create mode 100644 pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossTest.java diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java index 6e81b19d23ca..c2ec5d812f6a 100644 --- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java +++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java @@ -165,15 +165,36 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta } } long offsetOfNextBatch = _nextReadOffset; - // For read_uncommitted (the default), a non-contiguous returned batch implies data - // loss (records dropped before being read). For read_committed the offset gap is - // expected because the broker filters aborted transactional records, so we don't flag - // it as data loss. - boolean hasDataLoss = !_isReadCommitted && firstOffset > startOffset; + // A gap between the requested startOffset and the first returned offset does NOT by itself + // imply data loss. Transactional producers write commit/abort control records that occupy + // offsets but are never delivered to the consumer (even under read_uncommitted), so a + // contiguous stream of user records legitimately has offset gaps. Real data loss only + // happens when the requested startOffset is below the log's start offset, i.e. the broker + // has already deleted (via retention or truncation) records at or after startOffset. For + // read_committed we never flag loss because aborted-record gaps are always expected. + boolean hasDataLoss = false; + if (!_isReadCommitted && firstOffset > startOffset) { + hasDataLoss = getLogStartOffset(timeoutMs) > startOffset; + } return new KafkaMessageBatch(filteredRecords, records.size(), offsetOfNextBatch, firstOffset, lastMessageMetadata, hasDataLoss, batchSizeInBytes); } + /// Returns the log start (earliest available) offset for the partition, bounded by the same + /// 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) { + try { + return _consumer.beginningOffsets(List.of(_topicPartition), Duration.ofMillis(timeoutMs)) + .getOrDefault(_topicPartition, Long.MIN_VALUE); + } catch (Exception e) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Failed to read log start offset for {}", _topicPartition, e); + } + return Long.MIN_VALUE; + } + } + private static boolean isReadCommitted(KafkaPartitionLevelStreamConfig config) { String level = config.getKafkaIsolationLevel(); return level != null diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerDataLossTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerDataLossTest.java new file mode 100644 index 000000000000..bd00feb77335 --- /dev/null +++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerDataLossTest.java @@ -0,0 +1,175 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.stream.kafka30; + +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.record.TimestampType; +import org.apache.kafka.common.utils.Bytes; +import org.apache.pinot.plugin.stream.kafka.KafkaMessageBatch; +import org.apache.pinot.spi.stream.LongMsgOffset; +import org.apache.pinot.spi.stream.StreamConfig; +import org.testng.annotations.Test; + +import static org.apache.kafka.clients.consumer.ConsumerRecord.NULL_CHECKSUM; +import static org.apache.kafka.common.record.LegacyRecord.NO_TIMESTAMP; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyCollection; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; + + +/// Regression tests for [KafkaPartitionLevelConsumer] data-loss detection (DATA-2966). +/// +/// A gap between the requested `startOffset` and the first returned record offset must NOT be +/// reported as data loss when it is caused by transactional control records (commit/abort +/// markers occupy offsets but are never delivered to the consumer). Real loss is only flagged +/// when the broker's log start offset has advanced past the requested `startOffset`. +public class KafkaPartitionLevelConsumerDataLossTest { + private static final String TOPIC = "test-topic"; + private static final TopicPartition TOPIC_PARTITION = new TopicPartition(TOPIC, 0); + private static final String READ_COMMITTED = "read_committed"; + + /// Offset gap caused by transactional control records, but everything from startOffset is + /// still retained (logStartOffset <= startOffset) => no data loss. + @Test + public void testTransactionalGapWithRetainedDataIsNotDataLoss() { + Consumer mockConsumer = mock(Consumer.class); + // Requested 100, first user record is at 105 (offsets 100-104 were commit/abort markers). + when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(105L))); + when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class))) + .thenReturn(Map.of(TOPIC_PARTITION, 50L)); + + KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); + KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); + + assertFalse(batch.hasDataLoss()); + } + + /// Requested startOffset is below the log start offset: the broker deleted (retention or + /// truncation) records at/after startOffset => genuine data loss. + @Test + public void testStartOffsetBelowLogStartOffsetIsDataLoss() { + Consumer mockConsumer = mock(Consumer.class); + when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(150L))); + when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class))) + .thenReturn(Map.of(TOPIC_PARTITION, 150L)); + + KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); + KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); + + assertTrue(batch.hasDataLoss()); + } + + /// Contiguous batch (firstOffset == startOffset): no gap, so we must not even query the log + /// start offset. + @Test + public void testContiguousBatchIsNotDataLoss() { + Consumer mockConsumer = mock(Consumer.class); + when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(100L))); + + KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); + KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); + + assertFalse(batch.hasDataLoss()); + verify(mockConsumer, never()).beginningOffsets(anyCollection(), any(Duration.class)); + } + + /// Under read_committed, aborted-record gaps are always expected: never flag loss and never + /// pay for the extra broker round-trip. + @Test + public void testReadCommittedGapIsNotDataLoss() { + Consumer mockConsumer = mock(Consumer.class); + when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(105L))); + + KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(READ_COMMITTED), mockConsumer); + KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); + + assertFalse(batch.hasDataLoss()); + verify(mockConsumer, never()).beginningOffsets(anyCollection(), any(Duration.class)); + } + + /// If the log start offset cannot be determined, default to "no data loss" so a transient + /// broker hiccup does not manufacture a false StreamDataLoss alert. + @Test + public void testLogStartOffsetLookupFailureDefaultsToNoDataLoss() { + Consumer mockConsumer = mock(Consumer.class); + when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(105L))); + when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class))) + .thenThrow(new org.apache.kafka.common.errors.TimeoutException("boom")); + + KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); + KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); + + assertFalse(batch.hasDataLoss()); + } + + private static KafkaPartitionLevelConsumer createConsumerWithMock(StreamConfig streamConfig, + Consumer mockConsumer) { + class FakeKafkaPartitionLevelConsumer extends KafkaPartitionLevelConsumer { + FakeKafkaPartitionLevelConsumer(String clientId, StreamConfig streamConfig, int partition) { + super(clientId, streamConfig, partition); + } + + @Override + protected Consumer createConsumer(Properties consumerProp) { + return mockConsumer; + } + } + return new FakeKafkaPartitionLevelConsumer("clientId-test", streamConfig, 0); + } + + private static ConsumerRecords records(ConsumerRecord record) { + return new ConsumerRecords<>(Map.of(TOPIC_PARTITION, List.of(record))); + } + + private static ConsumerRecord record(long offset) { + return new ConsumerRecord<>(TOPIC, 0, offset, NO_TIMESTAMP, TimestampType.NO_TIMESTAMP_TYPE, NULL_CHECKSUM, 3, 5, + bytes("key"), bytes("value")); + } + + private static Bytes bytes(String value) { + return new Bytes(value.getBytes(StandardCharsets.UTF_8)); + } + + private static StreamConfig getStreamConfig(String isolationLevel) { + Map streamConfigMap = new HashMap<>(); + streamConfigMap.put("streamType", "kafka"); + streamConfigMap.put("stream.kafka.topic.name", TOPIC); + streamConfigMap.put("stream.kafka.broker.list", "localhost:9092"); + streamConfigMap.put("stream.kafka.consumer.factory.class.name", KafkaConsumerFactory.class.getName()); + streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass"); + if (isolationLevel != null) { + streamConfigMap.put("stream.kafka.isolation.level", isolationLevel); + } + return new StreamConfig("tableName_REALTIME", streamConfigMap); + } +} diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java index d85539436c62..7843abb4ab6e 100644 --- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java +++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java @@ -166,15 +166,36 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta } long offsetOfNextBatch = _nextReadOffset; - // For read_uncommitted (the default), a non-contiguous returned batch implies data - // loss (records dropped before being read). For read_committed the offset gap is - // expected because the broker filters aborted transactional records, so we don't flag - // it as data loss. - boolean hasDataLoss = !_isReadCommitted && firstOffset > startOffset; + // A gap between the requested startOffset and the first returned offset does NOT by itself + // imply data loss. Transactional producers write commit/abort control records that occupy + // offsets but are never delivered to the consumer (even under read_uncommitted), so a + // contiguous stream of user records legitimately has offset gaps. Real data loss only + // happens when the requested startOffset is below the log's start offset, i.e. the broker + // has already deleted (via retention or truncation) records at or after startOffset. For + // read_committed we never flag loss because aborted-record gaps are always expected. + boolean hasDataLoss = false; + if (!_isReadCommitted && firstOffset > startOffset) { + hasDataLoss = getLogStartOffset(timeoutMs) > startOffset; + } return new KafkaMessageBatch(filteredRecords, records.size(), offsetOfNextBatch, firstOffset, lastMessageMetadata, hasDataLoss, batchSizeInBytes); } + /// Returns the log start (earliest available) offset for the partition, bounded by the same + /// 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) { + try { + return _consumer.beginningOffsets(List.of(_topicPartition), Duration.ofMillis(timeoutMs)) + .getOrDefault(_topicPartition, Long.MIN_VALUE); + } catch (Exception e) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Failed to read log start offset for {}", _topicPartition, e); + } + return Long.MIN_VALUE; + } + } + private static boolean isReadCommitted(KafkaPartitionLevelStreamConfig config) { String level = config.getKafkaIsolationLevel(); return level != null diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossTest.java new file mode 100644 index 000000000000..dbe022dbd7c0 --- /dev/null +++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossTest.java @@ -0,0 +1,175 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.stream.kafka40; + +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.header.internals.RecordHeaders; +import org.apache.kafka.common.record.TimestampType; +import org.apache.kafka.common.utils.Bytes; +import org.apache.pinot.plugin.stream.kafka.KafkaMessageBatch; +import org.apache.pinot.spi.stream.LongMsgOffset; +import org.apache.pinot.spi.stream.StreamConfig; +import org.testng.annotations.Test; + +import static org.apache.kafka.common.record.RecordBatch.NO_TIMESTAMP; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyCollection; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; + + +/// Regression tests for [KafkaPartitionLevelConsumer] data-loss detection (DATA-2966). +/// +/// A gap between the requested `startOffset` and the first returned record offset must NOT be +/// reported as data loss when it is caused by transactional control records (commit/abort +/// markers occupy offsets but are never delivered to the consumer). Real loss is only flagged +/// when the broker's log start offset has advanced past the requested `startOffset`. +public class KafkaPartitionLevelConsumerDataLossTest { + private static final String TOPIC = "test-topic"; + private static final TopicPartition TOPIC_PARTITION = new TopicPartition(TOPIC, 0); + private static final String READ_COMMITTED = "read_committed"; + + /// Offset gap caused by transactional control records, but everything from startOffset is + /// still retained (logStartOffset <= startOffset) => no data loss. + @Test + public void testTransactionalGapWithRetainedDataIsNotDataLoss() { + Consumer mockConsumer = mock(Consumer.class); + // Requested 100, first user record is at 105 (offsets 100-104 were commit/abort markers). + when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(105L))); + when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class))) + .thenReturn(Map.of(TOPIC_PARTITION, 50L)); + + KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); + KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); + + assertFalse(batch.hasDataLoss()); + } + + /// Requested startOffset is below the log start offset: the broker deleted (retention or + /// truncation) records at/after startOffset => genuine data loss. + @Test + public void testStartOffsetBelowLogStartOffsetIsDataLoss() { + Consumer mockConsumer = mock(Consumer.class); + when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(150L))); + when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class))) + .thenReturn(Map.of(TOPIC_PARTITION, 150L)); + + KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); + KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); + + assertTrue(batch.hasDataLoss()); + } + + /// Contiguous batch (firstOffset == startOffset): no gap, so we must not even query the log + /// start offset. + @Test + public void testContiguousBatchIsNotDataLoss() { + Consumer mockConsumer = mock(Consumer.class); + when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(100L))); + + KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); + KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); + + assertFalse(batch.hasDataLoss()); + verify(mockConsumer, never()).beginningOffsets(anyCollection(), any(Duration.class)); + } + + /// Under read_committed, aborted-record gaps are always expected: never flag loss and never + /// pay for the extra broker round-trip. + @Test + public void testReadCommittedGapIsNotDataLoss() { + Consumer mockConsumer = mock(Consumer.class); + when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(105L))); + + KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(READ_COMMITTED), mockConsumer); + KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); + + assertFalse(batch.hasDataLoss()); + verify(mockConsumer, never()).beginningOffsets(anyCollection(), any(Duration.class)); + } + + /// If the log start offset cannot be determined, default to "no data loss" so a transient + /// broker hiccup does not manufacture a false StreamDataLoss alert. + @Test + public void testLogStartOffsetLookupFailureDefaultsToNoDataLoss() { + Consumer mockConsumer = mock(Consumer.class); + when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(105L))); + when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class))) + .thenThrow(new org.apache.kafka.common.errors.TimeoutException("boom")); + + KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); + KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); + + assertFalse(batch.hasDataLoss()); + } + + private static KafkaPartitionLevelConsumer createConsumerWithMock(StreamConfig streamConfig, + Consumer mockConsumer) { + class FakeKafkaPartitionLevelConsumer extends KafkaPartitionLevelConsumer { + FakeKafkaPartitionLevelConsumer(String clientId, StreamConfig streamConfig, int partition) { + super(clientId, streamConfig, partition); + } + + @Override + protected Consumer createConsumer(Properties consumerProp) { + return mockConsumer; + } + } + return new FakeKafkaPartitionLevelConsumer("clientId-test", streamConfig, 0); + } + + private static ConsumerRecords records(ConsumerRecord record) { + return new ConsumerRecords<>(Map.of(TOPIC_PARTITION, List.of(record))); + } + + private static ConsumerRecord record(long offset) { + return new ConsumerRecord<>(TOPIC, 0, offset, NO_TIMESTAMP, TimestampType.NO_TIMESTAMP_TYPE, 3, 5, bytes("key"), + bytes("value"), new RecordHeaders(), null); + } + + private static Bytes bytes(String value) { + return new Bytes(value.getBytes(StandardCharsets.UTF_8)); + } + + private static StreamConfig getStreamConfig(String isolationLevel) { + Map streamConfigMap = new HashMap<>(); + streamConfigMap.put("streamType", "kafka"); + streamConfigMap.put("stream.kafka.topic.name", TOPIC); + streamConfigMap.put("stream.kafka.broker.list", "localhost:9092"); + streamConfigMap.put("stream.kafka.consumer.factory.class.name", KafkaConsumerFactory.class.getName()); + streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass"); + if (isolationLevel != null) { + streamConfigMap.put("stream.kafka.isolation.level", isolationLevel); + } + return new StreamConfig("tableName_REALTIME", streamConfigMap); + } +} From 32760c8e577da8dad5026d94243819a3b313bcb8 Mon Sep 17 00:00:00 2001 From: swaminathanmanish <126024920+swaminathanmanish@users.noreply.github.com> Date: Mon, 10 Aug 2026 17:05:45 +0530 Subject: [PATCH 2/3] Replace mocked data-loss unit tests with real-broker integration tests Remove the Mockito-based KafkaPartitionLevelConsumerDataLossTest in kafka-3.0 and kafka-4.0 and add KafkaPartitionLevelConsumerDataLossIntegrationTest in each, exercising the transactional false-StreamDataLoss fix against a real broker (in-process embedded KRaft for 3.0, Testcontainers for 4.0): - transactional commit-marker gap with data retained -> no data loss - startOffset below log start (records deleted) -> data loss flagged - read_committed gap -> no data loss Co-Authored-By: Claude Opus 4.8 (1M context) --- ...nLevelConsumerDataLossIntegrationTest.java | 242 ++++++++++++++++++ ...fkaPartitionLevelConsumerDataLossTest.java | 175 ------------- ...nLevelConsumerDataLossIntegrationTest.java | 239 +++++++++++++++++ ...fkaPartitionLevelConsumerDataLossTest.java | 175 ------------- 4 files changed, 481 insertions(+), 350 deletions(-) create mode 100644 pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerDataLossIntegrationTest.java delete mode 100644 pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerDataLossTest.java create mode 100644 pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossIntegrationTest.java delete mode 100644 pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossTest.java diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerDataLossIntegrationTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerDataLossIntegrationTest.java new file mode 100644 index 000000000000..08fb6e884a76 --- /dev/null +++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerDataLossIntegrationTest.java @@ -0,0 +1,242 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.stream.kafka30; + +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.UUID; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.pinot.plugin.stream.kafka.KafkaMessageBatch; +import org.apache.pinot.plugin.stream.kafka30.server.EmbeddedKafkaCluster; +import org.apache.pinot.spi.stream.LongMsgOffset; +import org.apache.pinot.spi.stream.StreamConfig; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; + + +/// End-to-end (real embedded broker) regression tests for [KafkaPartitionLevelConsumer] data-loss +/// detection (see the fix for false `StreamDataLoss` on transactional Kafka topics). +/// +/// Unlike [KafkaPartitionLevelConsumerDataLossTest] (which mocks the Kafka consumer), these tests +/// run against an in-process [EmbeddedKafkaCluster], so they exercise the real transactional +/// control-record offset gaps and the real `beginningOffsets` round-trip added by the fix. +/// +/// Two scenarios, mirroring the reviewer's request: +/// 1. Perform a real transaction and confirm no data loss is reported for the (expected) offset gap +/// left by commit control records while the data is still retained. +/// 2. Delete offsets (advance the log start via [EmbeddedKafkaCluster#deleteRecordsBeforeOffset]) +/// and confirm data loss IS reported. +/// +/// Speed/stability: setup uses only synchronous broker calls (createTopics().all().get(), +/// commitTransaction()/flush(), deleteRecords().all().get()), so there are no fixed sleeps. Reads +/// use [#fetchUntilRecords] which polls at the same offset until data arrives, tolerating an empty +/// first poll (and the offset reset in the truncation case) without racing. +public class KafkaPartitionLevelConsumerDataLossIntegrationTest { + // Short per-poll timeout so an (occasional) empty first poll retries quickly instead of blocking; + // the happy path returns data on the first poll well within this bound. + private static final int FETCH_TIMEOUT_MS = 2000; + // Overall budget for a single logical fetch to return records (covers metadata propagation and, + // for the truncation case, the offset reset taking effect). + private static final long FETCH_MAX_WAIT_MS = 30000; + + // Transactional topic: two committed transactions of 10 records each. Under the default + // read_uncommitted isolation the commit control record after txn-1 occupies offset 10 (never + // delivered to the consumer), so txn-2's user records start at offset 11 -> a legitimate gap. + private static final String TXN_TOPIC = "txn-gap"; + private static final int RECORDS_PER_TXN = 10; + private static final long TXN1_COMMIT_MARKER_OFFSET = 10; + private static final long TXN2_FIRST_RECORD_OFFSET = 11; + + // Truncated topic: 30 contiguous records, then everything before offset 20 is deleted, so the + // log start offset advances to 20 (records at/after the requested startOffset were removed). + private static final String TRUNCATED_TOPIC = "truncated"; + private static final int TRUNCATED_TOPIC_RECORDS = 30; + private static final long TRUNCATE_BEFORE_OFFSET = 20; + + private EmbeddedKafkaCluster _kafkaCluster; + private String _kafkaBrokerAddress; + + @BeforeClass + public void setUp() + throws Exception { + Properties props = new Properties(); + props.setProperty(EmbeddedKafkaCluster.BROKER_COUNT_PROP, "1"); + _kafkaCluster = new EmbeddedKafkaCluster(); + _kafkaCluster.init(props); + _kafkaCluster.start(); + _kafkaBrokerAddress = _kafkaCluster.bootstrapServers(); + + // createTopic uses AdminClient.createTopics().all().get() -> synchronous, no sleep needed. + _kafkaCluster.createTopic(TXN_TOPIC, 1); + _kafkaCluster.createTopic(TRUNCATED_TOPIC, 1); + + // commitTransaction()/flush() are synchronous -> records are durable on return, no sleep needed. + produceTransactional(TXN_TOPIC, 2, RECORDS_PER_TXN); + producePlain(TRUNCATED_TOPIC, TRUNCATED_TOPIC_RECORDS); + + // deleteRecords().all().get() is synchronous -> log start offset advanced on return. + _kafkaCluster.deleteRecordsBeforeOffset(TRUNCATED_TOPIC, 0, TRUNCATE_BEFORE_OFFSET); + } + + @AfterClass + public void tearDown() { + try { + _kafkaCluster.deleteTopic(TXN_TOPIC); + _kafkaCluster.deleteTopic(TRUNCATED_TOPIC); + } finally { + _kafkaCluster.stop(); + } + } + + /// Scenario 1: a real committed transaction leaves an offset gap at the commit control record, + /// but all user data at/after the requested startOffset is still retained. This must NOT be + /// flagged as data loss (the pre-fix code did, raising false StreamDataLoss alerts). + @Test + public void testTransactionalGapWithRetainedDataIsNotDataLoss() + throws Exception { + // read_uncommitted (default) is the only mode where the pre-fix bug manifested. + StreamConfig streamConfig = streamConfig(TXN_TOPIC, null, null); + try (KafkaPartitionLevelConsumer consumer = + new KafkaPartitionLevelConsumer("txn-gap-client", streamConfig, 0)) { + // Seek to the commit-marker offset; the first delivered user record is txn-2's at offset 11. + KafkaMessageBatch batch = fetchUntilRecords(consumer, TXN1_COMMIT_MARKER_OFFSET); + + assertTrue(batch.getMessageCount() > 0, "Expected txn-2 records to be returned"); + // An offset gap MUST exist (first delivered offset is past the requested commit-marker offset) + // -- otherwise the test would pass without exercising the data-loss code path at all. + assertTrue(firstOffset(batch) > TXN1_COMMIT_MARKER_OFFSET, + "Expected an offset gap over the commit control record (first user record is offset " + + TXN2_FIRST_RECORD_OFFSET + ")"); + assertFalse(batch.hasDataLoss(), + "Offset gap from a transactional commit marker (data retained, logStart <= startOffset) " + + "must not be reported as data loss"); + } + } + + /// Scenario 2: records at/after the requested startOffset were deleted (log start offset advanced + /// past it). This IS genuine data loss and must be flagged. + @Test + public void testTruncatedStartOffsetIsDataLoss() + throws Exception { + // auto.offset.reset=earliest so the expired startOffset resets to the (advanced) log start. + StreamConfig streamConfig = streamConfig(TRUNCATED_TOPIC, null, "earliest"); + try (KafkaPartitionLevelConsumer consumer = + new KafkaPartitionLevelConsumer("truncated-client", streamConfig, 0)) { + // Request offset 0, which has been deleted (log start is now 20). + KafkaMessageBatch batch = fetchUntilRecords(consumer, 0); + + assertTrue(batch.getMessageCount() > 0, "Expected the retained tail of records to be returned"); + assertTrue(firstOffset(batch) >= TRUNCATE_BEFORE_OFFSET, + "First returned offset should be at/after the advanced log start"); + assertTrue(batch.hasDataLoss(), + "startOffset below the log start offset (records truncated) must be reported as data loss"); + } + } + + /// Scenario 3: under read_committed the same transactional gap must never be flagged as loss + /// (aborted/commit control gaps are always expected). This exercises the short-circuit that + /// skips the beginningOffsets round-trip entirely. + @Test + public void testReadCommittedGapIsNotDataLoss() + throws Exception { + StreamConfig streamConfig = streamConfig(TXN_TOPIC, "read_committed", null); + try (KafkaPartitionLevelConsumer consumer = + new KafkaPartitionLevelConsumer("txn-gap-rc-client", streamConfig, 0)) { + KafkaMessageBatch batch = fetchUntilRecords(consumer, TXN1_COMMIT_MARKER_OFFSET); + + assertTrue(batch.getMessageCount() > 0, "Expected txn-2 records to be returned"); + assertTrue(firstOffset(batch) > TXN1_COMMIT_MARKER_OFFSET, "Sanity: an offset gap must exist"); + assertFalse(batch.hasDataLoss(), "read_committed must never flag an offset gap as data loss"); + } + } + + /// Polls repeatedly at the same startOffset until a non-empty batch is returned (or the wait + /// budget elapses). Repeating the same startOffset hits the consumer's "no re-seek" path, so this + /// does not disturb offset positioning; it only tolerates an empty first poll while data is + /// fetched (and, for the truncation case, while the offset reset takes effect). + private KafkaMessageBatch fetchUntilRecords(KafkaPartitionLevelConsumer consumer, long startOffset) { + long deadlineMs = System.currentTimeMillis() + FETCH_MAX_WAIT_MS; + KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(startOffset), FETCH_TIMEOUT_MS); + while (batch.getMessageCount() == 0 && System.currentTimeMillis() < deadlineMs) { + batch = consumer.fetchMessages(new LongMsgOffset(startOffset), FETCH_TIMEOUT_MS); + } + return batch; + } + + private static long firstOffset(KafkaMessageBatch batch) { + return Long.parseLong(batch.getFirstMessageOffset().toString()); + } + + private StreamConfig streamConfig(String topic, String isolationLevel, String autoOffsetReset) { + Map streamConfigMap = new HashMap<>(); + streamConfigMap.put("streamType", "kafka"); + streamConfigMap.put("stream.kafka.topic.name", topic); + streamConfigMap.put("stream.kafka.broker.list", _kafkaBrokerAddress); + streamConfigMap.put("stream.kafka.consumer.factory.class.name", KafkaConsumerFactory.class.getName()); + streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass"); + if (isolationLevel != null) { + streamConfigMap.put("stream.kafka.isolation.level", isolationLevel); + } + if (autoOffsetReset != null) { + streamConfigMap.put("auto.offset.reset", autoOffsetReset); + } + return new StreamConfig("tableName_REALTIME", streamConfigMap); + } + + private void produceTransactional(String topic, int numTransactions, int recordsPerTransaction) { + Properties props = producerProps(); + props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "test-transaction-" + UUID.randomUUID()); + int seq = 0; + try (KafkaProducer producer = new KafkaProducer<>(props)) { + producer.initTransactions(); + for (int t = 0; t < numTransactions; t++) { + producer.beginTransaction(); + for (int i = 0; i < recordsPerTransaction; i++) { + producer.send(new ProducerRecord<>(topic, 0, null, "msg-" + (seq++))); + } + producer.commitTransaction(); + } + } + } + + private void producePlain(String topic, int count) { + try (KafkaProducer producer = new KafkaProducer<>(producerProps())) { + for (int i = 0; i < count; i++) { + producer.send(new ProducerRecord<>(topic, 0, null, "msg-" + i)); + } + producer.flush(); + } + } + + private Properties producerProps() { + Properties props = new Properties(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, _kafkaBrokerAddress); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + return props; + } +} diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerDataLossTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerDataLossTest.java deleted file mode 100644 index bd00feb77335..000000000000 --- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerDataLossTest.java +++ /dev/null @@ -1,175 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.pinot.plugin.stream.kafka30; - -import java.nio.charset.StandardCharsets; -import java.time.Duration; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Properties; -import org.apache.kafka.clients.consumer.Consumer; -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.common.record.TimestampType; -import org.apache.kafka.common.utils.Bytes; -import org.apache.pinot.plugin.stream.kafka.KafkaMessageBatch; -import org.apache.pinot.spi.stream.LongMsgOffset; -import org.apache.pinot.spi.stream.StreamConfig; -import org.testng.annotations.Test; - -import static org.apache.kafka.clients.consumer.ConsumerRecord.NULL_CHECKSUM; -import static org.apache.kafka.common.record.LegacyRecord.NO_TIMESTAMP; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyCollection; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; -import static org.testng.Assert.assertFalse; -import static org.testng.Assert.assertTrue; - - -/// Regression tests for [KafkaPartitionLevelConsumer] data-loss detection (DATA-2966). -/// -/// A gap between the requested `startOffset` and the first returned record offset must NOT be -/// reported as data loss when it is caused by transactional control records (commit/abort -/// markers occupy offsets but are never delivered to the consumer). Real loss is only flagged -/// when the broker's log start offset has advanced past the requested `startOffset`. -public class KafkaPartitionLevelConsumerDataLossTest { - private static final String TOPIC = "test-topic"; - private static final TopicPartition TOPIC_PARTITION = new TopicPartition(TOPIC, 0); - private static final String READ_COMMITTED = "read_committed"; - - /// Offset gap caused by transactional control records, but everything from startOffset is - /// still retained (logStartOffset <= startOffset) => no data loss. - @Test - public void testTransactionalGapWithRetainedDataIsNotDataLoss() { - Consumer mockConsumer = mock(Consumer.class); - // Requested 100, first user record is at 105 (offsets 100-104 were commit/abort markers). - when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(105L))); - when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class))) - .thenReturn(Map.of(TOPIC_PARTITION, 50L)); - - KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); - KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); - - assertFalse(batch.hasDataLoss()); - } - - /// Requested startOffset is below the log start offset: the broker deleted (retention or - /// truncation) records at/after startOffset => genuine data loss. - @Test - public void testStartOffsetBelowLogStartOffsetIsDataLoss() { - Consumer mockConsumer = mock(Consumer.class); - when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(150L))); - when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class))) - .thenReturn(Map.of(TOPIC_PARTITION, 150L)); - - KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); - KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); - - assertTrue(batch.hasDataLoss()); - } - - /// Contiguous batch (firstOffset == startOffset): no gap, so we must not even query the log - /// start offset. - @Test - public void testContiguousBatchIsNotDataLoss() { - Consumer mockConsumer = mock(Consumer.class); - when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(100L))); - - KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); - KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); - - assertFalse(batch.hasDataLoss()); - verify(mockConsumer, never()).beginningOffsets(anyCollection(), any(Duration.class)); - } - - /// Under read_committed, aborted-record gaps are always expected: never flag loss and never - /// pay for the extra broker round-trip. - @Test - public void testReadCommittedGapIsNotDataLoss() { - Consumer mockConsumer = mock(Consumer.class); - when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(105L))); - - KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(READ_COMMITTED), mockConsumer); - KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); - - assertFalse(batch.hasDataLoss()); - verify(mockConsumer, never()).beginningOffsets(anyCollection(), any(Duration.class)); - } - - /// If the log start offset cannot be determined, default to "no data loss" so a transient - /// broker hiccup does not manufacture a false StreamDataLoss alert. - @Test - public void testLogStartOffsetLookupFailureDefaultsToNoDataLoss() { - Consumer mockConsumer = mock(Consumer.class); - when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(105L))); - when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class))) - .thenThrow(new org.apache.kafka.common.errors.TimeoutException("boom")); - - KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); - KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); - - assertFalse(batch.hasDataLoss()); - } - - private static KafkaPartitionLevelConsumer createConsumerWithMock(StreamConfig streamConfig, - Consumer mockConsumer) { - class FakeKafkaPartitionLevelConsumer extends KafkaPartitionLevelConsumer { - FakeKafkaPartitionLevelConsumer(String clientId, StreamConfig streamConfig, int partition) { - super(clientId, streamConfig, partition); - } - - @Override - protected Consumer createConsumer(Properties consumerProp) { - return mockConsumer; - } - } - return new FakeKafkaPartitionLevelConsumer("clientId-test", streamConfig, 0); - } - - private static ConsumerRecords records(ConsumerRecord record) { - return new ConsumerRecords<>(Map.of(TOPIC_PARTITION, List.of(record))); - } - - private static ConsumerRecord record(long offset) { - return new ConsumerRecord<>(TOPIC, 0, offset, NO_TIMESTAMP, TimestampType.NO_TIMESTAMP_TYPE, NULL_CHECKSUM, 3, 5, - bytes("key"), bytes("value")); - } - - private static Bytes bytes(String value) { - return new Bytes(value.getBytes(StandardCharsets.UTF_8)); - } - - private static StreamConfig getStreamConfig(String isolationLevel) { - Map streamConfigMap = new HashMap<>(); - streamConfigMap.put("streamType", "kafka"); - streamConfigMap.put("stream.kafka.topic.name", TOPIC); - streamConfigMap.put("stream.kafka.broker.list", "localhost:9092"); - streamConfigMap.put("stream.kafka.consumer.factory.class.name", KafkaConsumerFactory.class.getName()); - streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass"); - if (isolationLevel != null) { - streamConfigMap.put("stream.kafka.isolation.level", isolationLevel); - } - return new StreamConfig("tableName_REALTIME", streamConfigMap); - } -} diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossIntegrationTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossIntegrationTest.java new file mode 100644 index 000000000000..4d6316f62988 --- /dev/null +++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossIntegrationTest.java @@ -0,0 +1,239 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.stream.kafka40; + +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.UUID; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.pinot.plugin.stream.kafka.KafkaMessageBatch; +import org.apache.pinot.plugin.stream.kafka40.utils.MiniKafkaCluster; +import org.apache.pinot.spi.stream.LongMsgOffset; +import org.apache.pinot.spi.stream.StreamConfig; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; + + +/// End-to-end (real broker) regression tests for [KafkaPartitionLevelConsumer] data-loss detection +/// (see the fix for false `StreamDataLoss` on transactional Kafka topics). +/// +/// Unlike [KafkaPartitionLevelConsumerDataLossTest] (which mocks the Kafka consumer), these tests +/// run against a real [MiniKafkaCluster], so they exercise the real transactional control-record +/// offset gaps and the real `beginningOffsets` round-trip added by the fix. +/// +/// Two scenarios, mirroring the reviewer's request: +/// 1. Perform a real transaction and confirm no data loss is reported for the (expected) offset gap +/// left by commit control records while the data is still retained. +/// 2. Delete offsets (advance the log start via [MiniKafkaCluster#deleteRecordsBeforeOffset]) and +/// confirm data loss IS reported. +/// +/// Speed/stability: setup uses only synchronous broker calls (createTopics().all().get(), +/// commitTransaction()/flush(), deleteRecords().all().get()), so there are no fixed sleeps. Reads +/// use [#fetchUntilRecords] which polls at the same offset until data arrives, tolerating an empty +/// first poll (and the offset reset in the truncation case) without racing. +public class KafkaPartitionLevelConsumerDataLossIntegrationTest { + // Short per-poll timeout so an (occasional) empty first poll retries quickly instead of blocking; + // the happy path returns data on the first poll well within this bound. + private static final int FETCH_TIMEOUT_MS = 2000; + // Overall budget for a single logical fetch to return records (covers metadata propagation and, + // for the truncation case, the offset reset taking effect). + private static final long FETCH_MAX_WAIT_MS = 30000; + + // Transactional topic: two committed transactions of 10 records each. Under the default + // read_uncommitted isolation the commit control record after txn-1 occupies offset 10 (never + // delivered to the consumer), so txn-2's user records start at offset 11 -> a legitimate gap. + private static final String TXN_TOPIC = "txn-gap"; + private static final int RECORDS_PER_TXN = 10; + private static final long TXN1_COMMIT_MARKER_OFFSET = 10; + private static final long TXN2_FIRST_RECORD_OFFSET = 11; + + // Truncated topic: 30 contiguous records, then everything before offset 20 is deleted, so the + // log start offset advances to 20 (records at/after the requested startOffset were removed). + private static final String TRUNCATED_TOPIC = "truncated"; + private static final int TRUNCATED_TOPIC_RECORDS = 30; + private static final long TRUNCATE_BEFORE_OFFSET = 20; + + private MiniKafkaCluster _kafkaCluster; + private String _kafkaBrokerAddress; + + @BeforeClass + public void setUp() + throws Exception { + _kafkaCluster = new MiniKafkaCluster("0"); + _kafkaCluster.start(); + _kafkaBrokerAddress = _kafkaCluster.getKafkaServerAddress(); + + // createTopic uses AdminClient.createTopics().all().get() -> synchronous, no sleep needed. + _kafkaCluster.createTopic(TXN_TOPIC, 1); + _kafkaCluster.createTopic(TRUNCATED_TOPIC, 1); + + // commitTransaction()/flush() are synchronous -> records are durable on return, no sleep needed. + produceTransactional(TXN_TOPIC, 2, RECORDS_PER_TXN); + producePlain(TRUNCATED_TOPIC, TRUNCATED_TOPIC_RECORDS); + + // deleteRecords().all().get() is synchronous -> log start offset advanced on return. + _kafkaCluster.deleteRecordsBeforeOffset(TRUNCATED_TOPIC, 0, TRUNCATE_BEFORE_OFFSET); + } + + @AfterClass + public void tearDown() { + try { + _kafkaCluster.deleteTopic(TXN_TOPIC); + _kafkaCluster.deleteTopic(TRUNCATED_TOPIC); + } finally { + _kafkaCluster.stop(); + } + } + + /// Scenario 1: a real committed transaction leaves an offset gap at the commit control record, + /// but all user data at/after the requested startOffset is still retained. This must NOT be + /// flagged as data loss (the pre-fix code did, raising false StreamDataLoss alerts). + @Test + public void testTransactionalGapWithRetainedDataIsNotDataLoss() + throws Exception { + // read_uncommitted (default) is the only mode where the pre-fix bug manifested. + StreamConfig streamConfig = streamConfig(TXN_TOPIC, null, null); + try (KafkaPartitionLevelConsumer consumer = + new KafkaPartitionLevelConsumer("txn-gap-client", streamConfig, 0)) { + // Seek to the commit-marker offset; the first delivered user record is txn-2's at offset 11. + KafkaMessageBatch batch = fetchUntilRecords(consumer, TXN1_COMMIT_MARKER_OFFSET); + + assertTrue(batch.getMessageCount() > 0, "Expected txn-2 records to be returned"); + // An offset gap MUST exist (first delivered offset is past the requested commit-marker offset) + // -- otherwise the test would pass without exercising the data-loss code path at all. + assertTrue(firstOffset(batch) > TXN1_COMMIT_MARKER_OFFSET, + "Expected an offset gap over the commit control record (first user record is offset " + + TXN2_FIRST_RECORD_OFFSET + ")"); + assertFalse(batch.hasDataLoss(), + "Offset gap from a transactional commit marker (data retained, logStart <= startOffset) " + + "must not be reported as data loss"); + } + } + + /// Scenario 2: records at/after the requested startOffset were deleted (log start offset advanced + /// past it). This IS genuine data loss and must be flagged. + @Test + public void testTruncatedStartOffsetIsDataLoss() + throws Exception { + // auto.offset.reset=earliest so the expired startOffset resets to the (advanced) log start. + StreamConfig streamConfig = streamConfig(TRUNCATED_TOPIC, null, "earliest"); + try (KafkaPartitionLevelConsumer consumer = + new KafkaPartitionLevelConsumer("truncated-client", streamConfig, 0)) { + // Request offset 0, which has been deleted (log start is now 20). + KafkaMessageBatch batch = fetchUntilRecords(consumer, 0); + + assertTrue(batch.getMessageCount() > 0, "Expected the retained tail of records to be returned"); + assertTrue(firstOffset(batch) >= TRUNCATE_BEFORE_OFFSET, + "First returned offset should be at/after the advanced log start"); + assertTrue(batch.hasDataLoss(), + "startOffset below the log start offset (records truncated) must be reported as data loss"); + } + } + + /// Scenario 3: under read_committed the same transactional gap must never be flagged as loss + /// (aborted/commit control gaps are always expected). This exercises the short-circuit that + /// skips the beginningOffsets round-trip entirely. + @Test + public void testReadCommittedGapIsNotDataLoss() + throws Exception { + StreamConfig streamConfig = streamConfig(TXN_TOPIC, "read_committed", null); + try (KafkaPartitionLevelConsumer consumer = + new KafkaPartitionLevelConsumer("txn-gap-rc-client", streamConfig, 0)) { + KafkaMessageBatch batch = fetchUntilRecords(consumer, TXN1_COMMIT_MARKER_OFFSET); + + assertTrue(batch.getMessageCount() > 0, "Expected txn-2 records to be returned"); + assertTrue(firstOffset(batch) > TXN1_COMMIT_MARKER_OFFSET, "Sanity: an offset gap must exist"); + assertFalse(batch.hasDataLoss(), "read_committed must never flag an offset gap as data loss"); + } + } + + /// Polls repeatedly at the same startOffset until a non-empty batch is returned (or the wait + /// budget elapses). Repeating the same startOffset hits the consumer's "no re-seek" path, so this + /// does not disturb offset positioning; it only tolerates an empty first poll while data is + /// fetched (and, for the truncation case, while the offset reset takes effect). + private KafkaMessageBatch fetchUntilRecords(KafkaPartitionLevelConsumer consumer, long startOffset) { + long deadlineMs = System.currentTimeMillis() + FETCH_MAX_WAIT_MS; + KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(startOffset), FETCH_TIMEOUT_MS); + while (batch.getMessageCount() == 0 && System.currentTimeMillis() < deadlineMs) { + batch = consumer.fetchMessages(new LongMsgOffset(startOffset), FETCH_TIMEOUT_MS); + } + return batch; + } + + private static long firstOffset(KafkaMessageBatch batch) { + return Long.parseLong(batch.getFirstMessageOffset().toString()); + } + + private StreamConfig streamConfig(String topic, String isolationLevel, String autoOffsetReset) { + Map streamConfigMap = new HashMap<>(); + streamConfigMap.put("streamType", "kafka"); + streamConfigMap.put("stream.kafka.topic.name", topic); + streamConfigMap.put("stream.kafka.broker.list", _kafkaBrokerAddress); + streamConfigMap.put("stream.kafka.consumer.factory.class.name", KafkaConsumerFactory.class.getName()); + streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass"); + if (isolationLevel != null) { + streamConfigMap.put("stream.kafka.isolation.level", isolationLevel); + } + if (autoOffsetReset != null) { + streamConfigMap.put("auto.offset.reset", autoOffsetReset); + } + return new StreamConfig("tableName_REALTIME", streamConfigMap); + } + + private void produceTransactional(String topic, int numTransactions, int recordsPerTransaction) { + Properties props = producerProps(); + props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "test-transaction-" + UUID.randomUUID()); + int seq = 0; + try (KafkaProducer producer = new KafkaProducer<>(props)) { + producer.initTransactions(); + for (int t = 0; t < numTransactions; t++) { + producer.beginTransaction(); + for (int i = 0; i < recordsPerTransaction; i++) { + producer.send(new ProducerRecord<>(topic, 0, null, "msg-" + (seq++))); + } + producer.commitTransaction(); + } + } + } + + private void producePlain(String topic, int count) { + try (KafkaProducer producer = new KafkaProducer<>(producerProps())) { + for (int i = 0; i < count; i++) { + producer.send(new ProducerRecord<>(topic, 0, null, "msg-" + i)); + } + producer.flush(); + } + } + + private Properties producerProps() { + Properties props = new Properties(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, _kafkaBrokerAddress); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + return props; + } +} diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossTest.java deleted file mode 100644 index dbe022dbd7c0..000000000000 --- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerDataLossTest.java +++ /dev/null @@ -1,175 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.pinot.plugin.stream.kafka40; - -import java.nio.charset.StandardCharsets; -import java.time.Duration; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Properties; -import org.apache.kafka.clients.consumer.Consumer; -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.common.header.internals.RecordHeaders; -import org.apache.kafka.common.record.TimestampType; -import org.apache.kafka.common.utils.Bytes; -import org.apache.pinot.plugin.stream.kafka.KafkaMessageBatch; -import org.apache.pinot.spi.stream.LongMsgOffset; -import org.apache.pinot.spi.stream.StreamConfig; -import org.testng.annotations.Test; - -import static org.apache.kafka.common.record.RecordBatch.NO_TIMESTAMP; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyCollection; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; -import static org.testng.Assert.assertFalse; -import static org.testng.Assert.assertTrue; - - -/// Regression tests for [KafkaPartitionLevelConsumer] data-loss detection (DATA-2966). -/// -/// A gap between the requested `startOffset` and the first returned record offset must NOT be -/// reported as data loss when it is caused by transactional control records (commit/abort -/// markers occupy offsets but are never delivered to the consumer). Real loss is only flagged -/// when the broker's log start offset has advanced past the requested `startOffset`. -public class KafkaPartitionLevelConsumerDataLossTest { - private static final String TOPIC = "test-topic"; - private static final TopicPartition TOPIC_PARTITION = new TopicPartition(TOPIC, 0); - private static final String READ_COMMITTED = "read_committed"; - - /// Offset gap caused by transactional control records, but everything from startOffset is - /// still retained (logStartOffset <= startOffset) => no data loss. - @Test - public void testTransactionalGapWithRetainedDataIsNotDataLoss() { - Consumer mockConsumer = mock(Consumer.class); - // Requested 100, first user record is at 105 (offsets 100-104 were commit/abort markers). - when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(105L))); - when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class))) - .thenReturn(Map.of(TOPIC_PARTITION, 50L)); - - KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); - KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); - - assertFalse(batch.hasDataLoss()); - } - - /// Requested startOffset is below the log start offset: the broker deleted (retention or - /// truncation) records at/after startOffset => genuine data loss. - @Test - public void testStartOffsetBelowLogStartOffsetIsDataLoss() { - Consumer mockConsumer = mock(Consumer.class); - when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(150L))); - when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class))) - .thenReturn(Map.of(TOPIC_PARTITION, 150L)); - - KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); - KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); - - assertTrue(batch.hasDataLoss()); - } - - /// Contiguous batch (firstOffset == startOffset): no gap, so we must not even query the log - /// start offset. - @Test - public void testContiguousBatchIsNotDataLoss() { - Consumer mockConsumer = mock(Consumer.class); - when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(100L))); - - KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); - KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); - - assertFalse(batch.hasDataLoss()); - verify(mockConsumer, never()).beginningOffsets(anyCollection(), any(Duration.class)); - } - - /// Under read_committed, aborted-record gaps are always expected: never flag loss and never - /// pay for the extra broker round-trip. - @Test - public void testReadCommittedGapIsNotDataLoss() { - Consumer mockConsumer = mock(Consumer.class); - when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(105L))); - - KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(READ_COMMITTED), mockConsumer); - KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); - - assertFalse(batch.hasDataLoss()); - verify(mockConsumer, never()).beginningOffsets(anyCollection(), any(Duration.class)); - } - - /// If the log start offset cannot be determined, default to "no data loss" so a transient - /// broker hiccup does not manufacture a false StreamDataLoss alert. - @Test - public void testLogStartOffsetLookupFailureDefaultsToNoDataLoss() { - Consumer mockConsumer = mock(Consumer.class); - when(mockConsumer.poll(any(Duration.class))).thenReturn(records(record(105L))); - when(mockConsumer.beginningOffsets(anyCollection(), any(Duration.class))) - .thenThrow(new org.apache.kafka.common.errors.TimeoutException("boom")); - - KafkaPartitionLevelConsumer consumer = createConsumerWithMock(getStreamConfig(null), mockConsumer); - KafkaMessageBatch batch = consumer.fetchMessages(new LongMsgOffset(100L), 10000); - - assertFalse(batch.hasDataLoss()); - } - - private static KafkaPartitionLevelConsumer createConsumerWithMock(StreamConfig streamConfig, - Consumer mockConsumer) { - class FakeKafkaPartitionLevelConsumer extends KafkaPartitionLevelConsumer { - FakeKafkaPartitionLevelConsumer(String clientId, StreamConfig streamConfig, int partition) { - super(clientId, streamConfig, partition); - } - - @Override - protected Consumer createConsumer(Properties consumerProp) { - return mockConsumer; - } - } - return new FakeKafkaPartitionLevelConsumer("clientId-test", streamConfig, 0); - } - - private static ConsumerRecords records(ConsumerRecord record) { - return new ConsumerRecords<>(Map.of(TOPIC_PARTITION, List.of(record))); - } - - private static ConsumerRecord record(long offset) { - return new ConsumerRecord<>(TOPIC, 0, offset, NO_TIMESTAMP, TimestampType.NO_TIMESTAMP_TYPE, 3, 5, bytes("key"), - bytes("value"), new RecordHeaders(), null); - } - - private static Bytes bytes(String value) { - return new Bytes(value.getBytes(StandardCharsets.UTF_8)); - } - - private static StreamConfig getStreamConfig(String isolationLevel) { - Map streamConfigMap = new HashMap<>(); - streamConfigMap.put("streamType", "kafka"); - streamConfigMap.put("stream.kafka.topic.name", TOPIC); - streamConfigMap.put("stream.kafka.broker.list", "localhost:9092"); - streamConfigMap.put("stream.kafka.consumer.factory.class.name", KafkaConsumerFactory.class.getName()); - streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass"); - if (isolationLevel != null) { - streamConfigMap.put("stream.kafka.isolation.level", isolationLevel); - } - return new StreamConfig("tableName_REALTIME", streamConfigMap); - } -} From c27cfad8a3e87f9b6499aac7ab0b96d2682a3912 Mon Sep 17 00:00:00 2001 From: swaminathanmanish <126024920+swaminathanmanish@users.noreply.github.com> Date: Mon, 10 Aug 2026 17:45:17 +0530 Subject: [PATCH 3/3] Log a WARN when the log-start-offset lookup fails during gap check Address review feedback: previously the beginningOffsets failure in getLogStartOffset was logged only at DEBUG, so a failure that defaults the gap to no-data-loss was effectively silent in production and could mask a genuine loss. Log it at WARN instead. Applied to kafka-3.0 and kafka-4.0. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../plugin/stream/kafka30/KafkaPartitionLevelConsumer.java | 5 ++--- .../plugin/stream/kafka40/KafkaPartitionLevelConsumer.java | 5 ++--- 2 files changed, 4 insertions(+), 6 deletions(-) diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java index c2ec5d812f6a..538a191d1354 100644 --- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java +++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java @@ -188,9 +188,8 @@ private long getLogStartOffset(int timeoutMs) { return _consumer.beginningOffsets(List.of(_topicPartition), Duration.ofMillis(timeoutMs)) .getOrDefault(_topicPartition, Long.MIN_VALUE); } catch (Exception e) { - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("Failed to read log start offset for {}", _topicPartition, 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); return Long.MIN_VALUE; } } diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java index 7843abb4ab6e..efd5f4daf529 100644 --- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java +++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java @@ -189,9 +189,8 @@ private long getLogStartOffset(int timeoutMs) { return _consumer.beginningOffsets(List.of(_topicPartition), Duration.ofMillis(timeoutMs)) .getOrDefault(_topicPartition, Long.MIN_VALUE); } catch (Exception e) { - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("Failed to read log start offset for {}", _topicPartition, 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); return Long.MIN_VALUE; } }