samarthjain
commented
2 years ago @FrankChen021 - I believe the code is doing the right thing. It may be doing an unnecessary seek, but it shouldn't cause an infinite retry loop.
In line 140 when recordSupplier.getEarliestSequenceNumber(streamPartition) is called it does the following:
Long currPos = getPosition(partition);
seekToEarliest(Collections.singleton(partition));
Long nextPos = getPosition(partition);
seek(partition, currPos);So it gets the current position for partition, seeks to the earliest offset in that partition, gets the position, and then seeks back or restores to the position it was at. The earliest offset is returned and stored in leastAvailableOffset variable.
Then, at line 148 recordSupplier.seek(streamPartition, nextOffset);, assumes that the call to getEarliestSequenceNumber possibly caused the current position to be different from the position it stored at the start in nextOffset. So it does an extra seek to restore the state as it was before. The reason I said it may be unnecessary is because I saw log lines like this:
getEarliestSequenceNumber() logs:
Seeking to EARLIEST offset of partition logs_metrics-14
Resetting offset for partition insight_logs_metrics-14 to offset 2944387224.
Seeking to offset 22400192044 for partition logs_metrics-14
and then
// reset the seek
recordSupplier.seek(streamPartition, nextOffset);
Seeking to offset 22400192044 for partition logs_metrics-14As you can see, two seeks were performed to seek to offset 22400192044.
Having said that, I recently ran into an issue where even though I have resetOffsetAutomatically configured, Druid didn't call reset. This resulted in a stuck kafka consumer that just kept spewing OffsetOutOfRangeException messages. Looking closely at the code, it turned out that the offset reported in outOfRangePartition was (22400192044) which was higher than the leastAvailableOffset (2944387224)reported by Kafka. As a result the condition leastAvailableOffset > nextOffset was never met and the consumer just kept sleeping for 30 seconds before retrying again.
I wonder if you also have run into something similar?
FrankChen021
saifat29
OliveBZH
FarhadF
sc-sityad
Affected Version
Since 0.16
Description
There's a configuration
resetOffsetAutomaticallyinKafkaIndexTaskTuningConfigthat 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
nextOffsetis 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
leastAvailableOffsetvariable should be used to reset the offset. Since there's a check(Line 152) that guarantees that theleastAvailableOffsetis greater than current reading offset, the automatic resetting also won't causes data duplication. The fixes looks like as followsI will open a PR to fix this.