stankiewicz commented on code in PR #38603:
URL: https://github.com/apache/beam/pull/38603#discussion_r3811051507
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java:
##########
@@ -152,29 +174,93 @@ public boolean advance() {
solaceOriginalRecord = receivedXmlMessage;
solaceMappedRecord =
getCurrentSource().getParseFn().apply(receivedXmlMessage);
receivedMessages.add(receivedXmlMessage);
+ messagesReceived.inc();
return true;
}
@Override
public void close() {
- finalizeReadyMessages();
sessionServiceCache.invalidate(readerUuid);
+ ActiveReadersRegistry.unregister(readerUuid);
+ ackExecutor.shutdown();
+ try {
+ if (!ackExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
+ ackExecutor.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ ackExecutor.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ void finalizeCheckpoint(long checkpointId) {
+ List<BytesXMLMessage> messagesToAck = new ArrayList<>();
+
+ synchronized (lock) {
+ SortedMap<Long, PendingCheckpoint> toAck =
pendingCheckpoints.headMap(checkpointId, true);
+ for (PendingCheckpoint cp : toAck.values()) {
+ messagesToAck.addAll(cp.messages);
+ }
+ toAck.clear();
+ }
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ for (BytesXMLMessage msg : messagesToAck) {
+ futures.add(
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ msg.ackMessage();
+ messagesAcked.inc();
+ } catch (IllegalStateException e) {
+ LOG.warn(
+ "SolaceIO.Read: Failed to acknowledge message with
applicationMessageId={}, ackMessageId={}. Session might be closed.",
+ msg.getApplicationMessageId(),
+ msg.getAckMessageId(),
+ e);
+ }
+ },
+ ackExecutor));
+ }
+
+ try {
+ CompletableFuture.allOf(futures.toArray(new
CompletableFuture<?>[0])).join();
+ } catch (Exception e) {
+ LOG.warn("SolaceIO.Read: Exception waiting for message
acknowledgements", e);
+ }
}
- public void finalizeReadyMessages() {
- BytesXMLMessage msg;
- while ((msg = safeToAckMessages.poll()) != null) {
- try {
- msg.ackMessage();
- } catch (IllegalStateException e) {
- LOG.error(
- "SolaceIO.Read: failed to acknowledge the message with
applicationMessageId={}, ackMessageId={}. Returning the message to queue to
retry.",
- msg.getApplicationMessageId(),
- msg.getAckMessageId(),
- e);
- safeToAckMessages.add(msg); // In case the error was transient, might
succeed later
- break; // Commit is only best effort
+ private void checkTimeouts() {
+ long now = clock.get();
+ List<PendingCheckpoint> expired = new ArrayList<>();
+ synchronized (lock) {
+ while (!pendingCheckpoints.isEmpty()) {
+ long oldestId = pendingCheckpoints.firstKey();
+ PendingCheckpoint oldest = pendingCheckpoints.get(oldestId);
+ if (oldest != null && now - oldest.timestamp > ackDeadline.toMillis())
{
+ pendingCheckpoints.remove(oldestId);
+ expired.add(oldest);
+ } else {
+ break;
+ }
+ }
+ }
+
+ for (PendingCheckpoint cp : expired) {
Review Comment:
i think this step should be optional/configurable as NACKing is not always
supported. By default we should just skip expired, as Solace will redeliver
those.
--
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]
