Re: Flink Kafka Consumer Behaviour

[Stephan Ewen]

The latest Flink has consumers for Kafka 0.8, 0.9, 0.10 - which one are you
using?

I would assume you use Flink with Kafka 0.8.x, because as far as I know,
starting from Kafka 0.9, offsets are not handled by ZooKeeper any more...

On Mon, Apr 24, 2017 at 12:18 AM, Meghashyam Sandeep V <
vr1meghash...@gmail.com> wrote:

> Hi All,
>
> Sorry for the miscommunication. I'm not using 0.8. I'm using latest
> available flink-kafka client. I don't see my app registered as a consumer
> group. I wanted to know if there is a way to monitor Kafka offsets.
>
> Thanks,
> Sandeep
>
> On Apr 23, 2017 9:38 AM, "Stephan Ewen"  wrote:
>
>> Since it is something special to Kafka 0.8, it could be implemented in a
>> simple addition to the ZooKeeperOffsetHandler used by the
>> FlinkKafkaConsumer08.
>>
>> Would you be willing to contribute this? That would certainly help
>> speeding up the resolution of the issue...
>>
>>
>> On Fri, Apr 21, 2017 at 2:33 AM, Tzu-Li (Gordon) Tai > > wrote:
>>
>>> One additional note:
>>>
>>> In FlinkKafkaConsumer 0.9+, the current read offset should already exist
>>> in Flink metrics.
>>> See https://issues.apache.org/jira/browse/FLINK-4186.
>>>
>>> But yes, this is still missing for 0.8, so you need to directly query ZK
>>> for this.
>>>
>>> Cheers,
>>> Gordon
>>>
>>>
>>> On 21 April 2017 at 8:28:09 AM, Tzu-Li (Gordon) Tai (tzuli...@apache.org)
>>> wrote:
>>>
>>> Hi Sandeep,
>>>
>>> It isn’t fixed yet, so I think external tools like the Kafka offset
>>> checker still won’t work.
>>> If you’re using 08 and is currently stuck with this issue, you can still
>>> directly query ZK to get the offsets.
>>>
>>> I think for FlinkKafkaConsumer09 the offset is exposed to Flink's metric
>>> system using Kafka’s own returned metrics, but for 08 this is still missing.
>>>
>>> There is this JIRA [1] that aims at exposing consumer lag across all
>>> Kafka consumer versions to Flink metrics. Perhaps it would make sense to
>>> also generally expose the offset for all Kafka consumer versions to Flink
>>> metrics as well.
>>>
>>> - Gordon
>>>
>>> [1] https://issues.apache.org/jira/browse/FLINK-6109
>>>
>>>
>>> On 19 April 2017 at 5:11:11 AM, sandeep6 (vr1meghash...@gmail.com)
>>> wrote:
>>>
>>> Is this fixed now? If not, is there any way to monitor kafka offset that
>>> is
>>> being processed by Flink? This should be a use case for everyone who uses
>>> Flink with Kafka.
>>>
>>>
>>>
>>> --
>>> View this message in context: http://apache-flink-user-maili
>>> ng-list-archive.2336050.n4.nabble.com/Flink-Kafka-Consumer-
>>> Behaviour-tp8257p12663.html
>>> Sent from the Apache Flink User Mailing List archive. mailing list
>>> archive at Nabble.com.
>>>
>>>
>>

Re: Flink Kafka Consumer Behaviour

[Meghashyam Sandeep V]

Hi All,

Sorry for the miscommunication. I'm not using 0.8. I'm using latest
available flink-kafka client. I don't see my app registered as a consumer
group. I wanted to know if there is a way to monitor Kafka offsets.

Thanks,
Sandeep

On Apr 23, 2017 9:38 AM, "Stephan Ewen"  wrote:

> Since it is something special to Kafka 0.8, it could be implemented in a
> simple addition to the ZooKeeperOffsetHandler used by the
> FlinkKafkaConsumer08.
>
> Would you be willing to contribute this? That would certainly help
> speeding up the resolution of the issue...
>
>
> On Fri, Apr 21, 2017 at 2:33 AM, Tzu-Li (Gordon) Tai 
> wrote:
>
>> One additional note:
>>
>> In FlinkKafkaConsumer 0.9+, the current read offset should already exist
>> in Flink metrics.
>> See https://issues.apache.org/jira/browse/FLINK-4186.
>>
>> But yes, this is still missing for 0.8, so you need to directly query ZK
>> for this.
>>
>> Cheers,
>> Gordon
>>
>>
>> On 21 April 2017 at 8:28:09 AM, Tzu-Li (Gordon) Tai (tzuli...@apache.org)
>> wrote:
>>
>> Hi Sandeep,
>>
>> It isn’t fixed yet, so I think external tools like the Kafka offset
>> checker still won’t work.
>> If you’re using 08 and is currently stuck with this issue, you can still
>> directly query ZK to get the offsets.
>>
>> I think for FlinkKafkaConsumer09 the offset is exposed to Flink's metric
>> system using Kafka’s own returned metrics, but for 08 this is still missing.
>>
>> There is this JIRA [1] that aims at exposing consumer lag across all
>> Kafka consumer versions to Flink metrics. Perhaps it would make sense to
>> also generally expose the offset for all Kafka consumer versions to Flink
>> metrics as well.
>>
>> - Gordon
>>
>> [1] https://issues.apache.org/jira/browse/FLINK-6109
>>
>>
>> On 19 April 2017 at 5:11:11 AM, sandeep6 (vr1meghash...@gmail.com) wrote:
>>
>> Is this fixed now? If not, is there any way to monitor kafka offset that
>> is
>> being processed by Flink? This should be a use case for everyone who uses
>> Flink with Kafka.
>>
>>
>>
>> --
>> View this message in context: http://apache-flink-user-maili
>> ng-list-archive.2336050.n4.nabble.com/Flink-Kafka-Consume
>> r-Behaviour-tp8257p12663.html
>> Sent from the Apache Flink User Mailing List archive. mailing list
>> archive at Nabble.com.
>>
>>
>

Re: Flink Kafka Consumer Behaviour

[Stephan Ewen]

Since it is something special to Kafka 0.8, it could be implemented in a
simple addition to the ZooKeeperOffsetHandler used by the
FlinkKafkaConsumer08.

Would you be willing to contribute this? That would certainly help speeding
up the resolution of the issue...


On Fri, Apr 21, 2017 at 2:33 AM, Tzu-Li (Gordon) Tai 
wrote:

> One additional note:
>
> In FlinkKafkaConsumer 0.9+, the current read offset should already exist
> in Flink metrics.
> See https://issues.apache.org/jira/browse/FLINK-4186.
>
> But yes, this is still missing for 0.8, so you need to directly query ZK
> for this.
>
> Cheers,
> Gordon
>
>
> On 21 April 2017 at 8:28:09 AM, Tzu-Li (Gordon) Tai (tzuli...@apache.org)
> wrote:
>
> Hi Sandeep,
>
> It isn’t fixed yet, so I think external tools like the Kafka offset
> checker still won’t work.
> If you’re using 08 and is currently stuck with this issue, you can still
> directly query ZK to get the offsets.
>
> I think for FlinkKafkaConsumer09 the offset is exposed to Flink's metric
> system using Kafka’s own returned metrics, but for 08 this is still missing.
>
> There is this JIRA [1] that aims at exposing consumer lag across all Kafka
> consumer versions to Flink metrics. Perhaps it would make sense to also
> generally expose the offset for all Kafka consumer versions to Flink
> metrics as well.
>
> - Gordon
>
> [1] https://issues.apache.org/jira/browse/FLINK-6109
>
>
> On 19 April 2017 at 5:11:11 AM, sandeep6 (vr1meghash...@gmail.com) wrote:
>
> Is this fixed now? If not, is there any way to monitor kafka offset that is
> being processed by Flink? This should be a use case for everyone who uses
> Flink with Kafka.
>
>
>
> --
> View this message in context: http://apache-flink-user-
> mailing-list-archive.2336050.n4.nabble.com/Flink-Kafka-Consumer-Behaviour-
> tp8257p12663.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>
>

Re: Flink Kafka Consumer Behaviour

[Tzu-Li (Gordon) Tai]

Hi Sandeep,

It isn’t fixed yet, so I think external tools like the Kafka offset checker 
still won’t work.
If you’re using 08 and is currently stuck with this issue, you can still 
directly query ZK to get the offsets.

I think for FlinkKafkaConsumer09 the offset is exposed to Flink's metric system 
using Kafka’s own returned metrics, but for 08 this is still missing.

There is this JIRA [1] that aims at exposing consumer lag across all Kafka 
consumer versions to Flink metrics. Perhaps it would make sense to also 
generally expose the offset for all Kafka consumer versions to Flink metrics as 
well.

- Gordon

[1] https://issues.apache.org/jira/browse/FLINK-6109


On 19 April 2017 at 5:11:11 AM, sandeep6 (vr1meghash...@gmail.com) wrote:

Is this fixed now? If not, is there any way to monitor kafka offset that is  
being processed by Flink? This should be a use case for everyone who uses  
Flink with Kafka.  



--  
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Kafka-Consumer-Behaviour-tp8257p12663.html
  
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.  

Re: Flink Kafka Consumer Behaviour

[Robert Metzger]

Thank you for investigating the issue. I've filed a JIRA:
https://issues.apache.org/jira/browse/FLINK-4822

On Wed, Oct 12, 2016 at 8:12 PM, Anchit Jatana  wrote:

> Hi Janardhan/Stephan,
>
> I just figured out what the issue is (Talking about Flink KafkaConnector08,
> don't know about Flink KafkaConnector09)
>
> The reason why- bin/kafka-consumer-groups.sh --zookeeper
>  --describe --group  is not showing any
> result
> is because of the absence of the
>
> /consumers//owners/ in the zookeeper.
>
> The flink application is creating and updating
> /consumers//offsets// but not creating
> "owners"
> Znode
>
> If I manually create the Znode using the following:
>
> create /consumers//owners “firstchildren”
>
> create /consumers//owners/ null
>
> It works fine, bin/kafka-consumer-groups.sh --zookeeper
> 
> --describe --group  starts pulling offset results for me.
>
> I think this needs to be corrected in the application: to check and create
> "/consumers//owners/" if it does not exist.
>
> Regards,
> Anchit
>
>
>
> --
> View this message in context: http://apache-flink-user-
> mailing-list-archive.2336050.n4.nabble.com/Flink-Kafka-Consumer-Behaviour-
> tp8257p9499.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>

Re: Flink Kafka Consumer Behaviour

[Anchit Jatana]

Hi Janardhan/Stephan,

I just figured out what the issue is (Talking about Flink KafkaConnector08,
don't know about Flink KafkaConnector09)

The reason why- bin/kafka-consumer-groups.sh --zookeeper
 --describe --group  is not showing any result
is because of the absence of the 

/consumers//owners/ in the zookeeper. 

The flink application is creating and updating
/consumers//offsets// but not creating "owners"
Znode 

If I manually create the Znode using the following:

create /consumers//owners “firstchildren”

create /consumers//owners/ null

It works fine, bin/kafka-consumer-groups.sh --zookeeper 
--describe --group  starts pulling offset results for me.

I think this needs to be corrected in the application: to check and create
"/consumers//owners/" if it does not exist.

Regards,
Anchit



--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Kafka-Consumer-Behaviour-tp8257p9499.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

Re: Flink Kafka Consumer Behaviour

[Robert Metzger]

Hi Prabhu,

I'm pretty sure that the Kafka 09 consumer commits offsets to Kafka when
checkpointing is turned on.

In the FlinkKafkaConsumerBase.notifyCheckpointComplete(), we call
fetcher.commitSpecificOffsetsToKafka(checkpointOffsets);, which calls
this.consumer.commitSync(offsetsToCommit); in
Kafka09Fetcher.commitSpecificOffsetsToKafka().


On Mon, Aug 8, 2016 at 8:24 PM, vpra...@gmail.com  wrote:

> Hi Stephan,
>
> The flink kafka 09 connector does not do offset commits to kafka when
> checkpointing is turned on. Is there a way to monitor the offset lag in
> this
> case,
>
> I am turning on a flink job that reads data from kafka (has about a week
> data - around 7 TB) , currently the approximate way that I use is the
> number
> of records read shown in the flink-UI and the last offset in kafka.
>
> Thanks,
> Prabhu
>
>
>
> --
> View this message in context: http://apache-flink-user-
> mailing-list-archive.2336050.n4.nabble.com/Flink-Kafka-Consumer-Behaviour-
> tp8257p8375.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>

Re: Flink Kafka Consumer Behaviour

[Stephan Ewen]

Hi!

I have not used the Kafka Offset Checker before, maybe someone who worked
with that can chime in.

Greetings,
Stephan


On Wed, Aug 3, 2016 at 4:59 PM, Janardhan Reddy  wrote:

> I can see that offsets are stored in zookeeper and are not returned when i
> query through kafka offset checker.
>
> Can you please tell me how to monitor kafka flink consumer lag for 0.8
> flink kafka consumer.
>
> On Wed, Aug 3, 2016 at 3:29 PM, Stephan Ewen  wrote:
>
>> Hi!
>>
>> Just check in the code. The 0.8 FlinkKafkaConsumer should always commit
>> offsets, regardless of whether checkpointing is enables. The 0.9
>> FlinkKafkaConsumer actually does not do any periodic offset committing when
>> checkpointing is disabled.
>>
>> Greetings,
>> Stephan
>>
>> On Wed, Aug 3, 2016 at 7:36 AM, Janardhan Reddy <
>> janardhan.re...@olacabs.com> wrote:
>>
>>> Checkpointing wasn't enabled in the streaming job, but the offsets
>>> should have been committed to zookeeper.
>>>
>>> But we don't see the offsets being written to zookeeper.
>>>
>>> On Tue, Aug 2, 2016 at 7:41 PM, Till Rohrmann 
>>> wrote:
>>>
 Hi Janardhan,

 Flink should commit the current offsets to Zookeeper whenever a
 checkpoint has been completed. In case that you disabled checkpointing,
 then the offsets will be periodically committed to ZooKeeper. The default
 value is 60s.

 Could it be that there wasn't yet a completed checkpoint? Which version
 of Flink are you using?

 Cheers,
 Till

 On Tue, Aug 2, 2016 at 7:26 PM, Janardhan Reddy <
 janardhan.re...@olacabs.com> wrote:

> Hi,
>
> When the run the following command i am getting that no topic is
> available for that consumer group.  i am suing
> flink-connector-kafka-0.8_${scala.version}(2.11).
>
> ./bin/kafka-consumer-groups.sh --zookeeper <> --group <> --describe
>
> No topic available for consumer group provided
>
>
> Does the kafka consumer commit offset to the broker always ? Do we
> need to enable checkpointing for the offsets to be committed ?
>
>
> Thanks
>


>>>
>>
>

Re: Flink Kafka Consumer Behaviour

[Janardhan Reddy]

I can see that offsets are stored in zookeeper and are not returned when i
query through kafka offset checker.

Can you please tell me how to monitor kafka flink consumer lag for 0.8
flink kafka consumer.

On Wed, Aug 3, 2016 at 3:29 PM, Stephan Ewen  wrote:

> Hi!
>
> Just check in the code. The 0.8 FlinkKafkaConsumer should always commit
> offsets, regardless of whether checkpointing is enables. The 0.9
> FlinkKafkaConsumer actually does not do any periodic offset committing when
> checkpointing is disabled.
>
> Greetings,
> Stephan
>
> On Wed, Aug 3, 2016 at 7:36 AM, Janardhan Reddy <
> janardhan.re...@olacabs.com> wrote:
>
>> Checkpointing wasn't enabled in the streaming job, but the offsets should
>> have been committed to zookeeper.
>>
>> But we don't see the offsets being written to zookeeper.
>>
>> On Tue, Aug 2, 2016 at 7:41 PM, Till Rohrmann 
>> wrote:
>>
>>> Hi Janardhan,
>>>
>>> Flink should commit the current offsets to Zookeeper whenever a
>>> checkpoint has been completed. In case that you disabled checkpointing,
>>> then the offsets will be periodically committed to ZooKeeper. The default
>>> value is 60s.
>>>
>>> Could it be that there wasn't yet a completed checkpoint? Which version
>>> of Flink are you using?
>>>
>>> Cheers,
>>> Till
>>>
>>> On Tue, Aug 2, 2016 at 7:26 PM, Janardhan Reddy <
>>> janardhan.re...@olacabs.com> wrote:
>>>
 Hi,

 When the run the following command i am getting that no topic is
 available for that consumer group.  i am suing
 flink-connector-kafka-0.8_${scala.version}(2.11).

 ./bin/kafka-consumer-groups.sh --zookeeper <> --group <> --describe

 No topic available for consumer group provided


 Does the kafka consumer commit offset to the broker always ? Do we need
 to enable checkpointing for the offsets to be committed ?


 Thanks

>>>
>>>
>>
>

Re: Flink Kafka Consumer Behaviour

[Janardhan Reddy]

Checkpointing wasn't enabled in the streaming job, but the offsets should
have been committed to zookeeper.

But we don't see the offsets being written to zookeeper.

On Tue, Aug 2, 2016 at 7:41 PM, Till Rohrmann  wrote:

> Hi Janardhan,
>
> Flink should commit the current offsets to Zookeeper whenever a checkpoint
> has been completed. In case that you disabled checkpointing, then the
> offsets will be periodically committed to ZooKeeper. The default value is
> 60s.
>
> Could it be that there wasn't yet a completed checkpoint? Which version of
> Flink are you using?
>
> Cheers,
> Till
>
> On Tue, Aug 2, 2016 at 7:26 PM, Janardhan Reddy <
> janardhan.re...@olacabs.com> wrote:
>
>> Hi,
>>
>> When the run the following command i am getting that no topic is
>> available for that consumer group.  i am suing
>> flink-connector-kafka-0.8_${scala.version}(2.11).
>>
>> ./bin/kafka-consumer-groups.sh --zookeeper <> --group <> --describe
>>
>> No topic available for consumer group provided
>>
>>
>> Does the kafka consumer commit offset to the broker always ? Do we need
>> to enable checkpointing for the offsets to be committed ?
>>
>>
>> Thanks
>>
>
>

Re: Flink Kafka Consumer Behaviour

[Till Rohrmann]

Hi Janardhan,

Flink should commit the current offsets to Zookeeper whenever a checkpoint
has been completed. In case that you disabled checkpointing, then the
offsets will be periodically committed to ZooKeeper. The default value is
60s.

Could it be that there wasn't yet a completed checkpoint? Which version of
Flink are you using?

Cheers,
Till

On Tue, Aug 2, 2016 at 7:26 PM, Janardhan Reddy  wrote:

> Hi,
>
> When the run the following command i am getting that no topic is available
> for that consumer group.  i am suing
> flink-connector-kafka-0.8_${scala.version}(2.11).
>
> ./bin/kafka-consumer-groups.sh --zookeeper <> --group <> --describe
>
> No topic available for consumer group provided
>
>
> Does the kafka consumer commit offset to the broker always ? Do we need to
> enable checkpointing for the offsets to be committed ?
>
>
> Thanks
>