Hi devs. I've created a small improvement ticket last week for the AWS Kinesis connector https://issues.apache.org/jira/browse/FLINK-39137. I'm willing to work on this ticket as my first code contribution. Can I get someone to sanity-check the ticket and assign it to me? Thanks a lot 🙏

for ddb it is implemented, for kinesis is there a usecase for this?

The use case is to allow having multiple consumers of a non-EFO streams. • The Kinesis APIs GetRecords operation has a limit of 5 requests/second for each shard. • The Flink Kinesis connector currently exhausts this API quota GetRecords fully • This means all consumers (the Kinesis connector included) get rate-limited and dashboards fill with warnings • Worst-case scenario is that other, less eager, consumers fall behind the stream, because they don't win the next GetRecords operation We've run into this issue whenever we've tried to connect Flink applications to Kinesis streams shared with other consumers. Kinesis EFO streams have additional costs.

I see. Makes sense then. Feel free to raise a PR

Thanks. I will

@Andreas Bube another question: i know we added a polling delay in kinesis connector. Are you using v5.1+ or v6.0 or v5.0?

Thanks for helping out with this, @Abhi Gupta. I appreciate it. I did see those changes. I built my code on top those. For reference you can see my changes: https://github.com/bube/flink-connector-aws/commit/ed31053c4d74cb588b06492c30d57a4d2c5e93e6. When we ran into the issues with Flink apps starving other consumers (firehoses) we were running 5.0. I believe that having a wait after pulling empty records is not enough solve the issue for active streams/shards. That's why I added an optional and configurable wait after non-empty record pulls. The legacy connectors also had a similar setting: https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/connectors/table/kinesis/#scan-shard-getrecords-intervalmillis

if you have some time today, i would like to get on a quick huddle on why the existing thing doesnt solve for you