Kinesis ProvisionedThroughputExceededException

[M Singh]

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+M4NvUUtmJ2OxuBCw@...%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

[Roman Grebennikov]

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 | [hidden email]

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+M4NvUUtmJ2OxuBCw@...%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

[M Singh]

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 <[hidden email]> 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 | [hidden email]

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+M4NvUUtmJ2OxuBCw@...%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

[Roman Grebennikov]

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 | [hidden email]

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 <[hidden email]> 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 | [hidden email]

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+M4NvUUtmJ2OxuBCw@...%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

[M Singh]

Thanks Roman for your response.  Mans

On Wednesday, June 17, 2020, 03:26:31 AM EDT, Roman Grebennikov <[hidden email]> wrote:

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 | [hidden email]

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 <[hidden email]> 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 | [hidden email]

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+M4NvUUtmJ2OxuBCw@...%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