This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new ea46137748c fix: Fail ingestion tasks when OffsetOutOfRangeException
is raised (#18226)
ea46137748c is described below
commit ea46137748c061c779533359a26bd102b6726f70
Author: Frank Chen <[email protected]>
AuthorDate: Tue Nov 3 10:12:29 2026 +0800
fix: Fail ingestion tasks when OffsetOutOfRangeException is raised (#18226)
* Handle Kafka OffsetOutOfRangeException
* Update doc
* Fail Kafka task only when offsets are below the earliest available offset
Only fail the task when the out-of-range offset has been removed from the
partition; keep waiting for offsets that have not been written yet.
Publish the data read so far before failing, and replace TaskStatusException
with UnrecoverableStreamReadException.
---
.../druid/indexing/kafka/KafkaIndexTaskRunner.java | 86 ++++++++------
.../druid/indexing/kafka/KafkaIndexTaskTest.java | 127 +++++++++++++++++++++
.../SeekableStreamIndexTaskRunner.java | 36 ++++--
.../UnrecoverableStreamReadException.java | 37 ++++++
4 files changed, 247 insertions(+), 39 deletions(-)
diff --git
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java
index 672309a9229..f56f49461f6 100644
---
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java
+++
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskRunner.java
@@ -31,12 +31,14 @@ import
org.apache.druid.indexing.seekablestream.SeekableStreamEndSequenceNumbers
import org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner;
import org.apache.druid.indexing.seekablestream.SeekableStreamSequenceNumbers;
import org.apache.druid.indexing.seekablestream.SequenceMetadata;
+import
org.apache.druid.indexing.seekablestream.UnrecoverableStreamReadException;
import
org.apache.druid.indexing.seekablestream.common.OrderedPartitionableRecord;
import org.apache.druid.indexing.seekablestream.common.OrderedSequenceNumber;
import org.apache.druid.indexing.seekablestream.common.RecordSupplier;
import org.apache.druid.indexing.seekablestream.common.StreamPartition;
import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.utils.CollectionUtils;
import org.apache.kafka.clients.consumer.OffsetOutOfRangeException;
@@ -88,9 +90,10 @@ public class KafkaIndexTaskRunner extends
SeekableStreamIndexTaskRunner<KafkaTop
}
catch (OffsetOutOfRangeException e) {
//
- // Handles OffsetOutOfRangeException, which is thrown if the seeked-to
- // offset is not present in the topic-partition. This can happen if
we're asking a task to read from data
- // that has not been written yet (which is totally legitimate). So let's
wait for it to show up
+ // Handles OffsetOutOfRangeException, which is thrown if the seeked-to
offset is not present in the
+ // topic-partition. This happens either when the offset has been removed
from the partition, e.g. due to the
+ // retention configuration, or when we're asking a task to read from
data that has not been written yet (which
+ // is totally legitimate). See possiblyResetOffsetsOrWait for how each
case is handled.
//
log.warn("OffsetOutOfRangeException with message [%s]", e.getMessage());
possiblyResetOffsetsOrWait(e.offsetOutOfRangePartitions(),
recordSupplier, toolbox);
@@ -112,46 +115,65 @@ public class KafkaIndexTaskRunner extends
SeekableStreamIndexTaskRunner<KafkaTop
));
}
+ /**
+ * Handles partitions whose next offset is out of range:
+ * <ul>
+ * <li>If the next offset is below the earliest available offset of the
partition, the data has been removed from
+ * the partition. If {@code resetOffsetAutomatically} is set, ask the
supervisor to reset the offsets. Otherwise,
+ * throw {@link UnrecoverableStreamReadException} so that the task
publishes what it has read and fails, instead
+ * of retrying an offset that will never become available again.</li>
+ * <li>Otherwise, the next offset has not been written yet, so wait for it
to show up.</li>
+ * </ul>
+ */
private void possiblyResetOffsetsOrWait(
Map<TopicPartition, Long> outOfRangePartitions,
RecordSupplier<KafkaTopicPartition, Long, KafkaRecordEntity>
recordSupplier,
TaskToolbox taskToolbox
- ) throws InterruptedException, IOException
+ ) throws InterruptedException, IOException, UnrecoverableStreamReadException
{
final String stream =
task.getIOConfig().getStartSequenceNumbers().getStream();
final boolean isMultiTopic = task.getIOConfig().isMultiTopic();
- final Map<TopicPartition, Long> resetPartitions = new HashMap<>();
- boolean doReset = false;
- if (task.getTuningConfig().isResetOffsetAutomatically()) {
- for (Map.Entry<TopicPartition, Long> outOfRangePartition :
outOfRangePartitions.entrySet()) {
- final TopicPartition topicPartition = outOfRangePartition.getKey();
- final long nextOffset = outOfRangePartition.getValue();
- // seek to the beginning to get the least available offset
- StreamPartition<KafkaTopicPartition> streamPartition =
StreamPartition.of(
- stream,
- new KafkaTopicPartition(isMultiTopic, topicPartition.topic(),
topicPartition.partition())
+ final Map<TopicPartition, Long> unavailablePartitions = new HashMap<>();
+ final Map<TopicPartition, Long> earliestOffsets = new HashMap<>();
+ for (Map.Entry<TopicPartition, Long> outOfRangePartition :
outOfRangePartitions.entrySet()) {
+ final TopicPartition topicPartition = outOfRangePartition.getKey();
+ final long nextOffset = outOfRangePartition.getValue();
+ // seek to the beginning to get the least available offset
+ final StreamPartition<KafkaTopicPartition> streamPartition =
StreamPartition.of(
+ stream,
+ new KafkaTopicPartition(isMultiTopic, topicPartition.topic(),
topicPartition.partition())
+ );
+ final Long leastAvailableOffset =
recordSupplier.getEarliestSequenceNumber(streamPartition);
+ if (leastAvailableOffset == null) {
+ throw new ISE(
+ "got null sequence number for partition[%s] when fetching from
kafka!",
+ topicPartition.partition()
);
- final Long leastAvailableOffset =
recordSupplier.getEarliestSequenceNumber(streamPartition);
- if (leastAvailableOffset == null) {
- throw new ISE(
- "got null sequence number for partition[%s] when fetching from
kafka!",
- topicPartition.partition()
- );
- }
- // reset the seek
- recordSupplier.seek(streamPartition, nextOffset);
- // Reset consumer offset if resetOffsetAutomatically is set to true
- // and the current message offset in the kafka partition is more than
the
- // next message offset that we are trying to fetch
- if (leastAvailableOffset > nextOffset) {
- doReset = true;
- resetPartitions.put(topicPartition, nextOffset);
- }
}
+ // reset the seek
+ recordSupplier.seek(streamPartition, nextOffset);
+ // The next message offset that we are trying to fetch is no longer
available
+ if (leastAvailableOffset > nextOffset) {
+ unavailablePartitions.put(topicPartition, nextOffset);
+ earliestOffsets.put(topicPartition, leastAvailableOffset);
+ }
+ }
+
+ if (!unavailablePartitions.isEmpty() &&
!task.getTuningConfig().isResetOffsetAutomatically()) {
+ throw new UnrecoverableStreamReadException(
+ StringUtils.format(
+ "Offsets[%s] are no longer available in the stream, the earliest
available offsets are [%s]. "
+ + "This may happen when the offsets have been deleted by the
retention configuration of the topic. "
+ + "Use the supervisor's reset API to resume reading from a valid
offset, "
+ + "or set 'resetOffsetAutomatically' to true in the supervisor's
tuningConfig.",
+ unavailablePartitions,
+ earliestOffsets
+ )
+ );
}
- if (doReset) {
- sendResetRequestAndWait(CollectionUtils.mapKeys(resetPartitions,
topicPartition -> StreamPartition.of(
+ if (!unavailablePartitions.isEmpty()) {
+ sendResetRequestAndWait(CollectionUtils.mapKeys(unavailablePartitions,
topicPartition -> StreamPartition.of(
stream,
new KafkaTopicPartition(isMultiTopic, topicPartition.topic(),
topicPartition.partition())
)), taskToolbox);
diff --git
a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
index be1082629a0..502ab4092e0 100644
---
a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
+++
b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
@@ -139,8 +139,12 @@ import org.apache.druid.server.security.Resource;
import org.apache.druid.server.security.ResourceAction;
import org.apache.druid.server.security.ResourceType;
import org.apache.druid.timeline.DataSegment;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.RecordsToDelete;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Header;
import org.joda.time.Duration;
import org.joda.time.Period;
@@ -2461,6 +2465,109 @@ public class KafkaIndexTaskTest extends
SeekableStreamIndexTaskTestBase
}
}
+ @Test
+ @Timeout(value = 60_000L, unit = TimeUnit.MILLISECONDS)
+ public void
testRunWithOffsetOutOfRangeExceptionBelowEarliestOffsetFailsTask() throws
Exception
+ {
+ insertData();
+ // Remove the records the task is going to read, as the retention of the
topic would do
+ deleteRecords(0, 5L);
+
+ final KafkaIndexTask task = createTask(
+ null,
+ new KafkaIndexTaskIOConfig(
+ 0,
+ "sequence0",
+ new SeekableStreamStartSequenceNumbers<>(topic,
ImmutableMap.of(new KafkaTopicPartition(false, topic, 0), 2L),
ImmutableSet.of()),
+ new SeekableStreamEndSequenceNumbers<>(topic, ImmutableMap.of(new
KafkaTopicPartition(false, topic, 0), 10L)),
+ kafkaServer.consumerProperties(),
+ KafkaSupervisorIOConfig.DEFAULT_POLL_TIMEOUT_MILLIS,
+ true,
+ null,
+ null,
+ INPUT_FORMAT,
+ null,
+ Duration.standardHours(2).getStandardMinutes(),
+ null
+ )
+ );
+
+ final TaskStatus status = runTask(task).get();
+
+ Assertions.assertEquals(TaskState.FAILED, status.getStatusCode());
+ Assertions.assertTrue(
+ status.getErrorMsg().contains("are no longer available in the stream"),
+ status.getErrorMsg()
+ );
+ Assertions.assertEquals(ImmutableList.of(), publishedDescriptors());
+ Assertions.assertNull(newDataSchemaMetadata());
+
+ final IngestionStatsAndErrors reportData = getTaskReportData();
+ Assertions.assertEquals(IngestionState.COMPLETED,
reportData.getIngestionState());
+ Assertions.assertEquals(status.getErrorMsg(), reportData.getErrorMsg());
+ }
+
+ @Test
+ @Timeout(value = 60_000L, unit = TimeUnit.MILLISECONDS)
+ public void
testRunWithOffsetOutOfRangeExceptionBelowEarliestOffsetPublishesReadData()
throws Exception
+ {
+ insertData();
+
+ final KafkaIndexTask task = createTask(
+ null,
+ new KafkaIndexTaskIOConfig(
+ 0,
+ "sequence0",
+ new SeekableStreamStartSequenceNumbers<>(topic,
ImmutableMap.of(new KafkaTopicPartition(false, topic, 0), 2L),
ImmutableSet.of()),
+ new SeekableStreamEndSequenceNumbers<>(topic, ImmutableMap.of(new
KafkaTopicPartition(false, topic, 0), 100L)),
+ kafkaServer.consumerProperties(),
+ KafkaSupervisorIOConfig.DEFAULT_POLL_TIMEOUT_MILLIS,
+ true,
+ null,
+ null,
+ INPUT_FORMAT,
+ null,
+ Duration.standardHours(2).getStandardMinutes(),
+ null
+ )
+ );
+
+ final ListenableFuture<TaskStatus> future = runTask(task);
+
+ final KafkaTopicPartition partition = new KafkaTopicPartition(false,
topic, 0);
+ while
(!Long.valueOf(13L).equals(task.getRunner().getCurrentOffsets().get(partition)))
{
+ Thread.sleep(25);
+ }
+
+ // While the task is paused, write more records and remove all of them, so
that the next offset the task reads
+ // is below the earliest available offset when it resumes.
+ task.getRunner().pause();
+ Assertions.assertEquals(Status.PAUSED, task.getRunner().getStatus());
+ // Let the fetch the consumer may have sent before pausing complete, so it
does not pick up the records below
+ Thread.sleep(1000);
+ insertData();
+ deleteRecords(0, getLatestOffset(0));
+ task.getRunner().resume();
+
+ final TaskStatus status = future.get();
+
+ Assertions.assertEquals(TaskState.FAILED, status.getStatusCode());
+ Assertions.assertTrue(
+ status.getErrorMsg().contains("are no longer available in the stream"),
+ status.getErrorMsg()
+ );
+
+ // The records read before the offsets became unavailable are published
+ assertEqualsExceptVersion(
+ ImmutableList.of(sdd("2010/P1D", 0), sdd("2011/P1D", 0),
sdd("2013/P1D", 0), sdd("2049/P1D", 0)),
+ publishedDescriptors()
+ );
+ Assertions.assertEquals(
+ new KafkaDataSourceMetadata(new
SeekableStreamEndSequenceNumbers<>(topic, ImmutableMap.of(partition, 13L))),
+ newDataSchemaMetadata()
+ );
+ }
+
@Test
@Timeout(value = 60_000L, unit = TimeUnit.MILLISECONDS)
public void testRunContextSequenceAheadOfStartingOffsets() throws Exception
@@ -2911,6 +3018,26 @@ public class KafkaIndexTaskTest extends
SeekableStreamIndexTaskTestBase
.sum();
}
+ private void deleteRecords(int partition, long beforeOffset) throws
ExecutionException, InterruptedException
+ {
+ try (Admin admin = kafkaServer.newAdminClient()) {
+ admin.deleteRecords(
+ Map.of(new TopicPartition(topic, partition),
RecordsToDelete.beforeOffset(beforeOffset))
+ ).all().get();
+ }
+ }
+
+ private long getLatestOffset(int partition) throws ExecutionException,
InterruptedException
+ {
+ final TopicPartition topicPartition = new TopicPartition(topic, partition);
+ try (Admin admin = kafkaServer.newAdminClient()) {
+ return admin.listOffsets(Map.of(topicPartition, OffsetSpec.latest()))
+ .partitionResult(topicPartition)
+ .get()
+ .offset();
+ }
+ }
+
private void insertData() throws ExecutionException, InterruptedException
{
insertData(records);
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
index 910940dd2f8..4696e36515f 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
@@ -479,6 +479,10 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
Throwable caughtException = null;
+ // Set when getRecords() reports that no further progress can be made. The
data read so far is still published,
+ // but the task fails with the message of this exception.
+ UnrecoverableStreamReadException unrecoverableReadException = null;
+
//milliseconds waited for created segments to be handed off
long handoffWaitMs = 0L;
@@ -688,13 +692,24 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
// calling getRecord() ensures that exceptions specific to
kafka/kinesis like OffsetOutOfRangeException
// are handled in the subclasses.
- List<OrderedPartitionableRecord<PartitionIdType, SequenceOffsetType,
RecordType>> records = getRecords(
- recordSupplier,
- toolbox
- );
+ List<OrderedPartitionableRecord<PartitionIdType, SequenceOffsetType,
RecordType>> records;
+ try {
+ records = getRecords(recordSupplier, toolbox);
- // note: getRecords() also updates assignment
- stillReading = !assignment.isEmpty();
+ // note: getRecords() also updates assignment
+ stillReading = !assignment.isEmpty();
+ }
+ catch (UnrecoverableStreamReadException e) {
+ log.error(
+ "Cannot continue reading the stream: %s Stopping reading. Data
read up to offsets[%s] will be "
+ + "persisted and published, then the task will exit with
FAILED status.",
+ e.getMessage(),
+ currOffsets
+ );
+ unrecoverableReadException = e;
+ records = Collections.emptyList();
+ stillReading = false;
+ }
SequenceMetadata<PartitionIdType, SequenceOffsetType>
sequenceToCheckpoint = null;
AppenderatorDriverAddResult pushTriggeringAddResult = null;
@@ -1015,6 +1030,11 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
}
ingestionState = IngestionState.COMPLETED;
+ if (unrecoverableReadException != null) {
+ final String errorMsg = unrecoverableReadException.getMessage();
+ toolbox.getTaskReportFileWriter().write(task.getId(),
getTaskCompletionReports(errorMsg, handoffWaitMs));
+ return TaskStatus.failure(task.getId(), errorMsg);
+ }
toolbox.getTaskReportFileWriter().write(task.getId(),
getTaskCompletionReports(null, handoffWaitMs));
return TaskStatus.success(task.getId());
}
@@ -2374,7 +2394,9 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
* @param recordSupplier
* @param toolbox
* @return list of records polled, can be empty but cannot be null
- * @throws Exception
+ * @throws UnrecoverableStreamReadException if no further progress can be
made reading the stream. The records read
+ * so far are published and then
the task fails.
+ * @throws Exception for other errors, in which case
the task fails without publishing
*/
@NotNull
protected abstract List<OrderedPartitionableRecord<PartitionIdType,
SequenceOffsetType, RecordType>> getRecords(
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/UnrecoverableStreamReadException.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/UnrecoverableStreamReadException.java
new file mode 100644
index 00000000000..e74db5bb68a
--- /dev/null
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/UnrecoverableStreamReadException.java
@@ -0,0 +1,37 @@
+/*
+ * 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.druid.indexing.seekablestream;
+
+/**
+ * Thrown by {@link SeekableStreamIndexTaskRunner#getRecords} when the task
cannot make any further progress reading
+ * the stream, for example because the offsets it must read next have been
removed from the stream.
+ *
+ * Unlike other exceptions thrown from the read loop, the runner stops
reading, publishes the data read so far and
+ * then fails the task with the message of this exception. Failing the task
makes the supervisor re-read the start
+ * offsets from the metadata store, so the problem shows up in the supervisor
status instead of the task retrying
+ * silently until its duration elapses.
+ */
+public class UnrecoverableStreamReadException extends Exception
+{
+ public UnrecoverableStreamReadException(String message)
+ {
+ super(message);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
