[FLINK-13124] Stop fails with Universal Kafka Consumer

• Type: Bug
• Status: Closed
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 1.9.0
• Fix Version/s: 1.9.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: Aljoscha Krettek
• Reporter: Konstantin Knauf
• Created: 05/Jul/19 13:36
• Updated: 15/Jul/19 11:31
• Resolved: 15/Jul/19 11:31

Description

When running the StateMachineExample (with the universal Kafka connector instead of 0.10) the Job always crashes with the following exception, when stopping it with flink stop <job-id>.

2019-07-05 13:16:49,809 INFO org.apache.flink.runtime.taskmanager.Task - Source: Custom Source (1/1) (1a22bba845872431e8695fc8f3793fcd) switched from RUNNING to FAILED.
java.lang.Exception: org.apache.flink.streaming.connectors.kafka.internal.Handover$ClosedException
taskmanager_1 | at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.checkThrowSourceExecutionException(SourceStreamTask.java:194)
taskmanager_1 | at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.performDefaultAction(SourceStreamTask.java:121)
taskmanager_1 | at org.apache.flink.streaming.runtime.tasks.StreamTask.run(StreamTask.java:268)
taskmanager_1 | at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:376)
taskmanager_1 | at org.apache.flink.runtime.taskmanager.Task.run(Task.java:675)
taskmanager_1 | at java.lang.Thread.run(Thread.java:748)
taskmanager_1 | Caused by: org.apache.flink.streaming.connectors.kafka.internal.Handover$ClosedException
taskmanager_1 | at org.apache.flink.streaming.connectors.kafka.internal.Handover.close(Handover.java:182)
taskmanager_1 | at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.cancel(KafkaFetcher.java:175)
taskmanager_1 | at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.cancel(FlinkKafkaConsumerBase.java:817)
taskmanager_1 | at org.apache.flink.streaming.api.operators.StreamSource.cancel(StreamSource.java:124)
taskmanager_1 | at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.cancelTask(SourceStreamTask.java:144)
taskmanager_1 | at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.finishTask(SourceStreamTask.java:150)
taskmanager_1 | at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:781)
taskmanager_1 | at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:656)
taskmanager_1 | at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.triggerCheckpoint(SourceStreamTask.java:160)
taskmanager_1 | at org.apache.flink.runtime.taskmanager.Task$1.run(Task.java:1120)
taskmanager_1 | at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1626)
taskmanager_1 | at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
taskmanager_1 | at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
taskmanager_1 | ... 1 more

Rebased on a `86bee8679112e76372a84083b1af18722644e1a0` the stacktrace looks like this:

taskmanager_1  | 2019-07-08 13:01:01,287 INFO  org.apache.flink.runtime.taskmanager.Task                     - Source: Custom Source (1/1) (54504e053415cee82a2a7b4293d325e9) switched from RUNNING to FAILED.
taskmanager_1  | org.apache.flink.streaming.connectors.kafka.internal.Handover$ClosedException
taskmanager_1  | 	at org.apache.flink.streaming.connectors.kafka.internal.Handover.close(Handover.java:182)
taskmanager_1  | 	at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.cancel(KafkaFetcher.java:175)
taskmanager_1  | 	at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.cancel(FlinkKafkaConsumerBase.java:814)
taskmanager_1  | 	at org.apache.flink.streaming.api.operators.StreamSource.cancel(StreamSource.java:124)
taskmanager_1  | 	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.cancelTask(SourceStreamTask.java:108)
taskmanager_1  | 	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.finishTask(SourceStreamTask.java:114)
taskmanager_1  | 	at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:729)
taskmanager_1  | 	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:604)
taskmanager_1  | 	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.triggerCheckpoint(SourceStreamTask.java:124)
taskmanager_1  | 	at org.apache.flink.runtime.taskmanager.Task$1.run(Task.java:1151)
taskmanager_1  | 	at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1626)
taskmanager_1  | 	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
taskmanager_1  | 	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
taskmanager_1  | 	at java.lang.Thread.run(Thread.java:748)

Rebased on 86bee8679112e76372a84083b1af18722644e1a0 without ExceptionUtils.rethrowException(error, error.getMessage()); in Handover#pollNext():

 2019-07-09 09:00:35,498 INFO  org.apache.flink.runtime.taskmanager.Task                     - Source: Custom Source (1/1) (8568229c7efcddf75545a503bdb737f8) switched from RUNNING to FAILED.
taskmanager_1  | org.apache.flink.util.FlinkException: org.apache.flink.streaming.connectors.kafka.internal.Handover$ClosedException
taskmanager_1  | 	at org.apache.flink.streaming.connectors.kafka.internal.Handover.pollNext(Handover.java:85)
taskmanager_1  | 	at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:131)
taskmanager_1  | 	at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:711)
taskmanager_1  | 	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:95)
taskmanager_1  | 	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:59)
taskmanager_1  | 	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:102)
taskmanager_1  | 	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:335)
taskmanager_1  | 	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:724)
taskmanager_1  | 	at java.lang.Thread.run(Thread.java:748)
taskmanager_1  | Caused by: org.apache.flink.streaming.connectors.kafka.internal.Handover$ClosedException
taskmanager_1  | 	at org.apache.flink.streaming.connectors.kafka.internal.Handover.close(Handover.java:184)
taskmanager_1  | 	at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.cancel(KafkaFetcher.java:175)
taskmanager_1  | 	at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.cancel(FlinkKafkaConsumerBase.java:814)
taskmanager_1  | 	at org.apache.flink.streaming.api.operators.StreamSource.cancel(StreamSource.java:124)
taskmanager_1  | 	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.cancelTask(SourceStreamTask.java:108)
taskmanager_1  | 	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.finishTask(SourceStreamTask.java:114)
taskmanager_1  | 	at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:729)
taskmanager_1  | 	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:604)
taskmanager_1  | 	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.triggerCheckpoint(SourceStreamTask.java:124)
taskmanager_1  | 	at org.apache.flink.runtime.taskmanager.Task$1.run(Task.java:1151)
taskmanager_1  | 	at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1626)
taskmanager_1  | 	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
taskmanager_1  | 	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
taskmanager_1  | 	... 1 more

Rebased on 86bee8679112e76372a84083b1af18722644e1a0 with the Kafka011 consumer.

org.apache.flink.streaming.connectors.kafka.internal.Handover$ClosedException
	at org.apache.flink.streaming.connectors.kafka.internal.Handover.close(Handover.java:182)
	at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.cancel(Kafka09Fetcher.java:181)
	at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.cancel(FlinkKafkaConsumerBase.java:814)
	at org.apache.flink.streaming.api.operators.StreamSource.cancel(StreamSource.java:124)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.cancelTask(SourceStreamTask.java:108)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.finishTask(SourceStreamTask.java:114)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:729)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:604)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.triggerCheckpoint(SourceStreamTask.java:124)
	at org.apache.flink.runtime.taskmanager.Task$1.run(Task.java:1151)
	at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1626)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

Issue Links

• blocks: FLINK-12749 Getting Started - Docker Playgrounds - Flink Cluster Playground
• links to: GitHub Pull Request #9090
• links to: GitHub Pull Request #9115