[
https://issues.apache.org/jira/browse/FLINK-37918?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18100332#comment-18100332
]
Mirai Kato commented on FLINK-37918:
------------------------------------
I would like to share my findings and a proposal.
h3. Request to add Affects Version/s
Please also add the following versions to the "Affects Version/s" field:
# flink-connector-aws-kinesis-streams 5.1.0
# flink-connector-aws-kinesis-streams 6.0.1
In my environment, I used flink-connector-aws-kinesis-streams 5.1.0-1.20
(READER_TYPE: POLLING, READER_EMPTY_RECORDS_FETCH_INTERVAL: 1000 ms).
With this configuration, the CloudWatch ReadProvisionedThroughputExceeded
metric remained above 0 continuously.
Approximately 10% of the GetRecords requests resulted in
ReadProvisionedThroughputExceeded.
The fix in FLINK-36947 reduced the number of GetRecords calls.
It applies an interval only when a GetRecords call returns 0 records.
However, in my environment, this fix does not fully resolve the problem.
My understanding is that this problem can happen when records arrive
continuously at a very high rate (for example, one record every 1 ms).
In this case, the interval is not applied, because the GetRecords call returns
records.
I also reviewed the source code of flink-connector-aws-kinesis-streams 6.0.1
and the current main branch.
In the current implementation, the interval is still applied only when the
returned record count is 0.
Therefore, it appears that the same behavior may also exist in these versions.
If it is better to track this issue separately, I can create a new Jira issue.
h3. Proposal for fix
I would like to propose applying the GetRecords interval at a fixed rate,
whether the returned records are empty or not.
Users should also be able to set the interval value in their configuration.
To check this idea, I edited the connector code so that it applies the interval
between GetRecords requests regardless of the number of records returned, built
a library from it, and used it in my environment.
After this change, ReadProvisionedThroughputExceeded no longer stayed above 0
continuously.
For the default value, FLINK-6365 discussed the GetRecords interval for the AWS
SDK for Java v1 connector (flink-connector-kinesis).
It set the default of SHARD_GETRECORDS_INTERVAL_MILLIS to 200 ms.
Based on that discussion, 200 ms might be a reasonable default value for the
GetRecords interval.
Thank you for considering this proposal.
> Restore the ability to set an interval for GetRecords calls to Kinesis shards.
> ------------------------------------------------------------------------------
>
> Key: FLINK-37918
> URL: https://issues.apache.org/jira/browse/FLINK-37918
> Project: Flink
> Issue Type: Improvement
> Components: Connectors / Kinesis
> Affects Versions: aws-connector-5.0.0
> Reporter: hajimeni
> Priority: Major
> Labels: pull-request-available
>
> h3. Background
> The previous Flink Kinesis connector (flink-connector-kinesis) provided a
> configuration parameter, SHARD_GETRECORDS_INTERVAL_MILLIS, which allowed
> users to set a specific interval between GetRecords calls for each shard.
> This functionality is absent in the new AWS Kinesis Streams connector
> (flink-connector-aws-kinesis-streams).
> h3. Problem
>
> The lack of a configurable interval for GetRecords calls in the new connector
> (KinesisStreamsSource) poses a significant challenge in scenarios with
> multiple consumers reading from the same Kinesis stream. Without the ability
> to increase the interval between GetRecords calls, consumers can easily
> exceed the AWS Kinesis limit of five GetRecords calls per second per shard.
> This leads to several issues: * Wasted API Calls and Increased Costs:
> Continuous, rapid calls that are likely to be throttled are inefficient and
> can lead to increased costs.
> * Operational Instability: In a multi-tenant or multi-application
> environment, the absence of this control makes it difficult to ensure stable
> and predictable data consumption across all consumers.
> * AWS documentation recommends adjusting the frequency of GetRecords calls
> to avoid these issues, especially when multiple consumers are involved. You
> can find this recommendation in the AWS Kinesis Developer Guide. (see:
> https://docs.aws.amazon.com/streams/latest/dev/kinesis-low-latency.html )
> h3. Feature Request
> We request the re-introduction of a configuration option, similar to
> SHARD_GETRECORDS_INTERVAL_MILLIS, in the flink-connector-aws-kinesis-streams
> connector. This would allow users to effectively manage the rate of
> GetRecords calls per shard, thereby preventing API throttling and ensuring
> the stability and efficiency of Flink applications that consume data from
> Kinesis Data Streams.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
