chamikaramj commented on code in PR #40605:
URL: https://github.com/apache/beam/pull/40605#discussion_r4416150409
##########
sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java:
##########
@@ -44,6 +48,18 @@ 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.
+ * Checkpoints not finalized within an hour are dropped; the broker
redelivers their messages.
+ */
+ private static final Cache<String, JmsCheckpointMark> LIVE_CHECKPOINTS =
+ CacheBuilder.newBuilder().expireAfterWrite(1, TimeUnit.HOURS).build();
Review Comment:
Probably also good to add a `maximumSize` if this could potentially lead to
OOMs by holding a large number of objects up to 1 hour.
##########
sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java:
##########
@@ -44,6 +48,18 @@ 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.
+ * Checkpoints not finalized within an hour are dropped; the broker
redelivers their messages.
+ */
+ private static final Cache<String, JmsCheckpointMark> LIVE_CHECKPOINTS =
+ CacheBuilder.newBuilder().expireAfterWrite(1, TimeUnit.HOURS).build();
+
+ // May be null when restored from a checkpoint written by a version without
this field.
+ private final @Nullable String id = UUID.randomUUID().toString();
Review Comment:
We should override `equals` and `hashCode` since we randomly generate the ID
here and two `CheckPointMark`s that are otherwise equal will end up being not
equal (if a runner ever checks that).
##########
sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java:
##########
@@ -44,6 +48,18 @@ 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.
+ * Checkpoints not finalized within an hour are dropped; the broker
redelivers their messages.
+ */
+ private static final Cache<String, JmsCheckpointMark> LIVE_CHECKPOINTS =
+ CacheBuilder.newBuilder().expireAfterWrite(1, TimeUnit.HOURS).build();
+
+ // May be null when restored from a checkpoint written by a version without
this field.
Review Comment:
Probably should add `@AvroDefault("null")` for backwards compatiblity.
--
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]
