jbonofre commented on code in PR #40605:
URL: https://github.com/apache/beam/pull/40605#discussion_r4452812176
##########
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:
Done, the registry now also has `maximumSize(10_000)`. An evicted entry
behaves like an expired one: a restored copy no longer finds the live
checkpoint, and the broker redelivers its messages. The original instance still
finalizes itself, since it holds its own state.
##########
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:
The checkpoint mark coder is `SerializableCoder`
(`UnboundedJmsSource#getCheckpointMarkCoder`), so `@AvroDefault` wouldn't
apply. The test Javadoc saying "AvroCoder" was stale; I fixed it.
Your compatibility concern does apply to Java serialization, though: adding
`id` changed the implicit `serialVersionUID`, so marks encoded before this
change failed with `InvalidClassException`. I pinned `serialVersionUID` to the
previous value (`-275230511397134772L`), so older marks decode with `id ==
null`. I also added `testDecodeCheckpointMarkWithoutId`, which decodes a mark
encoded by the previous class.
##########
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:
Agreed. They compare only `oldestMessageTimestamp` again, as before this PR,
so the random id doesn't affect equality. The registry lookup uses the id
directly, and `finalizeCheckpoint()` compares the live instance by reference.
##########
sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java:
##########
@@ -66,6 +82,15 @@ private JmsCheckpointMark(
/** Acknowledge all outstanding message. */
@Override
public void finalizeCheckpoint() {
+ if (messages == null && consumer == null && session == null) {
+ // Possibly a restored copy of a checkpoint whose state lives in another
instance.
+ JmsCheckpointMark live = LIVE_CHECKPOINTS.get(id);
+ if (live != null) {
+ live.finalizeCheckpoint();
+ return;
+ }
+ }
+ LIVE_CHECKPOINTS.remove(id);
try {
Review Comment:
A checkpoint without messages can still hold a consumer and session to close
(`CLIENT_ACKNOWLEDGE`), and it always increments `activeCheckpoints`, which the
reader waits on before closing its connection. So we can't return on `messages
== null` alone. `finalizeCheckpoint()` returns early when there is nothing to
acknowledge or release (`messages`, `consumer`, `session` and
`activeCheckpoints` all null).
##########
sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java:
##########
@@ -170,6 +170,37 @@ public void
testJmsCheckpointMarkClientAcknowledgeUnsafeNoSessionRecreation() th
assertEquals(0, activeCheckpoints.get());
}
+ @Test
+ public void testRestoredCheckpointMarkAcknowledgesLiveMessages() throws
Exception {
+ Message msg = Mockito.mock(Message.class);
+ JmsCheckpointMark.Preparer preparer =
+
JmsCheckpointMark.newPreparer(JmsIO.AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE);
+ preparer.add(msg);
+ AtomicInteger activeCheckpoints = new AtomicInteger(0);
+ JmsCheckpointMark mark =
+ preparer.newCheckpoint(
+ null, null, JmsIO.AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE,
activeCheckpoints);
+ int live = JmsCheckpointMark.liveCheckpointCount();
+
+ // serialize/deserialize drops the transient state, as when a runner
restores the mark
Review Comment:
Done. Both tests now round-trip through the source's checkpoint mark coder
(`SerializableCoder`) with `CoderUtils.clone`:
`JmsLocalTest#testRestoredCheckpointMarkAcknowledgesLiveMessages` and
`JmsIOTest#testRestoredCheckpointMarkFinalizedAfterReaderRecreation`.
--
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]
