ppawel commented on code in PR #39612:
URL: https://github.com/apache/beam/pull/39612#discussion_r3726730177
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/write/UnboundedSolaceWriter.java:
##########
@@ -218,6 +223,27 @@ public void publishResults(BeamContextWrapper context) {
}
}
+ public void waitForAcks(BeamContextWrapper context, Set<String>
messageIdsToAck) {
+ long timeoutMs = System.currentTimeMillis() + ACKS_FLUSHING_INTERVAL_SECS
* 1000;
+ while (!messageIdsToAck.isEmpty() && System.currentTimeMillis() <
timeoutMs) {
+ publishResults(context, messageIdsToAck);
+ if (!messageIdsToAck.isEmpty()) {
+ try {
+ Thread.sleep(10);
Review Comment:
This will increase my project's test suite run by 10x :) Maybe this could be
done without `sleep`, e.g. the waiting thread is woken up as soon as Solace
callback with the ack arrives, and then checks the ack queue.
--
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]
