apache / apache/druid

Infinite automatic Kafka offset resetting

Open
#11,658 11 comments 3 reactions 0 assignees View on GitHub
Area - Streaming Ingestion Bug
Dominant language
Java
Stars
14.1k
Forks
3.8k
Avg merge
2d 58m
Merged PRs (30d)
233

Description

### Affected Version

Since 0.16

### Description

There's a configuration `resetOffsetAutomatically` in `KafkaIndexTaskTuningConfig` that allows Kafka offset to be reset automatically once the Kafka offset is out of range. The error that offset is out of range typically occurs when messages in Kafka expires before the Druid ingestion task reads data from Kafka.

But current automatic resetting implementation uses a wrong offset to reset. That means the resetting does no take effect and causes another out of range error, and then automatic resetting is called again. The ingestion task falls into a dead loop.

### Problem Analysis

https://github.com/apache/druid/blob/59d257816b85dbeeca336b8e25d341d67bbc5697/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/IncrementalPublishingKafkaIndexTaskRunner.java#L134-L155

From the code(Line 148, Line 154) above we can see that, a variable `nextOffset` is used for automatic resetting. But this variable holds the offset we're currently reading from Kafka, and this is the offset that causes out of range exception(Line 134).

This means automatic resetting uses the offset which causes out of range to reset the offset. Of course, this resetting won't help and causes another out of range exception in the next round of polling messages from Kafka.

### How to fix

To fix this problem, the `leastAvailableOffset` variable should be used to reset the offset. Since there's a check(Line 152) that guarantees that the `leastAvailableOffset` is greater than current reading offset, the automatic resetting also won't causes data duplication. The fixes looks like as follows

```java
if (leastAvailableOffset > nextOffset) {
doReset = true;
resetPartitions.put(topicPartition, leastAvailableOffset);

recordSupplier.seek(streamPartition, leastAvailableOffset);
}
```

I will open a PR to fix this.

Contributor guide

Open the contributing guide

Research direction

Read extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/IncrementalPublishingKafkaIndexTaskRunner.java around lines 134-155, focusing on how nextOffset and leastAvailableOffset are used after an out-of-range error. Verify that automatic resetting targets leastAvailableOffset and that the ingestion task no longer repeats the out-of-range polling loop.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, kafka
Domain
stream-processing
Issue type
Bug
Difficulty
2/5
Estimated time
1-3 hours
Activity status
Stale
Clarity
Clearly specified
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.