sjvanrossum commented on code in PR #37164:
URL: https://github.com/apache/beam/pull/37164#discussion_r2640527608
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java:
##########
@@ -164,10 +194,13 @@ public Instant getWatermark() {
@Override
public UnboundedSource.CheckpointMark getCheckpointMark() {
-
- ImmutableList<BytesXMLMessage> bytesXMLMessages =
ImmutableList.copyOf(receivedMessages);
+ Queue<BytesXMLMessage> safeToAckMessages = new ConcurrentLinkedQueue<>();
+ safeToAckMessages.addAll(receivedMessages);
Review Comment:
I see, nvm. It seems like we're doing the same work that's expected of the
bundle finalizer though.
I'm trying to think if we could capture the outstanding checkpoint
finalization calls instead with an `ExecutorService` for example. When the
reader is closed you'd then call `ExecutorService#shutdownNow()` which returns
a collection of all `Runnable` items that will not be executed and those items
could be executed then instead.
--
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]
