diff --git a/docs/development/extensions-core/kafka-supervisor-reference.md b/docs/development/extensions-core/kafka-supervisor-reference.md
index a59177aa319b..9c615ff58398 100644
--- a/docs/development/extensions-core/kafka-supervisor-reference.md
+++ b/docs/development/extensions-core/kafka-supervisor-reference.md
@@ -190,7 +190,7 @@ The `tuningConfig` is optional and default parameters will be used if no `tuning
| `indexSpecForIntermediatePersists`| | Defines segment storage format options to be used at indexing time for intermediate persisted temporary segments. This can be used to disable dimension/metric compression on intermediate segments to reduce memory required for final merging. However, disabling compression on intermediate segments might increase page cache use while they are used before getting merged into final segment published, see [IndexSpec](#indexspec) for possible values. | no (default = same as `indexSpec`) |
| `reportParseExceptions` | Boolean | *DEPRECATED*. If true, exceptions encountered during parsing will be thrown and will halt ingestion; if false, unparseable rows and fields will be skipped. Setting `reportParseExceptions` to true will override existing configurations for `maxParseExceptions` and `maxSavedParseExceptions`, setting `maxParseExceptions` to 0 and limiting `maxSavedParseExceptions` to no more than 1. | no (default == false) |
| `handoffConditionTimeout` | Long | Milliseconds to wait for segment handoff. It must be >= 0, where 0 means to wait forever. | no (default == 0) |
-| `resetOffsetAutomatically` | Boolean | Controls behavior when Druid needs to read Kafka messages that are no longer available (i.e. when `OffsetOutOfRangeException` is encountered).
If false, the exception will bubble up, which will cause your tasks to fail and ingestion to halt. If this occurs, manual intervention is required to correct the situation; potentially using the [Reset Supervisor API](../../operations/api-reference.md#supervisors). This mode is useful for production, since it will make you aware of issues with ingestion.
If true, Druid will automatically reset to the earlier or latest offset available in Kafka, based on the value of the `useEarliestOffset` property (earliest if true, latest if false). Note that this can lead to data being _DROPPED_ (if `useEarliestOffset` is false) or _DUPLICATED_ (if `useEarliestOffset` is true) without your knowledge. Messages will be logged indicating that a reset has occurred, but ingestion will continue. This mode is useful for non-production situations, since it will make Druid attempt to recover from problems automatically, even if they lead to quiet dropping or duplicating of data.
This feature behaves similarly to the Kafka `auto.offset.reset` consumer property. | no (default == false) |
+| `resetOffsetAutomatically` | Boolean | Controls behavior when Druid needs to read Kafka messages that are no longer available (i.e. when `OffsetOutOfRangeException` is encountered).
If false, the exception will bubble up, which will cause your tasks to fail and ingestion to halt. If this occurs, manual intervention is required to correct the situation; potentially using the [Reset Supervisor API](../../operations/api-reference.md#supervisors). This mode is useful for production, since it will make you aware of issues with ingestion.
If true, Druid will automatically reset to the earliest offset available in Kafka. Note that this can lead to data being _DROPPED_ without your knowledge. Messages will be logged indicating that a reset has occurred, but ingestion will continue. This mode is useful for non-production situations, since it will make Druid attempt to recover from problems automatically, even if they lead to quiet dropping.
This feature behaves similarly to the Kafka `auto.offset.reset` consumer property. | no (default == false) |
| `workerThreads` | Integer | The number of threads that the supervisor uses to handle requests/responses for worker tasks, along with any other internal asynchronous operation. | no (default == min(10, taskCount)) |
| `chatThreads` | Integer | The number of threads that will be used for communicating with indexing tasks. | no (default == min(10, taskCount * replicas)) |
| `chatRetries` | Integer | The number of times HTTP requests to indexing tasks will be retried before considering tasks unresponsive. | no (default == 8) |
diff --git a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/IncrementalPublishingKafkaIndexTaskRunner.java b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/IncrementalPublishingKafkaIndexTaskRunner.java
index 662b8b03bec4..0775b78de7eb 100644
--- a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/IncrementalPublishingKafkaIndexTaskRunner.java
+++ b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/IncrementalPublishingKafkaIndexTaskRunner.java
@@ -126,38 +126,59 @@ private void possiblyResetOffsetsOrWait(
TaskToolbox taskToolbox
) throws InterruptedException, IOException
{
- final Map resetPartitions = new HashMap<>();
- boolean doReset = false;
+ final Map newOffsetInMetadata = new HashMap<>();
+
if (task.getTuningConfig().isResetOffsetAutomatically()) {
for (Map.Entry outOfRangePartition : outOfRangePartitions.entrySet()) {
final TopicPartition topicPartition = outOfRangePartition.getKey();
- final long nextOffset = outOfRangePartition.getValue();
- // seek to the beginning to get the least available offset
+ final long outOfRangeOffset = outOfRangePartition.getValue();
+
StreamPartition streamPartition = StreamPartition.of(
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 earliestAvailableOffset = recordSupplier.getEarliestSequenceNumber(streamPartition);
+ if (earliestAvailableOffset == null) {
+ throw new ISE("got null earliest 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);
+
+ if (outOfRangeOffset < earliestAvailableOffset) {
+ //In this case, it's probably because the messages are no longer in the Kafka cluster
+ // i.e. the messages in [outOfRangeOffset, earliestAvailableOffset) are lost.
+ // Since these lost messages can no longer be recovered,
+ // it's reasonable to reset the offset to the earliest available position to help ingestion resume.
+ log.warn("Automatically seeking Kafka offset to the earliest offset [%d]", earliestAvailableOffset);
+ recordSupplier.seek(streamPartition, earliestAvailableOffset);
+
+ newOffsetInMetadata.put(topicPartition, earliestAvailableOffset);
+ } else {
+ // There are two cases in theory here
+ // 1. outOfRangeOffset is in the range of [earliestAvailableOffset, latestAvailableOffset]
+ // 2. outOfRangeOffset is larger than latestAvailableOffset
+ //
+ // for scenario 1, we do nothing but just wait for a period time to retry
+ // since current offset is valid but maybe due to some temporary problem
+ //
+ // for scenario 2, how could this happen?
+ // Well, if the task first consumes from a topic on cluster A,
+ // and then supervisor spec is changed to consume from a same topic, where there are messages in this topic, on cluster B,
+ // this can lead to this case.
+ // For such case,
+ // offsets stored in meta should be cleared when submitting the supervisor spec,
+ // so the problem won't be left to manual reset or auto reset. Thus, we don't need to handle this complicated case here
+ log.warn("Offset [%d] is out of range of the available offsets for partition [%s]. It is likely that a manual offset reset of the supervisor is needed.",
+ outOfRangeOffset,
+ topicPartition);
}
}
}
- if (doReset) {
- sendResetRequestAndWait(CollectionUtils.mapKeys(resetPartitions, streamPartition -> StreamPartition.of(
+ if (!newOffsetInMetadata.isEmpty()) {
+ log.warn("Automatcally resetting offset in metadata to [%s]", newOffsetInMetadata.toString());
+
+ sendResetRequestAndWait(CollectionUtils.mapKeys(newOffsetInMetadata, streamPartition -> StreamPartition.of(
streamPartition.topic(),
streamPartition.partition()
)), taskToolbox);
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 9b2eff3f6ffd..16060172f9b4 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
@@ -1352,10 +1352,11 @@ protected void sendResetRequestAndWait(
);
if (result) {
- log.makeAlert("Offsets were reset automatically, potential data duplication or loss")
+ log.makeAlert("Offsets were reset automatically, potential data loss")
.addData("task", task.getId())
.addData("dataSource", task.getDataSource())
.addData("partitions", partitionOffsetMap.keySet())
+ .addData("offsets", partitionOffsetMap.values())
.emit();
requestPause();
@@ -1364,6 +1365,7 @@ protected void sendResetRequestAndWait(
.addData("task", task.getId())
.addData("dataSource", task.getDataSource())
.addData("partitions", ImmutableSet.copyOf(partitionOffsetMap.keySet()))
+ .addData("offsets", partitionOffsetMap.values())
.emit();
}
}