From 94cf005d80100a03b96a2d8b5a23aa2100755859 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jean-Baptiste=20Onofr=C3=A9?= Date: Thu, 15 Oct 2026 09:22:09 +0200 Subject: [PATCH 1/5] [JmsIO] Let restored checkpoint marks acknowledge live messages JmsCheckpointMark keeps its messages, consumer and session in transient fields, so a mark that is serialized and restored (or finalized after the reader was recreated) acknowledges nothing. Give each mark an id and keep live marks in a per-JVM registry so a restored copy can finalize the live one. Entries not finalized within an hour are dropped. --- .../beam/sdk/io/jms/JmsCheckpointMark.java | 41 +++++++++++++++++++ .../apache/beam/sdk/io/jms/JmsLocalTest.java | 31 ++++++++++++++ 2 files changed, 72 insertions(+) diff --git a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java index fdfd2c0c7bce..fe69338779ff 100644 --- a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java +++ b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java @@ -26,6 +26,8 @@ import java.util.ArrayList; import java.util.List; import java.util.Objects; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.ReentrantReadWriteLock; import org.apache.beam.sdk.io.UnboundedSource; @@ -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 LIVE_CHECKPOINTS = + new ConcurrentHashMap<>(); + + /** Live checkpoints not finalized after this long are dropped; the broker redelivers them. */ + private static final long LIVE_CHECKPOINT_MAX_AGE_MILLIS = 60 * 60 * 1000L; + + private final String id = UUID.randomUUID().toString(); + private final long creationTimeMillis = System.currentTimeMillis(); private Instant oldestMessageTimestamp; private transient @Nullable List messages; private transient @Nullable MessageConsumer consumer; @@ -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 { if (messages != null) { for (Message message : messages) { @@ -106,6 +131,19 @@ public void finalizeCheckpoint() { } } + private void registerLive() { + long now = System.currentTimeMillis(); + LIVE_CHECKPOINTS + .values() + .removeIf(m -> now - m.creationTimeMillis > LIVE_CHECKPOINT_MAX_AGE_MILLIS); + LIVE_CHECKPOINTS.put(id, this); + } + + @VisibleForTesting + static int liveCheckpointCount() { + return LIVE_CHECKPOINTS.size(); + } + @VisibleForTesting @Nullable List getMessages() { return messages; @@ -251,6 +289,9 @@ JmsCheckpointMark newCheckpoint( activeCheckpoints); messages.clear(); oldestMessageTimestamp = Instant.now(); + if (messagesCopy != null) { + checkpointMark.registerLive(); + } if (activeCheckpoints != null) { activeCheckpoints.incrementAndGet(); } diff --git a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java index 277c84151703..84386c8cb1f2 100644 --- a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java +++ b/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 + java.io.ByteArrayOutputStream bytes = new java.io.ByteArrayOutputStream(); + try (java.io.ObjectOutputStream out = new java.io.ObjectOutputStream(bytes)) { + out.writeObject(mark); + } + JmsCheckpointMark restored; + try (java.io.ObjectInputStream in = + new java.io.ObjectInputStream(new java.io.ByteArrayInputStream(bytes.toByteArray()))) { + restored = (JmsCheckpointMark) in.readObject(); + } + assertNull(restored.getMessages()); + + restored.finalizeCheckpoint(); + + Mockito.verify(msg, Mockito.times(1)).acknowledge(); + assertEquals(0, activeCheckpoints.get()); + assertEquals(live - 1, JmsCheckpointMark.liveCheckpointCount()); + } + /** Test the checkpoint mark default coder, which is actually AvroCoder. */ @Test public void testCheckpointMarkDefaultCoder() throws Exception { From 165d4f6ac1c702609309020c6209458740e078ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jean-Baptiste=20Onofr=C3=A9?= Date: Fri, 16 Oct 2026 17:36:43 +0200 Subject: [PATCH 2/5] [JmsIO] Test finalizing a restored checkpoint mark after reader recreation --- .../org/apache/beam/sdk/io/jms/JmsIOTest.java | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java index bdcf27bbfa49..70a2b26cee7d 100644 --- a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java +++ b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java @@ -598,6 +598,39 @@ public void testCheckpointMarkAndFinalizeSeparately() throws Exception { assertEquals(7, count(QUEUE)); } + @Test + public void testRestoredCheckpointMarkFinalizedAfterReaderRecreation() throws Exception { + UnboundedJmsReader reader = setupReaderForTest(); + assertTrue(startWithRetry(reader)); + assertTrue(advanceWithRetry(reader)); + assertTrue(advanceWithRetry(reader)); + + JmsCheckpointMark mark = (JmsCheckpointMark) reader.getCheckpointMark(); + assertEquals(10, count(QUEUE)); + + // the runner serializes the mark and recreates the reader from it + java.io.ByteArrayOutputStream bytes = new java.io.ByteArrayOutputStream(); + try (java.io.ObjectOutputStream out = new java.io.ObjectOutputStream(bytes)) { + out.writeObject(mark); + } + JmsCheckpointMark restored; + try (java.io.ObjectInputStream in = + new java.io.ObjectInputStream(new java.io.ByteArrayInputStream(bytes.toByteArray()))) { + restored = (JmsCheckpointMark) in.readObject(); + } + reader.close(); + UnboundedJmsReader recreated = + (UnboundedJmsReader) + ((JmsIO.UnboundedJmsSource) reader.getCurrentSource()) + .createReader(PipelineOptionsFactory.create(), restored); + assertTrue(startWithRetry(recreated)); + + // finalizing the restored copy acks the messages read by the original reader + restored.finalizeCheckpoint(); + + assertEquals(7, count(QUEUE)); + } + @Test public void testCheckpointMarkAndFinalizeSeparatelyIndividualAcknowledge() throws Exception { UnboundedJmsReader reader = setupReaderForTest(JmsIO.AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE); From 86d28a3a01ac8930d014600ffee52cd914eb40e6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?JB=20Onofr=C3=A9?= Date: Thu, 29 Oct 2026 11:59:55 +0100 Subject: [PATCH 3/5] [JmsIO] Address review comments on checkpoint registry Use a Guava Cache for live checkpoints, guard against a null id when restoring older checkpoints, make finalizeCheckpoint idempotent, include id in equals/hashCode, and use the checkpoint coder in the restore test. --- .../beam/sdk/io/jms/JmsCheckpointMark.java | 47 ++++++++++--------- .../apache/beam/sdk/io/jms/JmsLocalTest.java | 17 +++---- 2 files changed, 31 insertions(+), 33 deletions(-) diff --git a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java index fe69338779ff..920f7e2670e1 100644 --- a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java +++ b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java @@ -27,12 +27,14 @@ import java.util.List; import java.util.Objects; import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.ReentrantReadWriteLock; import org.apache.beam.sdk.io.UnboundedSource; import org.apache.beam.sdk.io.jms.JmsIO.AcknowledgeMode; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.Cache; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.CacheBuilder; import org.checkerframework.checker.nullness.qual.Nullable; import org.joda.time.Instant; import org.slf4j.Logger; @@ -51,15 +53,13 @@ class JmsCheckpointMark implements UnboundedSource.CheckpointMark, Serializable * 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 ConcurrentHashMap LIVE_CHECKPOINTS = - new ConcurrentHashMap<>(); + private static final Cache LIVE_CHECKPOINTS = + CacheBuilder.newBuilder().expireAfterWrite(1, TimeUnit.HOURS).build(); - /** Live checkpoints not finalized after this long are dropped; the broker redelivers them. */ - private static final long LIVE_CHECKPOINT_MAX_AGE_MILLIS = 60 * 60 * 1000L; - - private final String id = UUID.randomUUID().toString(); - private final long creationTimeMillis = System.currentTimeMillis(); + // May be null when restored from a checkpoint written by a version without this field. + private final @Nullable String id = UUID.randomUUID().toString(); private Instant oldestMessageTimestamp; private transient @Nullable List messages; private transient @Nullable MessageConsumer consumer; @@ -81,21 +81,26 @@ 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) { + public synchronized void finalizeCheckpoint() { + if (id != null) { + // Atomic, so only one copy of this checkpoint ever finalizes the live one. + JmsCheckpointMark live = LIVE_CHECKPOINTS.asMap().remove(id); + if (live != null && live != this) { + // This is a restored copy whose state lives in another instance. live.finalizeCheckpoint(); return; } } - LIVE_CHECKPOINTS.remove(id); + if (messages == null && consumer == null && session == null && activeCheckpoints == null) { + // Nothing to acknowledge or release, e.g. a restored copy that was already finalized. + return; + } try { if (messages != null) { for (Message message : messages) { message.acknowledge(); } + messages = null; } } catch (JMSException e) { // The effect of this is message not get acknowledged and thus will be redelivered. It is @@ -128,19 +133,16 @@ public void finalizeCheckpoint() { if (activeCheckpoints != null) { activeCheckpoints.decrementAndGet(); + activeCheckpoints = null; } } private void registerLive() { - long now = System.currentTimeMillis(); - LIVE_CHECKPOINTS - .values() - .removeIf(m -> now - m.creationTimeMillis > LIVE_CHECKPOINT_MAX_AGE_MILLIS); - LIVE_CHECKPOINTS.put(id, this); + LIVE_CHECKPOINTS.put(Objects.requireNonNull(id), this); } @VisibleForTesting - static int liveCheckpointCount() { + static long liveCheckpointCount() { return LIVE_CHECKPOINTS.size(); } @@ -177,12 +179,13 @@ public boolean equals(@Nullable Object o) { return false; } JmsCheckpointMark that = (JmsCheckpointMark) o; - return oldestMessageTimestamp.equals(that.oldestMessageTimestamp); + return Objects.equals(id, that.id) + && oldestMessageTimestamp.equals(that.oldestMessageTimestamp); } @Override public int hashCode() { - return Objects.hash(oldestMessageTimestamp); + return Objects.hash(id, oldestMessageTimestamp); } static Preparer newPreparer(AcknowledgeMode acknowledgeMode) { diff --git a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java index 84386c8cb1f2..24c46f94d1da 100644 --- a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java +++ b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java @@ -45,6 +45,7 @@ import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.testing.CoderProperties; +import org.apache.beam.sdk.util.CoderUtils; import org.apache.beam.sdk.util.SerializableUtils; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Throwables; import org.joda.time.Duration; @@ -180,18 +181,12 @@ public void testRestoredCheckpointMarkAcknowledgesLiveMessages() throws Exceptio JmsCheckpointMark mark = preparer.newCheckpoint( null, null, JmsIO.AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE, activeCheckpoints); - int live = JmsCheckpointMark.liveCheckpointCount(); + long live = JmsCheckpointMark.liveCheckpointCount(); - // serialize/deserialize drops the transient state, as when a runner restores the mark - java.io.ByteArrayOutputStream bytes = new java.io.ByteArrayOutputStream(); - try (java.io.ObjectOutputStream out = new java.io.ObjectOutputStream(bytes)) { - out.writeObject(mark); - } - JmsCheckpointMark restored; - try (java.io.ObjectInputStream in = - new java.io.ObjectInputStream(new java.io.ByteArrayInputStream(bytes.toByteArray()))) { - restored = (JmsCheckpointMark) in.readObject(); - } + // encode/decode drops the transient state, as when a runner restores the mark + Coder coder = + new JmsIO.UnboundedJmsSource(null).getCheckpointMarkCoder(); + JmsCheckpointMark restored = CoderUtils.clone(coder, mark); assertNull(restored.getMessages()); restored.finalizeCheckpoint(); From 2b0b034b1dd7a11580ad4c617197b11a0c87c484 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?JB=20Onofr=C3=A9?= Date: Thu, 29 Oct 2026 13:19:06 +0100 Subject: [PATCH 4/5] [JmsIO] Suppress ReferenceEquality warning in finalizeCheckpoint --- .../main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java | 1 + 1 file changed, 1 insertion(+) diff --git a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java index 920f7e2670e1..881ea5e64dcb 100644 --- a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java +++ b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java @@ -81,6 +81,7 @@ private JmsCheckpointMark( /** Acknowledge all outstanding message. */ @Override + @SuppressWarnings("ReferenceEquality") // live must be a different instance, not an equal one public synchronized void finalizeCheckpoint() { if (id != null) { // Atomic, so only one copy of this checkpoint ever finalizes the live one. From fd4f216b54405800d79fcb7c829dc003c72eebc7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?JB=20Onofr=C3=A9?= Date: Tue, 3 Nov 2026 10:20:54 +0100 Subject: [PATCH 5/5] [JmsIO] Bound checkpoint registry and keep older checkpoint marks restorable - Cap the live checkpoint registry at 10,000 entries. - Pin serialVersionUID to the value computed before the id field was added, so checkpoint marks encoded by previous versions still decode (with a null id) instead of failing with InvalidClassException. - Leave the random id out of equals/hashCode. - Round-trip checkpoint marks through the checkpoint mark coder in tests and add a test decoding a mark encoded before the id field existed. --- .../beam/sdk/io/jms/JmsCheckpointMark.java | 17 +++++++++----- .../org/apache/beam/sdk/io/jms/JmsIOTest.java | 19 +++++----------- .../apache/beam/sdk/io/jms/JmsLocalTest.java | 22 ++++++++++++++++++- 3 files changed, 37 insertions(+), 21 deletions(-) diff --git a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java index 881ea5e64dcb..6c19a554db8e 100644 --- a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java +++ b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java @@ -46,17 +46,22 @@ */ class JmsCheckpointMark implements UnboundedSource.CheckpointMark, Serializable { + // Pinned to the value computed for this class before the id field was added, so checkpoints + // written by those versions can still be restored. + private static final long serialVersionUID = -275230511397134772L; + 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. + * mark uses this registry to find the live checkpoint with the same id and finalize that one. To + * bound memory, at most 10,000 checkpoints are kept and those not finalized within an hour are + * dropped; the broker redelivers their messages. */ private static final Cache LIVE_CHECKPOINTS = - CacheBuilder.newBuilder().expireAfterWrite(1, TimeUnit.HOURS).build(); + CacheBuilder.newBuilder().maximumSize(10_000).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(); @@ -180,13 +185,13 @@ public boolean equals(@Nullable Object o) { return false; } JmsCheckpointMark that = (JmsCheckpointMark) o; - return Objects.equals(id, that.id) - && oldestMessageTimestamp.equals(that.oldestMessageTimestamp); + // The id is random and only used to find the live checkpoint, so it is not part of equality. + return oldestMessageTimestamp.equals(that.oldestMessageTimestamp); } @Override public int hashCode() { - return Objects.hash(id, oldestMessageTimestamp); + return Objects.hash(oldestMessageTimestamp); } static Preparer newPreparer(AcknowledgeMode acknowledgeMode) { diff --git a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java index 70a2b26cee7d..f01000db60ee 100644 --- a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java +++ b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java @@ -90,6 +90,7 @@ import org.apache.beam.sdk.transforms.Count; import org.apache.beam.sdk.transforms.Create; import org.apache.beam.sdk.transforms.SerializableBiFunction; +import org.apache.beam.sdk.util.CoderUtils; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Throwables; import org.apache.qpid.jms.JmsAcknowledgeCallback; @@ -608,21 +609,11 @@ public void testRestoredCheckpointMarkFinalizedAfterReaderRecreation() throws Ex JmsCheckpointMark mark = (JmsCheckpointMark) reader.getCheckpointMark(); assertEquals(10, count(QUEUE)); - // the runner serializes the mark and recreates the reader from it - java.io.ByteArrayOutputStream bytes = new java.io.ByteArrayOutputStream(); - try (java.io.ObjectOutputStream out = new java.io.ObjectOutputStream(bytes)) { - out.writeObject(mark); - } - JmsCheckpointMark restored; - try (java.io.ObjectInputStream in = - new java.io.ObjectInputStream(new java.io.ByteArrayInputStream(bytes.toByteArray()))) { - restored = (JmsCheckpointMark) in.readObject(); - } + // the runner encodes the mark and recreates the reader from it + JmsIO.UnboundedJmsSource source = (JmsIO.UnboundedJmsSource) reader.getCurrentSource(); + JmsCheckpointMark restored = CoderUtils.clone(source.getCheckpointMarkCoder(), mark); reader.close(); - UnboundedJmsReader recreated = - (UnboundedJmsReader) - ((JmsIO.UnboundedJmsSource) reader.getCurrentSource()) - .createReader(PipelineOptionsFactory.create(), restored); + UnboundedJmsReader recreated = source.createReader(PipelineOptionsFactory.create(), restored); assertTrue(startWithRetry(recreated)); // finalizing the restored copy acks the messages read by the original reader diff --git a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java index 24c46f94d1da..39c0b0ae90ff 100644 --- a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java +++ b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java @@ -35,6 +35,7 @@ import jakarta.jms.Session; import java.io.IOException; import java.io.NotSerializableException; +import java.util.Base64; import java.util.List; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -196,7 +197,26 @@ public void testRestoredCheckpointMarkAcknowledgesLiveMessages() throws Exceptio assertEquals(live - 1, JmsCheckpointMark.liveCheckpointCount()); } - /** Test the checkpoint mark default coder, which is actually AvroCoder. */ + /** A checkpoint mark encoded by the checkpoint mark coder before marks had an id. */ + private static final String CHECKPOINT_MARK_WITHOUT_ID = + "rO0ABXNyACxvcmcuYXBhY2hlLmJlYW0uc2RrLmlvLmptcy5KbXNDaGVja3BvaW50TWFya/wuL07g0x5MAgAB" + + "TAAWb2xkZXN0TWVzc2FnZVRpbWVzdGFtcHQAF0xvcmcvam9kYS90aW1lL0luc3RhbnQ7eHBzcgAVb3JnLmpv" + + "ZGEudGltZS5JbnN0YW50Lci+0MYOnM0CAAFKAAdpTWlsbGlzeHAAAAGLz+VoAA=="; + + @Test + public void testDecodeCheckpointMarkWithoutId() throws Exception { + Coder coder = + new JmsIO.UnboundedJmsSource(null).getCheckpointMarkCoder(); + JmsCheckpointMark restored = + CoderUtils.decodeFromByteArray( + coder, Base64.getDecoder().decode(CHECKPOINT_MARK_WITHOUT_ID)); + assertNull(restored.getMessages()); + + // nothing to acknowledge, and a missing id must not break finalization + restored.finalizeCheckpoint(); + } + + /** Test the checkpoint mark default coder, which is SerializableCoder. */ @Test public void testCheckpointMarkDefaultCoder() throws Exception { JmsCheckpointMark jmsCheckpointMark =