KAFKA-6581: Fix the ConsumerGroupCommand indefinite execution if one of the partition is unavailable. #4612
Conversation
…of the partition is unavailable. * Checks if partition available before calling consumer.position * Adds timeout on consumer.position() call.
|
@ijuma 0.10.0 branch doesn't seem to have jenkins.sh due to which above checks are failing. Can you please suggest what need to be done next? |
| consumer.seekToEnd(List(topicPartition).asJava) | ||
| val logEndOffset = consumer.position(topicPartition) | ||
| LogEndOffsetResult.LogEndOffset(logEndOffset) | ||
| zkUtils.getLeaderForPartition(topic, partition) match { |
There was a problem hiding this comment.
A better approach would be to use KafkaConsumer.endOffsets. Although its doc claims that it blocks indefinitely, it is in fact bound by the request timeout. I will submit a separate patch to correct this.
There was a problem hiding this comment.
KafkaConsumer in 0.10.0 doesn't have endOffsets. It was introduced from 0.10.1 i guess. 0.10.0 doesn't have request timeout too.
There was a problem hiding this comment.
Hmm, that's true. Is there any particular reason you need this in 0.10.0? We are unlikely to do another bug fix release for that version. Note that it should be possible to use tools from more recent versions on an 0.10.0 broker.
There was a problem hiding this comment.
We rely on this for consumer offset metrics and lose the visibility even if single partition becomes unavailable. We could but didn't want to maintain 2 versions just for this.
I see 0.10.0.2 as unreleased, no possibility there?
There was a problem hiding this comment.
It's very unlikely we'll release 0.10.0.2 at this point. Your next closest version is 0.11.0.2, which already has the fix that I suggested above.
Committer Checklist (excluded from commit message)