Thanks Roman for your response. Mans
On Wednesday, June 17, 2020, 03:26:31 AM EDT, Roman Grebennikov
<[email protected]> wrote:
#yiv4075825537 p.yiv4075825537MsoNormal, #yiv4075825537
p.yiv4075825537MsoNoSpacing{margin:0;}Hi,
It will occur if your job will reach SHARD_GETRECORDS_RETRIES consecutive
failed attempts to pull the data from kinesis.
So if you scale up the topic in kinesis and tune a bit backoff parameters, you
will lower the probability of this exception almost to zero (but with increased
costs and worst-case latency).
But yes, this is a main drawback of managed solutions - as far as you reach a
significant load, you need to pay more. Other managed option within AWS is to
switch to MSK, managed Kafka, which has no such significant restrictions.
And the final option is to wait until FLINK-17688 will be implemented (using
Kinesis enhanced fan-out, so Kinesis will push the data to consumer, instead of
consumer periodically pulling the data).
Roman Grebennikov | [email protected]
On Wed, Jun 17, 2020, at 04:39, M Singh wrote:
Thanks Roman for your response and advice.
>From my understanding increasing shards will increase throughput but still if
>more than 5 requests are made per shard/per second, and since we have 20 apps
>(and increasing) then the exception might occur.
Please let me know if I have missed anything.
Mans
On Tuesday, June 16, 2020, 03:29:59 PM EDT, Roman Grebennikov <[email protected]>
wrote:
Hi,
usually this exception is thrown by aws-java-sdk and means that your kinesis
stream is hitting a throughput limit (what a surprise). We experienced the same
thing when we had a single "event-bus" style stream and multiple flink apps
reading from it.
Each Kinesis partition has a limit of 5 poll operations per second. If you have
a stream with 4 partitions and 30 jobs reading from it, I guess that each job
is constantly hitting op limit for kinesis with default kinesis consumer
settings and it does an exponential back-off (by just sleeping for a small
period of time and then retrying).
You have two options here:
1. scale up the kinesis stream, so there will be more partitions and higher
overall throughput limits
2. tune kinesis consumer backoff parameters:
Our current ones, for example, look like this:
conf.put(ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS, "2000")
// we poll every 2s
conf.put(ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_BASE, "2000") //
in case of throughput error, initial timeout is 2s
conf.put(ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_MAX, "10000") //
we can go up to 10s pause
conf.put(ConsumerConfigConstants.SHARD_GETRECORDS_BACKOFF_EXPONENTIAL_CONSTANT,
"1.5") // multiplying pause to 1.5 on each next step
conf.put(ConsumerConfigConstants.SHARD_GETRECORDS_RETRIES, "100") // and
make up to 100 retries
with best regards,
Roman Grebennikov | [email protected]
On Mon, Jun 15, 2020, at 13:45, M Singh wrote:
Hi:
I am using multiple (almost 30 and growing) Flink streaming applications that
read from the same kinesis stream and get
ProvisionedThroughputExceededException exception which fails the job.
I have seen a reference
http://mail-archives.apache.org/mod_mbox/flink-user/201811.mbox/%3CCAJnSTVxpuOhCNTFTvEYd7Om4s=q2vz5-8+m4nvuutmj2oxu...@mail.gmail.com%3E
- which indicates there might be some solution perhaps in Flink 1.8/1.9.
I also see [FLINK-10536] Flink Kinesis Consumer leading to job failure due to
ProvisionedThroughputExceededException - ASF JIRA is still open.
So i wanted to find out
1. If this issue has been resolved and if so in which version ?
2. Is there any kinesis consumer with kinesis fanout available that can help
address this issue ?
3. Is there any specific parameter in kinesis consumer config that can address
this issue ?
If there is any other pointer/documentation/reference, please let me know.
Thanks
