Hi Gordon, - I'm currently running my flink program on 1.2 SNAPSHOT with kafka source and I have checkpoint enabled. When I look at the consumer offsets in kafka it appears to be stagnant and there is a huge lag. But I can see my flink program is in pace with kafka source in JMX metrics and outputs. Is there a way to identify why the offsets are not committed to kafka?
On which commit was your Kafka connector built? There was a recent change to the offset committing for Kafka 0.9 consumer, so identifying the exact commit will help clarify whether the recent change introduced any new problems. Also, what is your configured checkpoint interval? When checkpointing is enabled, the Kafka consumer only commits to Kafka when checkpoints are completed. So, offsets in Kafka are not updated until the next checkpoint is triggered. I am using 1.2-SNAPSHOT 'org.apache.flink', name: 'flink-connector-kafka-base_2.11', version: '1.2-SNAPSHOT' 'org.apache.flink', name: 'flink-connector-kafka-0.9_2.11', version: '1.2-SNAPSHOT' - In my current application we custom loggers for debugging purposes. Let’s say we want to find what’s happening for a particular user, we fire an api request to add the custom logger for that particular user and use it for logging along the data path. Is there a way to achieve this in flink? Are there any global mutable parameters that I can use to achieve this functionality? I’m not sure if I understand the use case correctly, but it seems like you will need to change configuration / behaviour of a specific Flink operator at runtime. To my knowledge, the best way to do this in Flink right now is to translate your original logging-trigger api requests to a stream of events fed to Flink. This stream of events will then basically be changes of your user logger behaviour, and your operators can change its logging behaviour according to this stream. I can send the changes as streams, but I need this change for all the operators in my pipeline. Instead of using coflatmap at each operator to combine the streams, is there a way to send a change to all the operators? - Can I pass on state between operators? If I need the state stored on previous operators, how can I fetch it? I don’t think this is possible. Fine, thanks. Thanks. On Mon, Oct 3, 2016 at 12:23 AM, Tzu-Li (Gordon) Tai <tzuli...@apache.org> wrote: > Hi! > > - I'm currently running my flink program on 1.2 SNAPSHOT with kafka source > and I have checkpoint enabled. When I look at the consumer offsets in kafka > it appears to be stagnant and there is a huge lag. But I can see my flink > program is in pace with kafka source in JMX metrics and outputs. Is there a > way to identify why the offsets are not committed to kafka? > > On which commit was your Kafka connector built? There was a recent change > to the offset committing for Kafka 0.9 consumer, so identifying the exact > commit will help clarify whether the recent change introduced any new > problems. Also, what is your configured checkpoint interval? When > checkpointing is enabled, the Kafka consumer only commits to Kafka when > checkpoints are completed. So, offsets in Kafka are not updated until the > next checkpoint is triggered. > > - In my current application we custom loggers for debugging purposes. > Let’s say we want to find what’s happening for a particular user, we fire > an api request to add the custom logger for that particular user and use it > for logging along the data path. Is there a way to achieve this in flink? > Are there any global mutable parameters that I can use to achieve this > functionality? > > I’m not sure if I understand the use case correctly, but it seems like you > will need to change configuration / behaviour of a specific Flink operator > at runtime. To my knowledge, the best way to do this in Flink right now is > to translate your original logging-trigger api requests to a stream of > events fed to Flink. This stream of events will then basically be changes > of your user logger behaviour, and your operators can change its logging > behaviour according to this stream. > > - Can I pass on state between operators? If I need the state stored on > previous operators, how can I fetch it? > > I don’t think this is possible. > > > Best Regards, > Gordon > > > On October 3, 2016 at 2:08:31 PM, Govindarajan Srinivasaraghavan ( > govindragh...@gmail.com) wrote: > > Hi, > > > > I have few questions on how I need to model my use case in flink. Please > advise. Thanks for the help. > > > > - I'm currently running my flink program on 1.2 SNAPSHOT with kafka source > and I have checkpoint enabled. When I look at the consumer offsets in kafka > it appears to be stagnant and there is a huge lag. But I can see my flink > program is in pace with kafka source in JMX metrics and outputs. Is there a > way to identify why the offsets are not committed to kafka? > > > > - In my current application we custom loggers for debugging purposes. > Let’s say we want to find what’s happening for a particular user, we fire > an api request to add the custom logger for that particular user and use it > for logging along the data path. Is there a way to achieve this in flink? > Are there any global mutable parameters that I can use to achieve this > functionality? > > > > - Can I pass on state between operators? If I need the state stored on > previous operators, how can I fetch it? > > > > Thanks > >
