[KAFKA-5556] KafkaConsumer.commitSync throws IllegalStateException: Attempt to retrieve exception from future which hasn't failed

• Type: Bug
• Status: Resolved
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 0.10.2.1, 0.11.0.0
• Fix Version/s: 0.11.0.1, 1.0.0
• Component/s: clients
• Assignee: Umesh Chaudhary
• Reporter: Damian Guy
• Created: 04/Jul/17 16:06
• Updated: 20/Jul/17 16:09
• Resolved: 20/Jul/17 16:09

Description

From the user list:
I have been running a streaming application on some data set. Things
usually run ok. Today I was trying to run the same application on Kafka
(ver 0.10.2.1 on Scala 2.12) installed in a Mesos DC/OS cluster. After
running for quite some time, I got the following exception ..

Exception in thread "StreamThread-1" java.lang.IllegalStateException:
> Attempt to retrieve exception from future which hasn't failed
> at
> org.apache.kafka.clients.consumer.internals.RequestFuture.exception(RequestFuture.java:99)
> at
> org.apache.kafka.clients.consumer.internals.RequestFuture.isRetriable(RequestFuture.java:89)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:590)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1124)
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commitOffsets(StreamTask.java:296)
> at
> org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:79)
> at
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188)
> at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:280)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.commitOne(StreamThread.java:807)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:794)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:769)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:647)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:361)

Looks like we should check if the future is done, i.e., check the return value from poll and retry if time is remaining and !future.isDone()

Issue Links

• links to: GitHub Pull Request #3489