MaxConnections understanding on FlinkKinesisProducer via KPL

[Vijay Balakrishnan]

Hi,
Trying to tune the KPL and FlinkKinesisProducer for Kinesis Data stream(KDS).
Getting following errors:
1.
Throttling
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:727)
 org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:612)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530)
at org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.producer.KinesisProducer.addUserRecord(KinesisProducer.java:536)

2. ERROR org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.producer.LogInputStreamReader  - [2020-06-18 15:49:24.238655] [0x00000ed6][0x00007fc2086c8700] [error] [shard_map.cc:150] Shard map update for stream "...._write" failed. Code: LimitExceededException Message: Rate exceeded for stream ..._write under account 753274046439.; retrying in 1500 ms

3. [AWS Log: ERROR](CurlHttpClient)Curl returned error code 28


https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kinesis.html#backpressure

https://github.com/awslabs/amazon-kinesis-producer/blob/master/java/amazon-kinesis-producer-sample/default_config.properties

https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-flink-timeout/

These are the KPL property changes I am planning to make.
 
RequestTimeput: 10000 //default 6000 ms

AggregationEnabled: true //default is true

ThreadPoolSize: 15 //default 10

MaxConnections: 48 //default 24 - this might have been a bottleneck when we flooded KPL with requests. Requests are sent in parallel over multiple connections to the backend.

RecordTtl: 10000 //default 30000 ms  - drop record after 10s.

FailIfThrottled: true //default false - so if throttled, don't retry.

 
We were using parallelism for sinks at 80. So each corresponds to 1 FlinkKinesisProducer. So, 80 * 10(ThreadPoolSize) = 800 threads. MaxConnections is 24 from KPL. 

I am not sure about the MaxConnections setting - what does 48 mean here -is it 40(sink parallelism) * 15(ThreadPoolSize) * 48 calls to the KDS backend via KPL ?

Any thoughts on how not to overwhelm KPL while handling real time streaming load to the Kinesis via the FlinkKinesisProducer ?

TIA,

[Tzu-Li (Gordon) Tai]

Hi Vijay,

I'm not entirely sure of the semantics between ThreadPoolSize and MaxConnections since they are all KPL configurations (this specific question would probably be better directed to AWS),
but my guess would be that the number of concurrent requests to the KPL backend is capped by MaxConnections. This is per parallel FlinkKinesisProducer subtask.

As for ThreadPoolSize, do note that the default threading model by KPL is PER_REQUEST, for which the KPL native process will launch a thread for each request.
Under heavy load, this would of course be an issue. Since you didn't explicitly mention this config, make sure to set this to POOLED to actually make use of a fixed thread pool for requests.

Overall, my suggestion is to set a reasonable queue limit for the number of records buffered by KPL's native process (by default it is unbounded).
Without that in place, under high load you would easily be resource exhausted, and can cause more unpredictable checkpointing times since the FlinkKinesisProducer would need to flush pending records on checkpoints (which ultimately also applies backpressure upstream).

BR,
Gordon

On Wed, Jul 22, 2020 at 5:21 AM Vijay Balakrishnan <[hidden email]> wrote:

Hi,
Trying to tune the KPL and FlinkKinesisProducer for Kinesis Data stream(KDS).
Getting following errors:
1.
Throttling
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:727)
 org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:612)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530)
at org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.producer.KinesisProducer.addUserRecord(KinesisProducer.java:536)

2. ERROR org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.producer.LogInputStreamReader  - [2020-06-18 15:49:24.238655] [0x00000ed6][0x00007fc2086c8700] [error] [shard_map.cc:150] Shard map update for stream "...._write" failed. Code: LimitExceededException Message: Rate exceeded for stream ..._write under account 753274046439.; retrying in 1500 ms

3. [AWS Log: ERROR](CurlHttpClient)Curl returned error code 28


https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kinesis.html#backpressure

https://github.com/awslabs/amazon-kinesis-producer/blob/master/java/amazon-kinesis-producer-sample/default_config.properties

https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-flink-timeout/

These are the KPL property changes I am planning to make.
 
RequestTimeput: 10000 //default 6000 ms

AggregationEnabled: true //default is true

ThreadPoolSize: 15 //default 10

MaxConnections: 48 //default 24 - this might have been a bottleneck when we flooded KPL with requests. Requests are sent in parallel over multiple connections to the backend.

RecordTtl: 10000 //default 30000 ms  - drop record after 10s.

FailIfThrottled: true //default false - so if throttled, don't retry.

 
We were using parallelism for sinks at 80. So each corresponds to 1 FlinkKinesisProducer. So, 80 * 10(ThreadPoolSize) = 800 threads. MaxConnections is 24 from KPL. 

I am not sure about the MaxConnections setting - what does 48 mean here -is it 40(sink parallelism) * 15(ThreadPoolSize) * 48 calls to the KDS backend via KPL ?

Any thoughts on how not to overwhelm KPL while handling real time streaming load to the Kinesis via the FlinkKinesisProducer ?

TIA,

[Vijay Balakrishnan]

Hi Gordon,

Thx for your reply.

FlinkKinesisProducer default is ThreadPool which is what I am using. So, does that mean only 10 threads are making calls to KDS by default ??

I see from the number of records coming to the KDS that I need only 1-2 shards. So, the bottleneck is on the KPL side.

Does this mean I have to set a QueueLimit of 500 as shown in the example below ??

From what you said, Total MaxConnections would then be by default: 24 * number of subtasks = 24 * 80 = 1920 connections to KDS.

KPL ThreadPoolSize would be 10 Threads by default - is this per subtask ? So, would it be 10 * number of subtasks = 10 * 80 = 800 Threads ??

I am trying to reconcile the diff above ? Somewhere I am flooding KPL with too many requests & it gives the curl 28 error.

So, calculating Queue Limit:

Based on this, my records size = 1600 bytes. I have 96 shards
Assuming - With the default RecordMaxBufferedTime of 100ms, a queue size of 100kB per shard should be sufficient.So, Queue size/shard=100KB
Queue Limit with 96 shards = (96 * 10^5)/ 1600 = 6000
Queue Limit with 4 shards = (4 * 10^5)/ 1600 = 0.25

Acc. to the docs:

By default, FlinkKinesisProducer does not backpressure. Instead, records that cannot be sent because of the rate restriction of 1 MB per second per shard are buffered in an unbounded queue and dropped when their RecordTtl expires.

To avoid data loss, you can enable backpressuring by restricting the size of the internal queue:

// 200 Bytes per record, 1 shard
kinesis.setQueueLimit(500);

On Tue, Jul 21, 2020 at 8:00 PM Tzu-Li (Gordon) Tai <[hidden email]> wrote:

Hi Vijay,

I'm not entirely sure of the semantics between ThreadPoolSize and MaxConnections since they are all KPL configurations (this specific question would probably be better directed to AWS),
but my guess would be that the number of concurrent requests to the KPL backend is capped by MaxConnections. This is per parallel FlinkKinesisProducer subtask.

As for ThreadPoolSize, do note that the default threading model by KPL is PER_REQUEST, for which the KPL native process will launch a thread for each request.
Under heavy load, this would of course be an issue. Since you didn't explicitly mention this config, make sure to set this to POOLED to actually make use of a fixed thread pool for requests.

Overall, my suggestion is to set a reasonable queue limit for the number of records buffered by KPL's native process (by default it is unbounded).
Without that in place, under high load you would easily be resource exhausted, and can cause more unpredictable checkpointing times since the FlinkKinesisProducer would need to flush pending records on checkpoints (which ultimately also applies backpressure upstream).

BR,
Gordon

On Wed, Jul 22, 2020 at 5:21 AM Vijay Balakrishnan <[hidden email]> wrote:

Hi,
Trying to tune the KPL and FlinkKinesisProducer for Kinesis Data stream(KDS).
Getting following errors:
1.
Throttling
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:727)
 org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:612)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530)
at org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.producer.KinesisProducer.addUserRecord(KinesisProducer.java:536)

2. ERROR org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.producer.LogInputStreamReader  - [2020-06-18 15:49:24.238655] [0x00000ed6][0x00007fc2086c8700] [error] [shard_map.cc:150] Shard map update for stream "...._write" failed. Code: LimitExceededException Message: Rate exceeded for stream ..._write under account 753274046439.; retrying in 1500 ms

3. [AWS Log: ERROR](CurlHttpClient)Curl returned error code 28


https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kinesis.html#backpressure

https://github.com/awslabs/amazon-kinesis-producer/blob/master/java/amazon-kinesis-producer-sample/default_config.properties

https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-flink-timeout/

These are the KPL property changes I am planning to make.
 
RequestTimeput: 10000 //default 6000 ms

AggregationEnabled: true //default is true

ThreadPoolSize: 15 //default 10

MaxConnections: 48 //default 24 - this might have been a bottleneck when we flooded KPL with requests. Requests are sent in parallel over multiple connections to the backend.

RecordTtl: 10000 //default 30000 ms  - drop record after 10s.

FailIfThrottled: true //default false - so if throttled, don't retry.

 
We were using parallelism for sinks at 80. So each corresponds to 1 FlinkKinesisProducer. So, 80 * 10(ThreadPoolSize) = 800 threads. MaxConnections is 24 from KPL. 

I am not sure about the MaxConnections setting - what does 48 mean here -is it 40(sink parallelism) * 15(ThreadPoolSize) * 48 calls to the KDS backend via KPL ?

Any thoughts on how not to overwhelm KPL while handling real time streaming load to the Kinesis via the FlinkKinesisProducer ?

TIA,

[Tzu-Li (Gordon) Tai]

ThreadPoolSize is per KPL instance, so yes that is per subtask.
As I previously mentioned, the maximum concurrent requests going to KDS would be capped by MaxConnections.

On Thu, Jul 23, 2020 at 6:25 AM Vijay Balakrishnan <[hidden email]> wrote:

Hi Gordon,

Thx for your reply.

FlinkKinesisProducer default is ThreadPool which is what I am using. So, does that mean only 10 threads are making calls to KDS by default ??

I see from the number of records coming to the KDS that I need only 1-2 shards. So, the bottleneck is on the KPL side.

Does this mean I have to set a QueueLimit of 500 as shown in the example below ??

From what you said, Total MaxConnections would then be by default: 24 * number of subtasks = 24 * 80 = 1920 connections to KDS.

KPL ThreadPoolSize would be 10 Threads by default - is this per subtask ? So, would it be 10 * number of subtasks = 10 * 80 = 800 Threads ??

I am trying to reconcile the diff above ? Somewhere I am flooding KPL with too many requests & it gives the curl 28 error.

So, calculating Queue Limit:

Based on this, my records size = 1600 bytes. I have 96 shards
Assuming - With the default RecordMaxBufferedTime of 100ms, a queue size of 100kB per shard should be sufficient.So, Queue size/shard=100KB
Queue Limit with 96 shards = (96 * 10^5)/ 1600 = 6000
Queue Limit with 4 shards = (4 * 10^5)/ 1600 = 0.25

Acc. to the docs:

By default, FlinkKinesisProducer does not backpressure. Instead, records that cannot be sent because of the rate restriction of 1 MB per second per shard are buffered in an unbounded queue and dropped when their RecordTtl expires.

To avoid data loss, you can enable backpressuring by restricting the size of the internal queue:

// 200 Bytes per record, 1 shard
kinesis.setQueueLimit(500);

On Tue, Jul 21, 2020 at 8:00 PM Tzu-Li (Gordon) Tai <[hidden email]> wrote:

Hi Vijay,

I'm not entirely sure of the semantics between ThreadPoolSize and MaxConnections since they are all KPL configurations (this specific question would probably be better directed to AWS),
but my guess would be that the number of concurrent requests to the KPL backend is capped by MaxConnections. This is per parallel FlinkKinesisProducer subtask.

As for ThreadPoolSize, do note that the default threading model by KPL is PER_REQUEST, for which the KPL native process will launch a thread for each request.
Under heavy load, this would of course be an issue. Since you didn't explicitly mention this config, make sure to set this to POOLED to actually make use of a fixed thread pool for requests.

Overall, my suggestion is to set a reasonable queue limit for the number of records buffered by KPL's native process (by default it is unbounded).
Without that in place, under high load you would easily be resource exhausted, and can cause more unpredictable checkpointing times since the FlinkKinesisProducer would need to flush pending records on checkpoints (which ultimately also applies backpressure upstream).

BR,
Gordon

On Wed, Jul 22, 2020 at 5:21 AM Vijay Balakrishnan <[hidden email]> wrote:

Hi,
Trying to tune the KPL and FlinkKinesisProducer for Kinesis Data stream(KDS).
Getting following errors:
1.
Throttling
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:727)
 org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:612)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530)
at org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.producer.KinesisProducer.addUserRecord(KinesisProducer.java:536)

2. ERROR org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.producer.LogInputStreamReader  - [2020-06-18 15:49:24.238655] [0x00000ed6][0x00007fc2086c8700] [error] [shard_map.cc:150] Shard map update for stream "...._write" failed. Code: LimitExceededException Message: Rate exceeded for stream ..._write under account 753274046439.; retrying in 1500 ms

3. [AWS Log: ERROR](CurlHttpClient)Curl returned error code 28


https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kinesis.html#backpressure

https://github.com/awslabs/amazon-kinesis-producer/blob/master/java/amazon-kinesis-producer-sample/default_config.properties

https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-flink-timeout/

These are the KPL property changes I am planning to make.
 
RequestTimeput: 10000 //default 6000 ms

AggregationEnabled: true //default is true

ThreadPoolSize: 15 //default 10

MaxConnections: 48 //default 24 - this might have been a bottleneck when we flooded KPL with requests. Requests are sent in parallel over multiple connections to the backend.

RecordTtl: 10000 //default 30000 ms  - drop record after 10s.

FailIfThrottled: true //default false - so if throttled, don't retry.

 
We were using parallelism for sinks at 80. So each corresponds to 1 FlinkKinesisProducer. So, 80 * 10(ThreadPoolSize) = 800 threads. MaxConnections is 24 from KPL. 

I am not sure about the MaxConnections setting - what does 48 mean here -is it 40(sink parallelism) * 15(ThreadPoolSize) * 48 calls to the KDS backend via KPL ?

Any thoughts on how not to overwhelm KPL while handling real time streaming load to the Kinesis via the FlinkKinesisProducer ?

TIA,