flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Stephan Ewen <se...@apache.org>
Subject Re: ExpiredIteratorException when reading from a Kinesis stream
Date Thu, 03 Nov 2016 18:27:40 GMT
Is it possible that you have stalls in your topology?

Reasons could be:

  - The data sink blocks or becomes slow for some periods (where are you
sending the data to?)

  - If you are using large state and a state backend that only supports
synchronous checkpointing, there may be a delay introduced by the checkpoint


On Thu, Nov 3, 2016 at 7:21 PM, Scott Kidder <kidder.scott@gmail.com> wrote:

> Hi Steffan & Josh,
>
> For what it's worth, I've been using the Kinesis connector with very good
> results on Flink 1.1.2 and 1.1.3. I updated the Flink Kinesis connector KCL
> and AWS SDK dependencies to the following versions:
>
> aws.sdk.version: 1.11.34
> aws.kinesis-kcl.version: 1.7.0
>
> My customizations are visible in this commit on my fork:
> https://github.com/apache/flink/commit/6d69f99d7cd52b3c2f039cb4d37518
> 859e159b32
>
> It might be worth testing with newer AWS SDK & KCL libraries to see if the
> problem persists.
>
> Best,
>
> --Scott Kidder
>
>
> On Thu, Nov 3, 2016 at 7:08 AM, Josh <jofo90@gmail.com> wrote:
>
>> Hi Gordon,
>>
>> Thanks for the fast reply!
>> You're right about the expired iterator exception occurring just before
>> each spike. I can't see any signs of long GC on the task managers... CPU
>> has been <15% the whole time when the spikes were taking place and I can't
>> see anything unusual in the task manager logs.
>>
>> But actually I just noticed that the Flink UI showed no successful
>> checkpoints during the time of the problem even though my checkpoint
>> interval is 15 minutes. So I guess this is probably some kind of Flink
>> problem rather than a problem with the Kinesis consumer. Unfortunately I
>> can't find anything useful in the logs so not sure what happened!
>>
>> Josh
>>
>>
>>
>> On Thu, Nov 3, 2016 at 12:44 PM, Tzu-Li (Gordon) Tai <tzulitai@apache.org
>> > wrote:
>>
>>> Hi Josh,
>>>
>>> That warning message was added as part of FLINK-4514. It pops out
>>> whenever a shard iterator was used after 5 minutes it was returned from
>>> Kinesis.
>>> The only time spent between after a shard iterator was returned and
>>> before it was used to fetch the next batch of records, is on deserializing
>>> and emitting of the records of the last fetched batch.
>>> So unless processing of the last fetched batch took over 5 minutes, this
>>> normally shouldn’t happen.
>>>
>>> Have you noticed any sign of long, constant full GC for your Flink task
>>> managers? From your description and check in code, the only possible guess
>>> I can come up with now is that
>>> the source tasks completely seized to be running for a period of time,
>>> and when it came back, the shard iterator was unexpectedly found to be
>>> expired. According to the graph you attached,
>>> when the iterator was refreshed and tasks successfully fetched a few
>>> more batches, the source tasks again halted, and so on.
>>> So you should see that same warning message right before every small
>>> peak within the graph.
>>>
>>> Best Regards,
>>> Gordon
>>>
>>>
>>> On November 3, 2016 at 7:46:42 PM, Josh (jofo90@gmail.com) wrote:
>>>
>>> Hey Gordon,
>>>
>>> I've been using Flink 1.2-SNAPSHOT for the past week (with FLINK-4514)
>>> with no problems, but yesterday the Kinesis consumer started behaving
>>> strangely... My Kinesis data stream is fairly constant at around 1.5MB/sec,
>>> however the Flink Kinesis consumer started to stop consuming for periods of
>>> time (see the spikes in graph attached which shows data consumed by the
>>> Flink Kinesis consumer)
>>>
>>> Looking in the task manager logs, there are no exceptions however there
>>> is this log message which I believe is related to the problem:
>>>
>>> 2016-11-03 09:27:53,782 WARN  org.apache.flink.streaming.co
>>> nnectors.kinesis.internals.ShardConsumer  - Encountered an unexpected
>>> expired iterator AAAAAAAAAAF8OJyh+X3yBnbtzUgIfXv+phS7PK
>>> ppd7q09/tduXG3lOhCmBGPUOlZul24tzSSM6KjHsQ+AbZY8MThKc
>>> SvGax/EoOIYoTELYbZmuwY4hgeqUsndxLIM0HL55iejroBV8YFmUmGwHsW8q
>>> kHsz//Ci4cxcLrGArHex3n+4E+aoZ9AtgTPEZOBjXY49g+VGsDb0bQN5FJUoUVEfnbupk96ore
>>> for shard KinesisStreamShard{streamName='stream001', shard='{ShardId:
>>> shardId-000000000000,HashKeyRange: {StartingHashKey: 0,EndingHashKey:
>>> 85070511730234615865841151857942042863},SequenceNumberRange:
>>> {StartingSequenceNumber: 495665429169236488921642479266
>>> 79091159472198219567464450,}}'}; refreshing the iterator ...
>>>
>>> Having restarted the job from my last savepoint, it's consuming the
>>> stream fine again with no problems.
>>>
>>> Do you have any idea what might be causing this, or anything I should do
>>> to investigate further?
>>>
>>> Cheers,
>>>
>>> Josh
>>>
>>> On Wed, Oct 5, 2016 at 4:55 AM, Tzu-Li (Gordon) Tai <tzulitai@apache.org
>>> > wrote:
>>>
>>>> Hi Steffen,
>>>>
>>>> Turns out that FLINK-4514 just missed Flink 1.1.2 and wasn’t included
>>>> in the release (I’ll update the resolve version in JIRA to 1.1.3, thanks
>>>> for noticing this!).
>>>> The Flink community is going to release 1.1.3 asap, which will include
>>>> the fix.
>>>> If you don’t want to wait for the release and want to try the fix now,
>>>> you can also build on the current “release-1.1” branch, which already
has
>>>> FLINK-4514 merged.
>>>> Sorry for the inconvenience. Let me know if you bump into any other
>>>> problems afterwards.
>>>>
>>>> Best Regards,
>>>> Gordon
>>>>
>>>>
>>>> On October 5, 2016 at 2:56:21 AM, Steffen Hausmann (
>>>> steffen@hausmann-family.de) wrote:
>>>>
>>>> Hi there,
>>>>
>>>> I'm running a Flink 1.1.2 job on EMR and Yarn that is reading events
>>>> from a Kinesis stream. However, after a while (the exact duration varies
>>>> and is in the order of minutes) the Kinesis source doesn't emit any
>>>> further events and hence Flink doesn't produce any further output.
>>>> Eventually, an ExpiredIteratorException occurs in one of the task,
>>>> causing the entire job to fail:
>>>>
>>>> > com.amazonaws.services.kinesis.model.ExpiredIteratorException:
>>>> Iterator expired. The iterator was created at time Mon Oct 03 18:40:30 UTC
>>>> 2016 while right now it is Mon Oct 03 18:45:33 UTC 2016 which is further
in
>>>> the future than the tolerated delay of 300000 milliseconds. (Service:
>>>> AmazonKinesis; Status Code: 400; Error Code: ExpiredIteratorException;
>>>> Request ID: dace9532-9031-54bc-8aa2-3cbfb136d590)
>>>>
>>>> This seems to be related to FLINK-4514, which is marked as resovled for
>>>> Flink 1.1.2. In contrast to what is describe in the ticket, the job I'm
>>>> running isn't suspended but hangs just a few minutes after the job has
>>>> been started.
>>>>
>>>> I've attached a log file showing the described behavior.
>>>>
>>>> Any idea what may be wrong?
>>>>
>>>> Thanks,
>>>> Steffen
>>>>
>>>>
>>>
>>
>

Mime
View raw message