[
https://issues.apache.org/jira/browse/FLINK-8896?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16392797#comment-16392797
]
ASF GitHub Bot commented on FLINK-8896:
---------------------------------------
Github user StephanEwen commented on the issue:
https://github.com/apache/flink/pull/5661
Good fix, merging for 1.4, 1.5, 1.6
> Kafka08Fetcher trying to look up topic "n/a" on partiton "-1"
> -------------------------------------------------------------
>
> Key: FLINK-8896
> URL: https://issues.apache.org/jira/browse/FLINK-8896
> Project: Flink
> Issue Type: Bug
> Components: Kafka Connector
> Affects Versions: 1.3.0, 1.3.1, 1.4.0, 1.3.2, 1.5.0, 1.4.1, 1.6.0
> Reporter: Nico Kruber
> Assignee: Nico Kruber
> Priority: Critical
> Fix For: 1.5.0, 1.6.0
>
>
> A user on the [mailing
> list|https://lists.apache.org/thread.html/fa96b09fc1d3a7efdb1bf7946489edafed8cdf138e933e9d0d8948a1@%3Cuser.flink.apache.org%3E]
> reported this error:
> {code}
> java.lang.RuntimeException: Unable to find a leader for partitions:
> [Partition: KafkaTopicPartition{topic='n/a', partition=-1},
> KafkaPartitionHandle=[n/a,-1], offset=(not set)]
> at
> org.apache.flink.streaming.connectors.kafka.internals.Kafka08Fetcher.findLeaderForPartitions(Kafka08Fetcher.java:495)
> at
> org.apache.flink.streaming.connectors.kafka.internals.Kafka08Fetcher.runFetchLoop(Kafka08Fetcher.java:205)
> at
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:449)
> at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:87)
> at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:55)
> at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:95)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:262)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702)
> at java.lang.Thread.run(Thread.java:748)
> {code}
> The root cause seems to be that {{Kafka08Fetcher#MARKER}} is in the
> {{unassignedPartitionsQueue}} more than once which could come from multiple
> calls to {{Kafka08Fetcher#cancel()}}. One code path leading to this is
> {{FlinkKafkaConsumerBase#cancel()}} being called in one thread and
> {{FlinkKafkaConsumerBase}}'s partition discovery loop thread dropping out
> before the first thread was able to call {{Kafka08Fetcher#cancel}}.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
