Poorvankbhatia commented on code in PR #26274:
URL: https://github.com/apache/flink/pull/26274#discussion_r1991889909
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java:
##########
@@ -101,19 +101,25 @@ public abstract class AsyncSinkWriter<InputT,
RequestEntryT extends Serializable
/**
* Buffer to hold request entries that should be persisted into the
destination, along with its
- * size in bytes.
+ * total size in bytes.
*
- * <p>A request entry contain all relevant details to make a call to the
destination. Eg, for
- * Kinesis Data Streams a request entry contains the payload and partition
key.
+ * <p>This buffer is managed using {@link BufferWrapper}, allowing sink
implementations to
+ * define their own optimized buffering strategies. By default, {@link
DequeBufferWrapper} is
+ * used.
*
- * <p>It seems more natural to buffer InputT, ie, the events that should
be persisted, rather
- * than RequestEntryT. However, in practice, the response of a failed
request call can make it
- * very hard, if not impossible, to reconstruct the original event. It is
much easier, to just
- * 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
Review Comment:
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](https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java#L391)
--
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]
