Fixes a bug in Kafka auto reset

[FrankChen021]

Fixes #11658

Description

The bug is detailed described in the issue above.

This PR fixes

1. the auto reset code by using the least available offset.
2. the doc that the auto reset function has nothing to do with the useEarliestOffset or useLatestOffset configuration parameters.

This PR has:

• been self-reviewed.
  • using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
• added documentation for new or modified features or behaviors.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added or updated version, license, or notice information in licenses.yaml
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
• added integration tests.
• been tested in a test Druid cluster.

[samarthjain]

Should this be earliestAvailableOffset and not outOfRangeOffset? I will have to trace the sendResetRequestAndWait code to better understand what it expects the offset to be.

[FrankChen021]

It should be the earliestAvailableOffset. This is a silly mistake. Thanks for pointing it out.

[samarthjain]

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]

@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]

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]

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]

@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.

[FrankChen021]

@samarthjain I'm almost exhausted these days. Give me more days to resolve your comments. Thanks!

[OliveBZH]

Hello, can we have a status on this merge ? Is it plan to merge it shortly ? Thanks

[github-actions]

This pull request has been marked as stale due to 60 days of inactivity.
It will be closed in 4 weeks if no further activity occurs. If you think
that's incorrect or this pull request should instead be reviewed, please simply
write any comment. Even if closed, you can still revive the PR at any time or
discuss it on the dev@druid.apache.org list.
Thank you for your contributions.

[github-actions]

This pull request/issue has been closed due to lack of activity. If you think that
is incorrect, or the pull request requires review, you can revive the PR at any time.