jbonofre commented on code in PR #40605: URL: https://github.com/apache/beam/pull/40605#discussion_r4411525134
########## sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java: ########## @@ -44,6 +46,20 @@ class JmsCheckpointMark implements UnboundedSource.CheckpointMark, Serializable private static final Logger LOG = LoggerFactory.getLogger(JmsCheckpointMark.class); + /** + * Live checkpoints of this JVM that still hold unacknowledged messages, by id. The messages, + * consumer and session of a checkpoint are transient, so a runner that serializes and restores + * the mark (or recreates the reader) would otherwise lose them and never acknowledge. A restored + * mark uses this registry to find the live checkpoint with the same id and finalize that one. + */ + private static final ConcurrentHashMap<String, JmsCheckpointMark> LIVE_CHECKPOINTS = Review Comment: Good point. I replaced the `ConcurrentHashMap` and the `removeIf` scan with a Guava `Cache` using `expireAfterWrite(1, HOURS)`. Registering a checkpoint is now a single `put`, and finalize uses an atomic `asMap().remove(id)`. -- 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]
