Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flink.apache.org
Content-Type: text/plain; charset=us-ascii
Mime-Version: 1.0 (Mac OS X Mail 9.3 \(3124\))
Subject: Re: Checkpoint takes long with FlinkKafkaConsumer
From: Kostas Kloudas <k.kloudas@data-artisans.com>
In-Reply-To: <CAMvK=Yozbt1hALQE4bdb=ZsUxYh+4SV5vVMgUmhKN5_D55EFAg@mail.gmail.com>
Date: Tue, 14 Jun 2016 14:23:45 +0200
Content-Transfer-Encoding: quoted-printable
Message-Id: <055BE947-709F-48C1-B935-D1BC978FF0C4@data-artisans.com>
References: <CAMvK=YovtOyzMpn0FB7yu9_MJPEWYh_9kmHy0k70w2VbUoh+Vw@mail.gmail.com> <ADE30D19-F872-4F66-83DF-56AFE48DE031@data-artisans.com> <CAMvK=Yozbt1hALQE4bdb=ZsUxYh+4SV5vVMgUmhKN5_D55EFAg@mail.gmail.com>
To: user@flink.apache.org
archived-at: Tue, 14 Jun 2016 12:23:57 -0000

Hi Hironori,

Could you also provide the logs of the taskManager?

As you described, it seems that the consumer is stuck in the polling =
loop, although Flink polls with
a timeout. This would normally mean that periodically it should release =
the lock for the checkpoints to go through.

The logs of the task manager can help at clarifying why this does not =
happen.

Thanks,
Kostas

> On Jun 14, 2016, at 12:48 PM, Hironori Ogibayashi =
<ogibayashi@gmail.com> wrote:
>=20
> Kostas,
>=20
> Thank you for your response.
> Yes, I am using latest Flink, which is 1.0.3.
>=20
> Thanks,
> Hironori
>=20
> 2016-06-14 19:02 GMT+09:00 Kostas Kloudas =
<k.kloudas@data-artisans.com>:
>> Hello Hironori,
>>=20
>> Are you using the latest Flink version?
>> There were some changes in the FlinkConsumer in the latest releases.
>>=20
>> Thanks,
>> Kostas
>>=20
>>> On Jun 14, 2016, at 11:52 AM, Hironori Ogibayashi =
<ogibayashi@gmail.com> wrote:
>>>=20
>>> Hello,
>>>=20
>>> I am running Flink job which reads topics from Kafka and write =
results
>>> to Redis. I use FsStatebackend with HDFS.
>>>=20
>>> I noticed that taking checkpoint takes serveral minutes and =
sometimes expires.
>>> ---
>>> 2016-06-14 17:25:40,734 INFO
>>> org.apache.flink.runtime.checkpoint.CheckpointCoordinator     -
>>> Completed checkpoint 1456 (in 257956 ms)
>>> 2016-06-14 17:25:40,735 INFO
>>> org.apache.flink.runtime.checkpoint.CheckpointCoordinator     -
>>> Triggering checkpoint 1457 @ 1465892740734
>>> 2016-06-14 17:35:40,735 INFO
>>> org.apache.flink.runtime.checkpoint.CheckpointCoordinator     -
>>> Checkpoint 1457 expired before completing.
>>> 2016-06-14 17:35:40,736 INFO
>>> org.apache.flink.runtime.checkpoint.CheckpointCoordinator     -
>>> Triggering checkpoint 1458 @ 1465893340735
>>> 2016-06-14 17:45:40,736 INFO
>>> org.apache.flink.runtime.checkpoint.CheckpointCoordinator     -
>>> Checkpoint 1458 expired before completing.
>>> 2016-06-14 17:45:40,737 INFO
>>> org.apache.flink.runtime.checkpoint.CheckpointCoordinator     -
>>> Triggering checkpoint 1459 @ 1465893940736
>>> 2016-06-14 17:55:40,738 INFO
>>> org.apache.flink.runtime.checkpoint.CheckpointCoordinator     -
>>> Checkpoint 1459 expired before completing.
>>> 2016-06-14 17:55:40,739 INFO
>>> org.apache.flink.runtime.checkpoint.CheckpointCoordinator     -
>>> Triggering checkpoint 1460 @ 1465894540738
>>> ---
>>>=20
>>> According to WebUI, checkpoint size is just 1MB. Why checkpointing
>>> takes so long?
>>>=20
>>> I took jstack during checkpointing. It looks that checkpointing =
thread
>>> is blocked in commitOffsets.
>>>=20
>>> ----
>>> "Async calls on Source: Custom Source -> Flat Map (2/3)" daemon
>>> prio=3D10 tid=3D0x00007f2b14010800 nid=3D0x1b89a waiting for monitor =
entry
>>> [0x00007f2b3ddfc000]
>>>  java.lang.Thread.State: BLOCKED (on object monitor)
>>>       at =
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09.commitOff=
sets(FlinkKafkaConsumer09.java:392)
>>>       - waiting to lock <0x0000000659111b58> (a
>>> org.apache.kafka.clients.consumer.KafkaConsumer)
>>>       at =
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.notifyC=
heckpointComplete(FlinkKafkaConsumerBase.java:169)
>>>       at =
org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.notifyO=
fCompletedCheckpoint(AbstractUdfStreamOperator.java:179)
>>>       at =
org.apache.flink.streaming.runtime.tasks.StreamTask.notifyCheckpointComple=
te(StreamTask.java:596)
>>>       - locked <0x0000000659111cc8> (a java.lang.Object)
>>>       at =
org.apache.flink.runtime.taskmanager.Task$3.run(Task.java:945)
>>>       at =
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
>>>       at java.util.concurrent.FutureTask.run(FutureTask.java:262)
>>>       at =
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:=
1145)
>>>       at =
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java=
:615)
>>>       at java.lang.Thread.run(Thread.java:745)
>>> ---
>>>=20
>>> Blocker is this.
>>>=20
>>> ---
>>> "Thread-9" daemon prio=3D10 tid=3D0x00007f2b2440d000 nid=3D0x1b838 =
runnable
>>> [0x00007f2b3dbfa000]
>>>  java.lang.Thread.State: RUNNABLE
>>>       at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
>>>       at =
sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
>>>       at =
sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:79)
>>>       at =
sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:87)
>>>       - locked <0x0000000659457dc8> (a sun.nio.ch.Util$2)
>>>       - locked <0x0000000659457db8> (a =
java.util.Collections$UnmodifiableSet)
>>>       - locked <0x0000000659457108> (a sun.nio.ch.EPollSelectorImpl)
>>>       at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:98)
>>>       at =
org.apache.kafka.common.network.Selector.select(Selector.java:425)
>>>       at =
org.apache.kafka.common.network.Selector.poll(Selector.java:254)
>>>       at =
org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:256)
>>>       at =
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPo=
ll(ConsumerNetworkClient.java:320)
>>>       at =
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(Con=
sumerNetworkClient.java:213)
>>>       at =
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(Con=
sumerNetworkClient.java:193)
>>>       at =
org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.jav=
a:908)
>>>       at =
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:85=
3)
>>>       at =
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09$ConsumerT=
hread.run(FlinkKafkaConsumer09.java:449)
>>>       - locked <0x0000000659111b58> (a
>>> org.apache.kafka.clients.consumer.KafkaConsumer)
>>> ---
>>>=20
>>> If someone could advise me of the cause or the way to investigate
>>> further, that would be appreciated.
>>>=20
>>> Thanks,
>>> Hironori
>>=20