Re: FlinkKafkaConsumer09

Thanks! On Fri, Jul 29, 2016 at 11:43 AM, Gordon Tai (戴資力) wrote: > Hi Max, > > Sure, I was planning to do so, but wanted to see if it was a reasonable > feature to add before opening a JIRA :) > Here's the new JIRA: https://issues.apache.org/jira/browse/FLINK-4280 > >

Re: FlinkKafkaConsumer09

Hi Max, Sure, I was planning to do so, but wanted to see if it was a reasonable feature to add before opening a JIRA :) Here's the new JIRA: https://issues.apache.org/jira/browse/FLINK-4280 Regards, Gordon On Fri, Jul 29, 2016 at 4:03 PM, Maximilian Michels wrote: > Hi Tai, >

Re: FlinkKafkaConsumer09

Hi Tai, Should definitely be possible. Would you mind opening a JIRA issue with the description you posted? Thanks, Max On Thu, Jul 28, 2016 at 11:16 AM, Tai Gordon wrote: > Hi Kevin, > > Just a re-clarification: for Kafka 0.9 it would be “earliest”, & “smallest” > for the

Re: FlinkKafkaConsumer09

Hi Kevin, Just a re-clarification: for Kafka 0.9 it would be “earliest”, & “smallest” for the older Kafka 0.8. I’m wondering whether or not it is reasonable to add a Flink-specific way to set the consumer’s starting position to “earliest” and “latest”, without respecting the external Kafka

Re: FlinkKafkaConsumer09

Thank you Gordon and Max, Thank you Gordon, that explains the behaviour a bit better to me. I am now adding the timestamp to the group ID and that is a good workaround for now. The "smallest" option is unfortunately not available in this version of the FlinkKafkaConsumer class. Cheers,

Re: FlinkKafkaConsumer09

Hi Kevin, You need to use properties.setProperty("auto.offset.reset", "smallest") for Kafka 9 to start from the smallest offset. Note, that in Kafka 8 you need to use properties.setProperty("auto.offset.reset", "earliest") to achieve the same behavior. Kafka keeps track of the offsets per group

Re: FlinkKafkaConsumer09

Hi Kevin, Was the same “group.id” used before? What may be happening is that on startup of the consumer (not from failure restore), any existing committed offset for the groupId in Kafka’s brokers will be used as the starting point. The “auto.offset.reset” is only respected when no committed