chamikaramj commented on code in PR #39612:
URL: https://github.com/apache/beam/pull/39612#discussion_r3797700537
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/write/UnboundedBatchedSolaceWriter.java:
##########
@@ -130,12 +139,16 @@ public void finishBundle(FinishBundleContext context)
throws IOException {
}
getCurrentBundle().clear();
- publishResults(BeamContextWrapper.of(context));
+ if (getDeliveryMode() == DeliveryMode.PERSISTENT && messageIdsToAck !=
null) {
+ waitForAcks(BeamContextWrapper.of(context), messageIdsToAck);
Review Comment:
Don't we need to invoke publishResults in this path as well ?
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/write/UnboundedSolaceWriter.java:
##########
@@ -218,6 +230,37 @@ public void publishResults(BeamContextWrapper context) {
}
}
+ public void waitForAcks(BeamContextWrapper context, Set<String>
messageIdsToAck) {
+ BlockingQueue<PublishResult> queue =
+ solaceSessionServiceWithProducer().getPublishedResultsQueue();
+ long timeoutMs = System.currentTimeMillis() + ACKS_FLUSHING_INTERVAL_SECS
* 1000;
Review Comment:
Should we use exponential backoff here and set a larger overall timeout (O
(minutes)) to support potentially large bundles and/or API rate limit issues ?
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/write/UnboundedBatchedSolaceWriter.java:
##########
@@ -130,12 +139,16 @@ public void finishBundle(FinishBundleContext context)
throws IOException {
}
getCurrentBundle().clear();
- publishResults(BeamContextWrapper.of(context));
+ if (getDeliveryMode() == DeliveryMode.PERSISTENT && messageIdsToAck !=
null) {
+ waitForAcks(BeamContextWrapper.of(context), messageIdsToAck);
+ } else {
+ publishResults(BeamContextWrapper.of(context), null);
+ }
}
@OnTimer("bundle_flusher")
public void flushBundle(OnTimerContext context) throws IOException {
- publishResults(BeamContextWrapper.of(context));
+ publishResults(BeamContextWrapper.of(context), null);
Review Comment:
What would happen if the same results get published via this timer and the
finishBundle above ? If this is safe, probably add a comment to clarify.
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/write/UnboundedStreamingSolaceWriter.java:
##########
@@ -133,6 +148,10 @@ public void processElement(
@FinishBundle
public void finishBundle(FinishBundleContext context) {
- publishResults(BeamContextWrapper.of(context));
+ if (getDeliveryMode() == DeliveryMode.PERSISTENT) {
+ waitForAcks(BeamContextWrapper.of(context), messageIdsToAck);
Review Comment:
Ditto.
--
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]
