Re: Flink is looking for Kafka topic "n/a"

2018-03-08 Thread Nico Kruber
I think, I found a code path (race between threads) that may lead to two markers being in the list. I created https://issues.apache.org/jira/browse/FLINK-8896 to track this and will have a pull request ready (probably) today. Nico On 07/03/18 10:09, Mu Kong wrote: > Hi Gordon, > > Thanks for

Re: Flink is looking for Kafka topic "n/a"

2018-03-07 Thread Mu Kong
Hi Gordon, Thanks for your response. I think I've misspoken about the failure after "n/a" exception. The behavior after this exception would be: switched from RUNNING to CANCELING switched from CANCELING to CANCELED Try to restart or fail the job "X" () if no longer

Re: Flink is looking for Kafka topic "n/a"

2018-03-06 Thread Tzu-Li Tai
Hi Mu, You mentioned that the job stopped after the "n/a" topic error, but the job failed to recover. What exception did you encounter in the restart executions? Was it the same error? This would verify if we actually should be removing more than one of these special MARKER partition states. On

Re: Flink is looking for Kafka topic "n/a"

2018-03-06 Thread Mu Kong
Hi Nico, Thanks for your prompt response. I'm using Flink 1.3.0 for this job. Please let me know if you need more information. Best regards, Mu On Tue, Mar 6, 2018 at 10:17 PM, Nico Kruber wrote: > Hi Mu, > which version of flink are you using? I checked the latest

Re: Flink is looking for Kafka topic "n/a"

2018-03-06 Thread Nico Kruber
Hi Mu, which version of flink are you using? I checked the latest branches for 1.2 - 1.5 to look for findLeaderForPartitions at line 205 in Kafka08Fetcher but they did not match. From what I can see in the code, there is a MARKER partition state with topic "n/a" but that is explicitly removed from