[FLINK-37298] Added Pluggable Components for BatchStrategy & BufferWrapper in AsyncSinkWriter. #26274

Poorvankbhatia · 2025-03-08T15:57:21Z

What is the purpose of the change

FLIP-509

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> with a pluggable buffer wrapper, allowing for alternative data structures that optimize lookup, prioritization, or performance.

Brief change log

Added BufferWrapper interface to allow pluggable buffering strategies instead of a fixed Deque.
Implemented DequeBufferWrapper as the default buffer wrapper using ArrayDeque.
Introduced BatchCreator interface to enable custom batching logic.
Added SimpleBatchCreator as the default batch creator, replicating existing batch formation logic.
Created BatchCreationResult class to encapsulate batch metadata (entries, size, record count).
Modified AsyncSinkWriter 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

flinkbot · 2025-03-08T16:14:27Z

CI report:

f3c3386 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

vahmed-hamdy

Overall looks good, thanks for delivering this. Just a couple of questions and nits.

vahmed-hamdy · 2025-03-11T10:48:06Z

...rs/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/Batch.java

+ *
+ * @param <RequestEntryT> the type of request entry in this batch
+ */
+@PublicEvolving


I am not sure if this fits better as @Internal or PublicEvolving? It is not intended to be exposed to public users

Hey @vahmed-hamdy , So Batch is created by BatchCreator interface, which can be inherited and implemented by the Sink Implementor. So i was under the impression that anything that external developers are expected to use should NOT be @Internal as they might change across releases. Flink sink developers might create custom implementations of BatchCreator or BufferWrapper for better performance or sink-specific logic. But please correct me if i am wrong. Will change it if required.

yeah I don't have a strong opinion here, I also see classes like AsyncSinkWriterConfiguration and RateLimitingStrategy marked PublicEvolving for the same reason. let's keep it then

vahmed-hamdy · 2025-03-11T10:48:20Z

...k-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/BatchCreator.java

+ *
+ * @param <RequestEntryT> the type of the request entries to be batched
+ */
+@PublicEvolving


Same as Batch

Replied above.

vahmed-hamdy · 2025-03-11T10:51:17Z

...-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/BufferWrapper.java

+ *
+ * @param <RequestEntryT> The type of request entries being buffered.
+ */
+@PublicEvolving


Same as above

Replied above.

vahmed-hamdy · 2025-03-11T10:57:44Z

...ctor-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterTest.java

+        AsyncSinkWriterImpl initialSinkWriter =
+                new AsyncSinkWriterImplBuilder().context(sinkInitContext).build();
+
+        assertThat(initialSinkWriter.getBufferedRequestEntries())


We don't test using default when passing nulls?

Added null BatchCreator & BatchWrapper to AsyncSinkWriterImpl in the test case.

vahmed-hamdy · 2025-03-11T11:00:56Z

...ector-base/src/main/java/org/apache/flink/connector/base/sink/writer/SimpleBatchCreator.java

+     * @param maxBatchSizeInBytes the maximum cumulative size in bytes allowed for any single batch
+     */
+    private SimpleBatchCreator(long maxBatchSizeInBytes) {
+        this.maxBatchSizeInBytes = maxBatchSizeInBytes;


nit:

Suggested change

this.maxBatchSizeInBytes = maxBatchSizeInBytes;

this.maxBatchSizeInBytes = Preconditions.checkArgument(maxBatchSizeInBytes > 0);

Also can test that

Added this and a corresponding test case as well.

vahmed-hamdy

Looks Good, thanks for the huge efforts

...-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/BufferWrapper.java

davidradl · 2025-03-12T16:05:31Z

...onnector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java

-     * construct a new (retry) request entry from the response and add that back to the queue for
-     * later retry.
+     * <p>The buffer stores {@link RequestEntryWrapper} objects rather than raw {@link
+     * RequestEntryT} instances, as buffering wrapped entries allows for better tracking of size and


I am not seeing the main code issuing retry anywhere ie. add(entry, true) .

The retry logic is handled inside the completeRequest() method.

Specifically, when an entry fails, it is re-added at the head of the buffer using:

while (iterator.hasPrevious()) { addEntryToBuffer(iterator.previous(), true); }

This ensures failed entries are retried on the next flush.
Let me know if you’d like any further clarification! This part of the code has not been changed and it remains the same. Reference

...onnector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java

davidradl · 2025-03-12T16:16:54Z

could we have the docs change to publicise this change and advice on supplying pluggable strategies and BufferWrapper.

Poorvankbhatia · 2025-03-12T16:38:41Z

could we have the docs change to publicise this change and advice on supplying pluggable strategies and BufferWrapper.

Thanks for the suggestion! The JavaDocs for BatchCreator and BufferWrapper already document their purpose and usage.

Regarding additional documentation changes since there's an ongoing vote on this FLIP, would it make sense to wait until finalized and then follow up with documentation updates in a separate PR?

Let me know what you think—happy to adjust based on the preferred approach! :)

Poorvankbhatia · 2025-03-18T06:13:35Z

Hey @dannycranmer can you help with this review?

[FLINK-37298] Add custom batch handling in AsyncSinkWriter.

06749e6

flinkbot added the component=Connectors/Common label Mar 8, 2025

License added for new files.

bb90ed4

vahmed-hamdy suggested changes Mar 11, 2025

View reviewed changes

Review Comments Incorporated

96550f2

vahmed-hamdy approved these changes Mar 11, 2025

View reviewed changes

davidradl reviewed Mar 12, 2025

View reviewed changes

...-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/BufferWrapper.java Show resolved Hide resolved

davidradl reviewed Mar 12, 2025

View reviewed changes

...-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/BufferWrapper.java Outdated Show resolved Hide resolved

Corrected comments based on review.

c96de8c

davidradl reviewed Mar 12, 2025

View reviewed changes

...onnector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java Outdated Show resolved Hide resolved

Comment correction

49ceaa7

Spotless check done

f3c3386

Poorvankbhatia force-pushed the FLINK-37298-async_custom_batch branch from 8b97772 to f3c3386 Compare March 13, 2025 09:17

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-37298] Added Pluggable Components for BatchStrategy & BufferWrapper in AsyncSinkWriter. #26274

[FLINK-37298] Added Pluggable Components for BatchStrategy & BufferWrapper in AsyncSinkWriter. #26274

Poorvankbhatia commented Mar 8, 2025

flinkbot commented Mar 8, 2025 •

edited

Loading

vahmed-hamdy left a comment

vahmed-hamdy Mar 11, 2025 •

edited

Loading

Poorvankbhatia Mar 11, 2025

vahmed-hamdy Mar 11, 2025

vahmed-hamdy Mar 11, 2025

Poorvankbhatia Mar 11, 2025

vahmed-hamdy Mar 11, 2025

Poorvankbhatia Mar 11, 2025

vahmed-hamdy Mar 11, 2025

Poorvankbhatia Mar 11, 2025

vahmed-hamdy Mar 11, 2025

Poorvankbhatia Mar 11, 2025

vahmed-hamdy left a comment

davidradl Mar 12, 2025 •

edited

Loading

Poorvankbhatia Mar 12, 2025

davidradl commented Mar 12, 2025

Poorvankbhatia commented Mar 12, 2025

Poorvankbhatia commented Mar 18, 2025

	this.maxBatchSizeInBytes = maxBatchSizeInBytes;
	this.maxBatchSizeInBytes = Preconditions.checkArgument(maxBatchSizeInBytes > 0);

[FLINK-37298] Added Pluggable Components for BatchStrategy & BufferWrapper in AsyncSinkWriter. #26274

Are you sure you want to change the base?

[FLINK-37298] Added Pluggable Components for BatchStrategy & BufferWrapper in AsyncSinkWriter. #26274

Conversation

Poorvankbhatia commented Mar 8, 2025

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Mar 8, 2025 • edited Loading

CI report:

vahmed-hamdy left a comment

Choose a reason for hiding this comment

vahmed-hamdy Mar 11, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vahmed-hamdy left a comment

Choose a reason for hiding this comment

davidradl Mar 12, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidradl commented Mar 12, 2025

Poorvankbhatia commented Mar 12, 2025

Poorvankbhatia commented Mar 18, 2025

flinkbot commented Mar 8, 2025 •

edited

Loading

vahmed-hamdy Mar 11, 2025 •

edited

Loading

davidradl Mar 12, 2025 •

edited

Loading