Hi Gordon,
Do I need to clone and build release-1.1 branch to test this?
I currently use flinlk 1.1.2 runtime. When is the plan to release it
in 1.1.3?
Best Regards
Varaga
On Tue, Oct 4, 2016 at 9:25 AM, Tzu-Li (Gordon) Tai <tzuli...@apache.org>
wrote:
> Hi,
>
> Helping out here: this is the PR for async Kafka offset committing -
> https://github.com/apache/flink/pull/2574.
> It has already been merged into the master and release-1.1 branches, so
> you can try out the changes now if you’d like.
> The change should also be included in the 1.1.3 release, which the Flink
> community is discussing to release soon.
>
> Will definitely be helpful if you can provide feedback afterwards!
>
> Best Regards,
> Gordon
>
>
> On October 3, 2016 at 9:40:14 PM, Chakravarthy varaga (
> chakravarth...@gmail.com) wrote:
>
> Hi Stephan,
>
> Is the Async kafka offset commit released in 1.3.1?
>
> Varaga
>
> On Wed, Sep 28, 2016 at 9:49 AM, Chakravarthy varaga <
> chakravarth...@gmail.com> wrote:
>
>> Hi Stephan,
>>
>> That should be great. Let me know once the fix is done and the
>> snapshot version to use, I'll check and revert then.
>> Can you also share the JIRA that tracks the issue?
>>
>> With regards to offset commit issue, I'm not sure as to how to
>> proceed here. Probably I'll use your fix first and see if the problem
>> reoccurs.
>>
>> Thanks much
>> Varaga
>>
>> On Tue, Sep 27, 2016 at 7:46 PM, Stephan Ewen <se...@apache.org> wrote:
>>
>>> @CVP
>>>
>>> Flink stores in checkpoints in your case only the Kafka offsets (few
>>> bytes) and the custom state (e).
>>>
>>> Here is an illustration of the checkpoint and what is stored (from the
>>> Flink docs).
>>> https://ci.apache.org/projects/flink/flink-docs-master/inter
>>> nals/stream_checkpointing.html
>>>
>>>
>>> I am quite puzzled why the offset committing problem occurs only for one
>>> input, and not for the other.
>>> I am preparing a fix for 1.2, possibly going into 1.1.3 as well.
>>> Could you try out a snapshot version to see if that fixes your problem?
>>>
>>> Greetings,
>>> Stephan
>>>
>>>
>>>
>>> On Tue, Sep 27, 2016 at 2:24 PM, Chakravarthy varaga <
>>> chakravarth...@gmail.com> wrote:
>>>
>>>> Hi Stefan,
>>>>
>>>> Thanks a million for your detailed explanation. I appreciate it.
>>>>
>>>> - The *zookeeper bundled with kafka 0.9.0.1* was used to start
>>>> zookeeper. There is only 1 instance (standalone) of zookeeper running on my
>>>> localhost (ubuntu 14.04)
>>>> - There is only 1 Kafka broker (*version: 0.9.0.1* )
>>>>
>>>> With regards to Flink cluster there's only 1 JM & 2 TMs started
>>>> with no HA. I presume this does not use zookeeper anyways as it runs as
>>>> standalone cluster.
>>>>
>>>>
>>>> BTW., The kafka connector version that I use is as suggested in
>>>> the flink connectors page
>>>>
>>>>
>>>>
>>>>
>>>> *. <dependency>
>>>> <groupId>org.apache.flink</groupId>
>>>> <artifactId>flink-connector-kafka-0.9_2.10</artifactId>
>>>> <version>1.1.1</version> </dependency>*
>>>>
>>>> Do you see any issues with versions?
>>>>
>>>> 1) Do you have benchmarks wrt., to checkpointing in flink?
>>>>
>>>> 2) There isn't detailed explanation on what states are stored as
>>>> part of the checkpointing process. For ex., If I have pipeline like
>>>> *source -> map -> keyBy -> map -> sink, my assumption on what's stored
>>>> is:*
>>>>
>>>> * a) The source stream's custom watermarked records*
>>>>
>>>> * b) Intermediate states of each of the transformations in the
>>>> pipeline*
>>>>
>>>> * c) Delta of Records stored from the previous sink*
>>>>
>>>> * d) Custom States (SayValueState as in my case) - Essentially
>>>> this is what I bother about storing.*
>>>> * e) All of my operators*
>>>>
>>>> Is my understanding right?
>>>>
>>>> 3) Is there a way in Flink to checkpoint only d) as stated above
>>>>
>>>> 4) Can you apply checkpointing to only streams and certain
>>>> operators (say I wish to store aggregated values part of the
>>>> transformation)
>>>>
>>>> Best Regards
>>>> CVP
>>>>
>>>>
>>>> On Mon, Sep 26, 2016 at 6:18 PM, Stephan Ewen <se...@apache.org> wrote:
>>>>
>>>>> Thanks, the logs were very helpful!
>>>>>
>>>>> TL:DR - The offset committing to ZooKeeper is very slow and prevents
>>>>> proper starting of checkpoints.
>>>>>
>>>>> Here is what is happening in detail:
>>>>>
>>>>> - Between the point when the TaskManager receives the "trigger
>>>>> checkpoint" message and when the point when the KafkaSource actually
>>>>> starts
>>>>> the checkpoint is a long time (many seconds) - for one of the Kafka Inputs
>>>>> (the other is fine).
>>>>> - The only way this delayed can be introduced is if another
>>>>> checkpoint related operation (such as trigger() or notifyComplete() ) is
>>>>> still in progress when the checkpoint is started. Flink does not perform
>>>>> concurrent checkpoint operations on a single operator, to ease the
>>>>> concurrency model for users.
>>>>> - The operation that is still in progress must be the committing of
>>>>> the offsets (to ZooKeeper or Kafka). That also explains why this only
>>>>> happens once one side receives the first record. Before that, there is
>>>>> nothing to commit.
>>>>>
>>>>>
>>>>> What Flink should fix:
>>>>> - The KafkaConsumer should run the commit operations asynchronously,
>>>>> to not block the "notifyCheckpointComplete()" method.
>>>>>
>>>>> What you can fix:
>>>>> - Have a look at your Kafka/ZooKeeper setup. One Kafka Input works
>>>>> well, the other does not. Do they go against different sets of brokers, or
>>>>> different ZooKeepers? Is the metadata for one input bad?
>>>>> - In the next Flink version, you may opt-out of committing offsets
>>>>> to Kafka/ZooKeeper all together. It is not important for Flink's
>>>>> checkpoints anyways.
>>>>>
>>>>> Greetings,
>>>>> Stephan
>>>>>
>>>>>
>>>>> On Mon, Sep 26, 2016 at 5:13 PM, Chakravarthy varaga <
>>>>> chakravarth...@gmail.com> wrote:
>>>>>
>>>>>> Hi Stefan,
>>>>>>
>>>>>> Please find my responses below.
>>>>>>
>>>>>> - What source are you using for the slow input?
>>>>>> * [CVP] - Both stream as pointed out in my first mail, are Kafka
>>>>>> Streams*
>>>>>> - How large is the state that you are checkpointing?
>>>>>>
>>>>>> *[CVP] - I have enabled checkpointing on the StreamEnvironment as
>>>>>> below.*
>>>>>>
>>>>>>
>>>>>>
>>>>>> * final StreamExecutionEnvironment streamEnv =
>>>>>> StreamExecutionEnvironment.getExecutionEnvironment();
>>>>>> streamEnv.setStateBackend(new
>>>>>> FsStateBackend("file:///tmp/flink/checkpoints"));
>>>>>> streamEnv.enableCheckpointing(10000); *
>>>>>>
>>>>>>
>>>>>> * In terms of the state stored, the KS1 stream has payload of
>>>>>> 100K events/second, while KS2 have about 1 event / 10 minutes...
>>>>>> basically
>>>>>> the operators perform flatmaps on 8 fields of tuple (all fields are
>>>>>> primitives). If you look at the states' sizes in dashboard they are in
>>>>>> Kb... *
>>>>>> - Can you try to see in the log if actually the state snapshot
>>>>>> takes that long, or if it simply takes long for the checkpoint
>>>>>> barriers to travel through the stream due to a lot of backpressure?
>>>>>> [CVP] -There are no back pressure atleast from the sample
>>>>>> computation in the flink dashboard. 100K/second is low load for flink's
>>>>>> benchmarks. I could not quite get the barriers vs snapshot state. I have
>>>>>> attached the Task Manager log (DEBUG) info if that will interest you.
>>>>>>
>>>>>> I have attached the checkpoints times' as .png from the
>>>>>> dashboard. Basically if you look at checkpoint IDs 28 & 29 &30- you'd
>>>>>> see that the checkpoints take more than a minute in each case. Before
>>>>>> these
>>>>>> checkpoints, the KS2 stream did not have any events. As soon as an
>>>>>> event(should be in bytes) was generated, the checkpoints went slow and
>>>>>> subsequently a minute more for every checkpoint thereafter.
>>>>>>
>>>>>> This log was collected from the standalone flink cluster with 1
>>>>>> job manager & 2 TMs. 1 TM was running this application with checkpointing
>>>>>> (parallelism=1)
>>>>>>
>>>>>> Please let me know if you need further info.,
>>>>>>
>>>>>>
>>>>>>
>>>>>> On Fri, Sep 23, 2016 at 6:26 PM, Stephan Ewen <se...@apache.org>
>>>>>> wrote:
>>>>>>
>>>>>>> Hi!
>>>>>>>
>>>>>>> Let's try to figure that one out. Can you give us a bit more
>>>>>>> information?
>>>>>>>
>>>>>>> - What source are you using for the slow input?
>>>>>>> - How large is the state that you are checkpointing?
>>>>>>> - Can you try to see in the log if actually the state snapshot
>>>>>>> takes that long, or if it simply takes long for the checkpoint barriers
>>>>>>> to
>>>>>>> travel through the stream due to a lot of backpressure?
>>>>>>>
>>>>>>> Greetings,
>>>>>>> Stephan
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> On Fri, Sep 23, 2016 at 3:35 PM, Fabian Hueske <fhue...@gmail.com>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> Hi CVP,
>>>>>>>>
>>>>>>>> I'm not so much familiar with the internals of the checkpointing
>>>>>>>> system, but maybe Stephan (in CC) has an idea what's going on here.
>>>>>>>>
>>>>>>>> Best, Fabian
>>>>>>>>
>>>>>>>> 2016-09-23 11:33 GMT+02:00 Chakravarthy varaga <
>>>>>>>> chakravarth...@gmail.com>:
>>>>>>>>
>>>>>>>>> Hi Aljoscha & Fabian,
>>>>>>>>>
>>>>>>>>> I have a stream application that has 2 stream source as below.
>>>>>>>>>
>>>>>>>>> KeyedStream<String, String> *ks1* = ds1.keyBy("*") ;
>>>>>>>>> KeyedStream<Tuple2<String, V>, String> *ks2* =
>>>>>>>>> ds2.flatMap(split T into k-v pairs).keyBy(0);
>>>>>>>>>
>>>>>>>>> ks1.connect(ks2).flatMap(X);
>>>>>>>>> //X is a CoFlatMapFunction that inserts and removes elements
>>>>>>>>> from ks2 into a key-value state member. Elements from ks1 are matched
>>>>>>>>> against that state. the CoFlatMapFunction operator maintains
>>>>>>>>> ValueState<Tuple2<Long, Long>>;
>>>>>>>>>
>>>>>>>>> //ks1 is streaming about 100K events/sec from kafka topic
>>>>>>>>> //ks2 is streaming about 1 event every 10 minutes...
>>>>>>>>> Precisely when the 1st event is consumed from this stream, checkpoint
>>>>>>>>> takes
>>>>>>>>> 2 minutes straight away.
>>>>>>>>>
>>>>>>>>> The version of flink is 1.1.2.
>>>>>>>>>
>>>>>>>>> I tried to use checkpoint every 10 Secs using a FsStateBackend...
>>>>>>>>> What I notice is that the checkpoint duration is almost 2 minutes for
>>>>>>>>> many
>>>>>>>>> cases, while for the other cases it varies from 100 ms to 1.5 minutes
>>>>>>>>> frequently. I'm attaching the snapshot of the dashboard for your
>>>>>>>>> reference.
>>>>>>>>>
>>>>>>>>> Is this an issue with flink checkpointing?
>>>>>>>>>
>>>>>>>>> Best Regards
>>>>>>>>> CVP
>>>>>>>>>
>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>>
>
