Hi,
I just tried it with Kafka 0.8.2.0 and 0.8.2.1 and for both versions
everything worked fine.
How many partitions does your topic have?
Can you send me the full logs of the Kafka consumer?
On Tue, Apr 19, 2016 at 6:05 PM, Balaji Rajagopalan <
balaji.rajagopa...@olacabs.com> wrote:
> Flink version : 1.0.0
> Kafka version : 0.8.2.1
>
> Try to use a topic which has no message posted to it, at the time flink
> starts.
>
> On Tue, Apr 19, 2016 at 5:41 PM, Robert Metzger <rmetz...@apache.org>
> wrote:
>
>> Can you provide me with the exact Flink and Kafka versions you are using
>> and the steps to reproduce the issue?
>>
>> On Tue, Apr 19, 2016 at 2:06 PM, Balaji Rajagopalan <
>> balaji.rajagopa...@olacabs.com> wrote:
>>
>>> It does not seem to fully work if there is no data in the kafka stream,
>>> the flink application emits this error and bails, could this be missed use
>>> case in the fix.
>>>
>>> On Tue, Apr 19, 2016 at 3:58 PM, Robert Metzger <rmetz...@apache.org>
>>> wrote:
>>>
>>>> Hi,
>>>>
>>>> I'm sorry, the documentation in the JIRA issue is a bit incorrect. The
>>>> issue has been fixed in all versions including and after 1.0.0. Earlier
>>>> releases (0.10, 0.9) will fail when the leader changes.
>>>> However, you don't necessarily need to upgrade to Flink 1.0.0 to
>>>> resolve the issue: With checkpointing enabled, your job will fail on a
>>>> leader change, then Flink will restart the Kafka consumers and they'll find
>>>> the new leaders.
>>>> Starting from Flink 1.0.0 the Kafka consumer will handle leader changes
>>>> without failing.
>>>>
>>>> Regards,
>>>> Robert
>>>>
>>>> On Tue, Apr 19, 2016 at 12:17 PM, Balaji Rajagopalan <
>>>> balaji.rajagopa...@olacabs.com> wrote:
>>>>
>>>>> I am facing this exception repeatedly while trying to consume from
>>>>> kafka topic. It seems it was reported in 1.0.0 and fixed in 1.0.0, how can
>>>>> I be sure that is fixed in the version of flink that I am using, does it
>>>>> require me to install patch updates ?
>>>>>
>>>>> Caused by: java.lang.RuntimeException: Unable to find a leader for
>>>>> partitions: [FetchPartition {topic=capi, partition=0,
>>>>> offset=-915623761776}]
>>>>> at
>>>>> org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher.findLeaderForPartitions(LegacyFetcher.java:323)
>>>>> at
>>>>> org.apache.flink.streaming.connectors.kafka.internals.LegacyFetcher.run(LegacyFetcher.java:162)
>>>>> at
>>>>> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08.run(FlinkKafkaConsumer08.java:316)
>>>>> at
>>>>> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:78)
>>>>> at
>>>>> org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:56)
>>>>> at
>>>>> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:224)
>>>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
>>>>> at java.lang.Thread.run(Thread.java:745)
>>>>>
>>>>> https://issues.apache.org/jira/browse/FLINK-3368
>>>>>
>>>>
>>>>
>>>
>>
>
