[
https://issues.apache.org/jira/browse/FLINK-3368?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15140540#comment-15140540
]
Robert Metzger commented on FLINK-3368:
---------------------------------------
Flink's mechanism to detect the leading brokers and the assignment to the
TopicPartitions works properly.
Also, no leader information is kept across restarts.
I think the issue is that Kafka needs a lot of time to until all broker return
up-to-date metadata. In my experiment (with 400 partitions on 7 brokers and a
replication of 2) it took 13 minutes for all brokers returning valid
information again.
There are also some issues in Kafka which are not fixed in Kafka 0.8, relating
to metadata information in the brokers:
https://issues.apache.org/jira/browse/KAFKA-1367 (Broker topic metadata not
kept in sync with ZooKeeper)
https://issues.apache.org/jira/browse/KAFKA-972 (MetadataRequest returns stale
list of brokers)
(Cloudera has
[backported|http://www.cloudera.com/documentation/kafka/latest/topics/kafka_fi_131.html]
the fixes in their distribution)
Why are the high level java kafka consumers are not suffering from this?
Kafka is randomly choosing the broker to query for the metadata. Flink is using
always the first one. By randomly asking around, they'll eventually hit a
broker with valid information.
Also, Kafka is not failing all consumers if one is trying to pull data from the
wrong broker. They'll probably query again for metadata and then reconnect to
the new leader, without failing everything.
Therefore, I suggest to implement the mechanisms to handle leader changes into
the LegacyFetcher of our Flink Kafka 0.8 consumer.
> Kafka 0.8 consumer fails to recover from broker shutdowns
> ---------------------------------------------------------
>
> Key: FLINK-3368
> URL: https://issues.apache.org/jira/browse/FLINK-3368
> Project: Flink
> Issue Type: Bug
> Components: Kafka Connector
> Affects Versions: 1.0.0
> Reporter: Robert Metzger
> Assignee: Robert Metzger
> Priority: Blocker
>
> It seems that the Kafka Consumer (0.8) fails to restart a job after it failed
> due to a Kafka broker shutdown.
> {code}
> java.lang.Exception: Unable to get last offset for partitions [FetchPartition
> {topic=a, partition=13, offset=-915623761776}, FetchPartition {topic=b,
> partition=13, offset=-915623761776}, FetchPartition {topic=c, partition=13,
> offset=-915623761776}, FetchPartition {topic=d, partition=13,
> offset=-915623761776}, FetchPartition {topic=e, partition=13,
> offset=-915623761776}, FetchPartition {topic=f, partition=13,
> offset=-915623761776}, FetchPartition {topic=g, partition=13,
> offset=-915623761776}].
> Exception for partition 13: kafka.common.NotLeaderForPartitionException
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
> at
> sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
> at
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
> at java.lang.reflect.Constructor.newInstance(Constructor.java:422)
> at java.lang.Class.newInstance(Class.java:442)
> at kafka.common.ErrorMapping$.exceptionFor(ErrorMapping.scala:86)
> at kafka.common.ErrorMapping.exceptionFor(ErrorMapping.scala)
> at
> org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher$SimpleConsumerThread.getLastOffset(LegacyFetcher.java:551)
> at
> org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher$SimpleConsumerThread.run(LegacyFetcher.java:379)
> {code}
> I haven't understood the cause of this issue, but I'll investigate it.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
