[ https://issues.apache.org/jira/browse/FLINK-13124?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Konstantin Knauf updated FLINK-13124: ------------------------------------- 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>}}. {noformat} taskmanager_1 | 2019-07-05 14:46:51,348 INFO org.apache.flink.runtime.taskmanager.Task - Source: Custom Source (1/1) (6dc79bc2d9654d140a2bab51456658de) switched from RUNNING to FAILED. taskmanager_1 | 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:298) taskmanager_1 | at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:403) taskmanager_1 | at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:685) taskmanager_1 | at org.apache.flink.runtime.taskmanager.Task.run(Task.java:515) 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:818) taskmanager_1 | at org.apache.flink.streaming.api.operators.StreamSource.cancel(StreamSource.java:134) 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:808) taskmanager_1 | at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:683) 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:1130) 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 {noformat} Rebased on a `86bee8679112e76372a84083b1af18722644e1a0` the StackTrace looks like this: {noformat} 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) {noformat} was: 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>}}. {noformat} taskmanager_1 | 2019-07-05 14:46:51,348 INFO org.apache.flink.runtime.taskmanager.Task - Source: Custom Source (1/1) (6dc79bc2d9654d140a2bab51456658de) switched from RUNNING to FAILED. taskmanager_1 | 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:298) taskmanager_1 | at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:403) taskmanager_1 | at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:685) taskmanager_1 | at org.apache.flink.runtime.taskmanager.Task.run(Task.java:515) 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:818) taskmanager_1 | at org.apache.flink.streaming.api.operators.StreamSource.cancel(StreamSource.java:134) 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:808) taskmanager_1 | at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:683) 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:1130) 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 {noformat} > Stop fails with Universal Kafka Consumer > ---------------------------------------- > > Key: FLINK-13124 > URL: https://issues.apache.org/jira/browse/FLINK-13124 > Project: Flink > Issue Type: Bug > Components: Connectors / Kafka > Affects Versions: 1.9.0 > Reporter: Konstantin Knauf > Priority: Blocker > > 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>}}. > {noformat} > taskmanager_1 | 2019-07-05 14:46:51,348 INFO > org.apache.flink.runtime.taskmanager.Task - Source: > Custom Source (1/1) (6dc79bc2d9654d140a2bab51456658de) switched from RUNNING > to FAILED. > taskmanager_1 | 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:298) > taskmanager_1 | at > org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:403) > taskmanager_1 | at > org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:685) > taskmanager_1 | at > org.apache.flink.runtime.taskmanager.Task.run(Task.java:515) > 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:818) > taskmanager_1 | at > org.apache.flink.streaming.api.operators.StreamSource.cancel(StreamSource.java:134) > 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:808) > taskmanager_1 | at > org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:683) > 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:1130) > 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 > {noformat} > Rebased on a `86bee8679112e76372a84083b1af18722644e1a0` the StackTrace looks > like this: > {noformat} > 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) > {noformat} -- This message was sent by Atlassian JIRA (v7.6.3#76005)