Poorvankbhatia opened a new pull request, #26274: URL: https://github.com/apache/flink/pull/26274
## What is the purpose of the change [FLIP-509](https://cwiki.apache.org/confluence/display/FLINK/FLIP-509+Add+pluggable+Batching+for+Async+Sink) * The current AsyncSinkWriter handles batching using fixed size constraints (maxBatchSize & maxBatchSizeInBytes). However, this approach lacks flexibility for more complex batching strategies like grouping records by partition key (important for sinks like Cassandra). * This change introduces a pluggable batching mechanism where sink implementers can define custom batching logic instead of relying on fixed queue-based buffering. * It also decouples the buffer implementation by replacing the direct use of Deque<RequestEntryWrapper<T>> with a pluggable buffer wrapper, allowing for alternative data structures that optimize lookup, prioritization, or performance. ## Brief change log * Added BufferWrapper<T> interface to allow pluggable buffering strategies instead of a fixed Deque. * Implemented DequeBufferWrapper<T> as the default buffer wrapper using ArrayDeque. * Introduced BatchCreator<T> interface to enable custom batching logic. * Added SimpleBatchCreator<T> as the default batch creator, replicating existing batch formation logic. * Created BatchCreationResult<T> class to encapsulate batch metadata (entries, size, record count). * Modified AsyncSinkWriter<T> to incorporate both the pluggable components. ## Verifying this change * Added tests for BufferWrapper and its default implementation (DequeBufferWrapper). * Added tests for BatchCreator to validate the default batching behavior. * Ensured batched requests are correctly formed and processed using the new interfaces. * Verified that the default implementations (SimpleBatchCreator and DequeBufferWrapper) maintain existing behavior without requiring changes from users. ## 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)`: yes - The serializers: no - The runtime per-record code paths (performance sensitive): no - 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]
