vahmed-hamdy opened a new pull request #18013: URL: https://github.com/apache/flink/pull/18013
## What is the purpose of the change User stories: As a Flink user, I’d like to use Kinesis Data Streams as sink for my data pipeline. Context: The AsyncSinkWriter currently has a static batch size. We’d like to allow Sink users to specify what batch size they would like to use as they are in a better place to choose a batching logic/size for their needs. We’d like to also allow Sink developers to make use of the provided configuration mechanism for their Sink implementations. ### Scope: Implement an asynchronous sink for Kinesis Data Streams (KDS) by inheriting the AsyncSinkBase class. The implementation can for now reside in its own module in flink-connectors. The module and package name can be anything reasonable e.g. flink-connector-aws-kinesis for the module name and org.apache.flink.connector.aws.kinesis for the package name. Side-note: There will be additional work later to move these implementations somewhere else (see theongoing discussion) (http://mail-archives.apache.org/mod_mbox/flink-dev/202106.mbox/%3ccagznd0ebmgud327_j4gvdyaoygaewxmjz9kzn33fv0v+j8g...@mail.gmail.com%3e). The implementation must use the Kinesis Java Client (https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/kinesis/AmazonKinesisClient.html). The implementation must allow users to configure the Kinesis Client, with reasonable default settings. Implement an asynchornous sink writer for KDS by extending the AsyncSinkWriter. The implementation must deal with failed requests and retry them using the requeueFailedRequestEntry method. If possible, the implementation should batch multiple requests (PutRecordsRequestEntry objects) to KDS for increased throughput. The implemented Sink Writer will be used by the Sink class that will be created as part of this story. Unit/Integration testing. Use Kinesalite (in-memory Kinesis simulation). We already use this in KinesisTableApiITCase. Java / code-level docs. ### Brief change log Added Kinesis Data Streams Sink into a new module flink-connectors/flink-connector-kinesis-async Edit: flink-connectors/flink-connector-kinesis-async is now flink-connectors/flink-connector-aws (as of 2021-10-01) Verifying this change This change added tests and can be verified as follows: Added integration tests set up with Kinesalite Does this pull request potentially affect one of the following parts: Dependencies (does it add or upgrade a dependency): no The public API, i.e., is any changed class annotated with @Public(Evolving): no The serializers: no The runtime per-record code paths (performance sensitive): yes Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no The S3 file system connector: no Documentation Does this pull request introduce a new feature? yes If yes, how is the feature documented? JavaDocs -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
