[FLINK-24041] [FLIP-171] Generic AsyncSinkBase

• Type: New Feature
• Status: Resolved
• Priority: Major
• Resolution: Done
• Fix Version/s: 1.15.0
• Component/s: Connectors / Common
• Labels: pull-request-available
• Assignee: Zichen Liu
• Reporter: Zichen Liu
• Created: 29/Aug/21 16:38
• Updated: 22/Dec/22 18:31
• Resolved: 14/Feb/22 13:06

Description

Motivation

Apache Flink has a rich connector ecosystem that can persist data in various destinations. Flink natively supports Apache Kafka, Amazon Kinesis Data Streams, Elasticsearch, HBase, and many more destinations. Additional connectors are maintained in Apache Bahir or directly on GitHub. The basic functionality of these sinks is quite similar. They batch events according to user defined buffering hints, sign requests and send them to the respective endpoint, retry unsuccessful or throttled requests, and participate in checkpointing. They primarily just differ in the way they interface with the destination. Yet, all the above-mentioned sinks are developed and maintained independently.

We hence propose to create a sink that abstracts away this common functionality into a generic sink. Adding support for a new destination then just means creating a lightweight shim that only implements the specific interfaces of the destination using a client that supports async requests. Having a common abstraction will reduce the effort required to maintain all these individual sinks. It will also make it much easier and faster to create integrations with additional destinations. Moreover, improvements or bug fixes to the core of the sink will benefit all implementations that are based on it.

The design of the sink focusses on extensibility and a broad support of destinations. The core of the sink is kept generic and free of any connector specific dependencies. The sink is designed to participate in checkpointing to provide at-least once semantics, but it is limited to destinations that provide a client that supports async requests. 

References

More details to be found https://cwiki.apache.org/confluence/display/FLINK/FLIP-171%3A+Async+Sink

Issue Links

• causes: FLINK-25846 [FLIP-171] Async Sink does not gracefully shutdown on Cancel
• causes: FLINK-25792 Async Sink Base is too being flushed too frequently resulting in backpressure even when buffer is near empty
• causes: FLINK-25811 [FLIP-171] Fix generic AsyncSinkWriter retrying requests in reverse order
• Dependent: FLINK-24229 DynamoDB implementation of Async Sink
• Dependent: FLINK-24234 [FLIP-171] Byte Based & Time Based Flushing for AsyncSinkBase
• Dependent: FLINK-24370 [FLIP-171] Documentation for Generic AsyncSinkBase
• Dependent: FLINK-25610 [FLIP-171] Kinesis Firehose implementation of Async Sink Table API
• Dependent: FLINK-24227 [FLIP-171] KDS implementation of Async Sink
• Dependent: FLINK-24228 [FLIP-171] Firehose implementation of Async Sink
• Dependent: FLINK-24905 [FLIP-171] KDS implementation of Async Sink Table API
• Dependent: FLINK-30488 OpenSearch implementation of Async Sink
• links to: GitHub Pull Request #17068
• links to: GitHub Pull Request #17213
• links to: GitHub Pull Request #17244
• links to: GitHub Pull Request #18483
• links to: GitHub Pull Request #18488