Skip to content

Fixes a bug in Kafka auto reset #12008

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
FrankChen021 wants to merge 6 commits into from
Closed

Fixes a bug in Kafka auto reset #12008

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
16d4acb
Fix infinite auto reset
FrankChen021 Nov 30, 2021
86da238
Update doc
FrankChen021 Nov 30, 2021
4701b3f
resolve comments
FrankChen021 Dec 1, 2021
fd849e2
Add offset to alert
FrankChen021 Dec 18, 2021
cc7a704
Resolve comments
FrankChen021 Dec 18, 2021
2016807
Merge branch 'master' into auto-reset-bug
FrankChen021 Dec 19, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/development/extensions-core/kafka-supervisor-reference.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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).<br/><br/>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.<br/><br/>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.<br/><br/>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).<br/><br/>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.<br/><br/>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.<br/><br/>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) |
Expand Down
61 changes: 41 additions & 20 deletions .../main/java/org/apache/druid/indexing/kafka/IncrementalPublishingKafkaIndexTaskRunner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,38 +126,59 @@ private void possiblyResetOffsetsOrWait(
TaskToolbox taskToolbox
) throws InterruptedException, IOException
{
final Map<TopicPartition, Long> resetPartitions = new HashMap<>();
boolean doReset = false;
final Map<TopicPartition, Long> newOffsetInMetadata = new HashMap<>();

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
final long outOfRangeOffset = outOfRangePartition.getValue();

StreamPartition<Integer> 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);
Comment thread
FrankChen021 marked this conversation as resolved.
// 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) {

@samarthjain samarthjain Dec 2, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this logic can further be refined here to call reset within this block itself. Also, it looks like an OffsetOutOfRangeException is thrown when the offset for the partition is either larger or smaller than the range of offsets the server has for the given partition. So the case of earliestAvailableOffset <= outofRangeOffset <= latestAvailableOffset doesn't apply.

So it could look something like this:

        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.

          logger.warn("Seeking kafka offset to earliest offset: " + earliestAvailableOffset);
          recordSupplier.seek(streamPartition, earliestAvailableOffset);
          // TBD: still need to confirm if this should be outOfRangeOffset or earliestAvailableOffset
          newOffsetInMetadata.put(topicPartition, outOfRangeOffset);
          logger.warn("Resetting offset in metadata for "
                      + topicPartition
                      + " to earliest offset: "
                      + earliestAvailableOffset);
          sendResetRequestAndWait(CollectionUtils.mapKeys(
              newOffsetInMetadata,
              streamPartition -> StreamPartition.of(
                  streamPartition.topic(),
                  streamPartition.partition()
              )
          ), taskToolbox);
        } else {
          // With the offset not in range (earliestAvailableOffset, latestAvailableOffset), there is not much we can do
          // but wait for the available offsets in the partition to arrive in the range.
          logger.warn("Offset "
                      + outOfRangeOffset
                      + " is out of range of the available offsets for "
                      + topicPartition
                      + ". It is likely that a manual offset reset of the supervisor is needed");

          log.warn("Retrying in %dms", task.getPollRetryMs());
          pollRetryLock.lockInterruptibly();
          try {
            long nanos = TimeUnit.MILLISECONDS.toNanos(task.getPollRetryMs());
            while (nanos > 0L && !pauseRequested && !stopRequested.get()) {
              nanos = isAwaitingRetry.awaitNanos(nanos);
            }
          }
          finally {
            pollRetryLock.unlock();
          }
        }

@samarthjain samarthjain Dec 2, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@gianm - could you explain the scenario in which the offset we are asking the consumer to seek to could possibly be higher than the latest available offset in Kafka?

@Nonnull
  @Override
  protected List<OrderedPartitionableRecord<Integer, Long, KafkaRecordEntity>> getRecords(
      RecordSupplier<Integer, Long, KafkaRecordEntity> recordSupplier,
      TaskToolbox toolbox
  ) throws Exception
  {
    try {
      return recordSupplier.poll(task.getIOConfig().getPollTimeout());
    }
    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
      //
      log.warn("OffsetOutOfRangeException with message [%s]", e.getMessage());
      possiblyResetOffsetsOrWait(e.offsetOutOfRangePartitions(), recordSupplier, toolbox);
      return Collections.emptyList();
    }
  }

@gianm gianm Dec 2, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For a partition that we've already committed some offset for, the starting offset given to a task for that partition is going to be the committed offset plus 1. So if no new messages have been written since the last commit, the starting offset will not exist yet.

@samarthjain samarthjain Dec 2, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the clarification, @gianm!

@FrankChen021 - I would probably get rid of the log line I mentioned then and have the else block as it is (but moved as else condition for if (outOfRangeOffset < earliestAvailableOffset)

@FrankChen021 FrankChen021 Dec 18, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@samarthjain I kept the code block as it was because original code reset the meta for all partition out of the partition loop and only reset for one time.

//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);

Comment thread
FrankChen021 marked this conversation as resolved.
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);
Expand Down
4 changes: 3 additions & 1 deletion ...src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand All @@ -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();
}
}
Expand Down