From 16d4acb17c51314ca575cd5e9dd863b0d20c503e Mon Sep 17 00:00:00 2001 From: frank chen Date: Tue, 30 Nov 2021 21:15:52 +0800 Subject: [PATCH 1/5] Fix infinite auto reset --- ...ementalPublishingKafkaIndexTaskRunner.java | 33 +++++++++---------- 1 file changed, 15 insertions(+), 18 deletions(-) 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..cc3096da483b 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,35 @@ 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 nextFetchingOffset = 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() - ); + 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 (nextFetchingOffset < leastAvailableOffset) { + // reset offset to the least available position since it's unable to read messages from nextFetchingOffset + recordSupplier.seek(streamPartition, leastAvailableOffset); + + newOffsetInMetadata.put(topicPartition, nextFetchingOffset); } } } - if (doReset) { - sendResetRequestAndWait(CollectionUtils.mapKeys(resetPartitions, streamPartition -> StreamPartition.of( + if (!newOffsetInMetadata.isEmpty()) { + sendResetRequestAndWait(CollectionUtils.mapKeys(newOffsetInMetadata, streamPartition -> StreamPartition.of( streamPartition.topic(), streamPartition.partition() )), taskToolbox); From 86da238bcbad8573f7842e5ea2711c1322a52501 Mon Sep 17 00:00:00 2001 From: frank chen Date: Tue, 30 Nov 2021 21:23:24 +0800 Subject: [PATCH 2/5] Update doc --- docs/development/extensions-core/kafka-supervisor-reference.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/development/extensions-core/kafka-supervisor-reference.md b/docs/development/extensions-core/kafka-supervisor-reference.md index e03458529c58..c8041310096b 100644 --- a/docs/development/extensions-core/kafka-supervisor-reference.md +++ b/docs/development/extensions-core/kafka-supervisor-reference.md @@ -189,7 +189,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 least 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) | From 4701b3f5d9b554fb651997a91a37e069c1c639f0 Mon Sep 17 00:00:00 2001 From: frank chen Date: Wed, 1 Dec 2021 21:33:57 +0800 Subject: [PATCH 3/5] resolve comments --- .../kafka-supervisor-reference.md | 2 +- ...ementalPublishingKafkaIndexTaskRunner.java | 37 +++++++++++++++---- 2 files changed, 30 insertions(+), 9 deletions(-) diff --git a/docs/development/extensions-core/kafka-supervisor-reference.md b/docs/development/extensions-core/kafka-supervisor-reference.md index c8041310096b..d3d067afc2d4 100644 --- a/docs/development/extensions-core/kafka-supervisor-reference.md +++ b/docs/development/extensions-core/kafka-supervisor-reference.md @@ -189,7 +189,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 least 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) | +| `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 cc3096da483b..7f3eec4cc420 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 @@ -131,24 +131,45 @@ private void possiblyResetOffsetsOrWait( if (task.getTuningConfig().isResetOffsetAutomatically()) { for (Map.Entry outOfRangePartition : outOfRangePartitions.entrySet()) { final TopicPartition topicPartition = outOfRangePartition.getKey(); - final long nextFetchingOffset = outOfRangePartition.getValue(); + final long outOfRangeOffset = outOfRangePartition.getValue(); StreamPartition streamPartition = StreamPartition.of( topicPartition.topic(), topicPartition.partition() ); - final Long leastAvailableOffset = recordSupplier.getEarliestSequenceNumber(streamPartition); - if (leastAvailableOffset == null) { + 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()); } - if (nextFetchingOffset < leastAvailableOffset) { - // reset offset to the least available position since it's unable to read messages from nextFetchingOffset - recordSupplier.seek(streamPartition, leastAvailableOffset); - - newOffsetInMetadata.put(topicPartition, nextFetchingOffset); + if (outOfRangeOffset < earliestAvailableOffset) { + // + // In this case, it's probably because partition expires before the Druid could read from next offset + // so the messages in [outOfRangeOffset, earliestAvailableOffset) is lost. + // These lost messages could not be restored even a manual reset is performed + // So, it's reasonable to reset the offset the earliest available position + // + recordSupplier.seek(streamPartition, earliestAvailableOffset); + newOffsetInMetadata.put(topicPartition, outOfRangeOffset); + } 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 + // } } } From fd849e2e7f7505c70e2bb31c76cfe7dc60f956b5 Mon Sep 17 00:00:00 2001 From: frank chen Date: Sat, 18 Dec 2021 17:46:29 +0800 Subject: [PATCH 4/5] Add offset to alert Signed-off-by: frank chen --- .../seekablestream/SeekableStreamIndexTaskRunner.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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 65f7f4aa63d2..6ac17a22896c 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 @@ -1351,10 +1351,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(); @@ -1363,6 +1364,7 @@ protected void sendResetRequestAndWait( .addData("task", task.getId()) .addData("dataSource", task.getDataSource()) .addData("partitions", ImmutableSet.copyOf(partitionOffsetMap.keySet())) + .addData("offsets", partitionOffsetMap.values()) .emit(); } } From cc7a7048fffc8f2066f7fbf96009cf1c6b6bc081 Mon Sep 17 00:00:00 2001 From: frank chen Date: Sat, 18 Dec 2021 17:57:53 +0800 Subject: [PATCH 5/5] Resolve comments --- ...ementalPublishingKafkaIndexTaskRunner.java | 21 +++++++++++-------- 1 file changed, 12 insertions(+), 9 deletions(-) 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 7f3eec4cc420..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 @@ -145,16 +145,15 @@ private void possiblyResetOffsetsOrWait( } if (outOfRangeOffset < earliestAvailableOffset) { - // - // In this case, it's probably because partition expires before the Druid could read from next offset - // so the messages in [outOfRangeOffset, earliestAvailableOffset) is lost. - // These lost messages could not be restored even a manual reset is performed - // So, it's reasonable to reset the offset the earliest available position - // + //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, outOfRangeOffset); + + 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 @@ -169,12 +168,16 @@ private void possiblyResetOffsetsOrWait( // 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 (!newOffsetInMetadata.isEmpty()) { + log.warn("Automatcally resetting offset in metadata to [%s]", newOffsetInMetadata.toString()); + sendResetRequestAndWait(CollectionUtils.mapKeys(newOffsetInMetadata, streamPartition -> StreamPartition.of( streamPartition.topic(), streamPartition.partition()