This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-11872-f7a575432fba5c35d36659c16e1590e83ad1f296 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 305f655b6903246ad8de9175e2b59d9ca2bc4a1b Author: Yu, Wang <[email protected]> AuthorDate: Wed Oct 28 16:39:56 2026 +0000 [Fix][Connector-V2] Restore legacy RabbitMQ checkpoint state (#11872) Co-authored-by: davidzollo <[email protected]> Co-authored-by: David Zollo <[email protected]> --- docs/en/connectors/source/Rabbitmq.md | 11 ++ docs/zh/connectors/source/Rabbitmq.md | 9 ++ .../rabbitmq/source/RabbitmqSourceState.java | 29 ++++- .../rabbitmq/source/RabbitmqSplitEnumerator.java | 3 + .../rabbitmq/RabbitmqSplitEnumeratorTest.java | 41 +++++++ .../e2e/connector/rabbitmq/RabbitmqIT.java | 121 +++++++++++++++++++++ 6 files changed, 211 insertions(+), 3 deletions(-) diff --git a/docs/en/connectors/source/Rabbitmq.md b/docs/en/connectors/source/Rabbitmq.md index f677459313..1484dc7067 100644 --- a/docs/en/connectors/source/Rabbitmq.md +++ b/docs/en/connectors/source/Rabbitmq.md @@ -368,6 +368,17 @@ RabbitMQ source supports JSON by default and Protobuf when `format` is set to `p When a SeaTunnel task fails or crashes, the RabbitMQ connection drops, and RabbitMQ automatically requeues any unacknowledged messages. Upon job restoration from a checkpoint, the reader resumes processing without message loss. +## Restoring legacy checkpoints + +Checkpoints containing the legacy `RabbitmqSourceState` serialized by SeaTunnel 2.3.13 can be read +by the current RabbitMQ source. That legacy state did not contain assigned queue splits, so restore +starts with an empty assignment map and rediscovers the queues in the current configuration. Keep +the queue configuration and broker data available during upgrade; this compatibility layer does not +reconstruct message offsets or acknowledgements that were absent from the old checkpoint. + +Current checkpoints retain assigned splits. Broker messages that were not acknowledged before a +failure can be redelivered; legacy-state compatibility does not add an exactly-once guarantee. + ## Changelog <ChangeLog /> diff --git a/docs/zh/connectors/source/Rabbitmq.md b/docs/zh/connectors/source/Rabbitmq.md index d53e2595ae..b9bbe5b87d 100644 --- a/docs/zh/connectors/source/Rabbitmq.md +++ b/docs/zh/connectors/source/Rabbitmq.md @@ -365,6 +365,15 @@ RabbitMQ Source 默认支持 JSON。当 `format` 设置为 `protobuf` 时支持 当 SeaTunnel 任务失败或异常退出时，与 RabbitMQ 的连接会断开，RabbitMQ 会自动将所有未确认（unacknowledged）的消息重新入队（requeue）。在任务从检查点恢复后，Reader 可以重新处理这些消息，避免数据丢失。 +## 恢复旧版本检查点 + +当前 RabbitMQ source 可以读取 SeaTunnel 2.3.13 序列化的旧版 `RabbitmqSourceState`。 +旧状态没有保存已分配的队列 split，因此恢复时先建立空的分配映射，再重新发现当前配置中的队列。 +升级时应保留队列配置和 broker 数据；此兼容层不会重建旧检查点中原本没有的消息位点或确认信息。 + +当前版本的检查点会保存已分配的 split。故障前尚未确认的 broker 消息可能重新投递， +旧状态兼容本身不提供额外的 Exactly-Once 保证。 + ## 变更日志 <ChangeLog /> diff --git a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceState.java b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceState.java index d9846611cd..8aa08ea120 100644 --- a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceState.java +++ b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceState.java @@ -17,7 +17,30 @@ package org.apache.seatunnel.connectors.seatunnel.rabbitmq.source; -import java.io.Serializable; +import org.apache.seatunnel.connectors.seatunnel.rabbitmq.split.RabbitmqSplitEnumeratorState; -/** State for RabbitMQ source reader. */ -public class RabbitmqSourceState implements Serializable {} +import java.util.Collections; + +/** + * Legacy checkpoint state created by RabbitMQ sources before the split enumerator state fix. + * + * @deprecated only used to restore checkpoints written by older connector versions + */ +@Deprecated +public class RabbitmqSourceState extends RabbitmqSplitEnumeratorState { + private static final long serialVersionUID = -1143819030309308746L; + + public RabbitmqSourceState() { + super(Collections.emptyMap()); + } + + /** + * Replaces a legacy instance whose serialized stream has no current superclass state. + * + * <p>Serializable constructors do not initialize those absent fields. A new replacement + * guarantees an empty assignment map and keeps the legacy marker used for restore logging. + */ + private Object readResolve() { + return new RabbitmqSourceState(); + } +} diff --git a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSplitEnumerator.java b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSplitEnumerator.java index 2523c40577..b06e7968f0 100644 --- a/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSplitEnumerator.java +++ b/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSplitEnumerator.java @@ -84,6 +84,9 @@ public class RabbitmqSplitEnumerator RabbitmqSplitEnumeratorState checkpointState) { this(context, rabbitmqConfig, queues); + if (checkpointState instanceof RabbitmqSourceState) { + log.info("Restoring RabbitMQ source from legacy checkpoint state"); + } if (checkpointState != null && checkpointState.getAssignedSplits() != null) { this.assignedSplits.putAll(checkpointState.getAssignedSplits()); for (String assignedSplitId : checkpointState.getAssignedSplits().keySet()) { diff --git a/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqSplitEnumeratorTest.java b/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqSplitEnumeratorTest.java index 22b3d2f045..f454ae398b 100644 --- a/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqSplitEnumeratorTest.java +++ b/seatunnel-connectors-v2/connector-rabbitmq/src/test/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/RabbitmqSplitEnumeratorTest.java @@ -20,6 +20,7 @@ package org.apache.seatunnel.connectors.seatunnel.rabbitmq; import org.apache.seatunnel.api.source.SourceSplitEnumerator; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqConfig; +import org.apache.seatunnel.connectors.seatunnel.rabbitmq.source.RabbitmqSourceState; import org.apache.seatunnel.connectors.seatunnel.rabbitmq.source.RabbitmqSplitEnumerator; import org.apache.seatunnel.connectors.seatunnel.rabbitmq.split.RabbitmqSplit; import org.apache.seatunnel.connectors.seatunnel.rabbitmq.split.RabbitmqSplitEnumeratorState; @@ -28,7 +29,11 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.mockito.Mockito; +import java.io.ByteArrayInputStream; +import java.io.ObjectInputStream; +import java.io.ObjectStreamClass; import java.util.Arrays; +import java.util.Base64; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -40,6 +45,12 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; public class RabbitmqSplitEnumeratorTest { + private static final long LEGACY_STATE_SERIAL_VERSION_UID = -1143819030309308746L; + + // Serialized by RabbitmqSourceState from SeaTunnel 2.3.13. + private static final String LEGACY_CHECKPOINT_STATE = + "rO0ABXNyAE1vcmcuYXBhY2hlLnNlYXR1bm5lbC5jb25uZWN0b3JzLnNlYXR1bm5lbC5yYWJiaXRtcS5zb3VyY2UuUmFiYml0bXFTb3VyY2VTdGF0ZfAgVqbzFPa2AgAAeHA="; + /** * Tests the split assignment logic of the {@link RabbitmqSplitEnumerator} in a Multi-Table * scenario. @@ -157,4 +168,34 @@ public class RabbitmqSplitEnumeratorTest { Assertions.assertTrue(newState.getAssignedSplits().containsKey("queue_A")); Assertions.assertTrue(newState.getAssignedSplits().containsKey("queue_B")); } + + @Test + public void testRestoreLegacySourceState() throws Exception { + Assertions.assertEquals( + LEGACY_STATE_SERIAL_VERSION_UID, + ObjectStreamClass.lookup(RabbitmqSourceState.class).getSerialVersionUID()); + + RabbitmqSplitEnumeratorState legacyState; + try (ObjectInputStream input = + new ObjectInputStream( + new ByteArrayInputStream( + Base64.getDecoder().decode(LEGACY_CHECKPOINT_STATE)))) { + Object state = input.readObject(); + Assertions.assertInstanceOf(RabbitmqSourceState.class, state); + legacyState = Assertions.assertInstanceOf(RabbitmqSplitEnumeratorState.class, state); + } + Assertions.assertNotNull(legacyState.getAssignedSplits()); + Assertions.assertTrue(legacyState.getAssignedSplits().isEmpty()); + + @SuppressWarnings("unchecked") + SourceSplitEnumerator.Context<RabbitmqSplit> context = + mock(SourceSplitEnumerator.Context.class); + List<String> queues = Arrays.asList("queue_A", "queue_B"); + + RabbitmqSplitEnumerator enumerator = + new RabbitmqSplitEnumerator( + context, mock(RabbitmqConfig.class), queues, legacyState); + + Assertions.assertEquals(queues.size(), enumerator.currentUnassignedSplitSize()); + } } diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rabbitmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rabbitmq/RabbitmqIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rabbitmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rabbitmq/RabbitmqIT.java index 2c3ea073ef..1deb77ab43 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rabbitmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rabbitmq/RabbitmqIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rabbitmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rabbitmq/RabbitmqIT.java @@ -21,6 +21,10 @@ import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory; import org.apache.seatunnel.shade.com.typesafe.config.ConfigRenderOptions; import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.source.Boundedness; +import org.apache.seatunnel.api.source.Collector; +import org.apache.seatunnel.api.source.SourceReader; +import org.apache.seatunnel.api.source.SourceSplitEnumerator; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.PhysicalColumn; import org.apache.seatunnel.api.table.catalog.TableIdentifier; @@ -41,6 +45,9 @@ import org.apache.seatunnel.api.table.type.SeaTunnelRowType; import org.apache.seatunnel.connectors.seatunnel.rabbitmq.client.RabbitmqClient; import org.apache.seatunnel.connectors.seatunnel.rabbitmq.config.RabbitmqConfig; import org.apache.seatunnel.connectors.seatunnel.rabbitmq.source.DeliveryMessage; +import org.apache.seatunnel.connectors.seatunnel.rabbitmq.source.RabbitmqSource; +import org.apache.seatunnel.connectors.seatunnel.rabbitmq.split.RabbitmqSplit; +import org.apache.seatunnel.connectors.seatunnel.rabbitmq.split.RabbitmqSplitEnumeratorState; import org.apache.seatunnel.core.starter.seatunnel.args.ClientCommandArgs; import org.apache.seatunnel.core.starter.seatunnel.command.SeaTunnelConfValidateCommand; import org.apache.seatunnel.core.starter.utils.CommandLineUtils; @@ -58,6 +65,7 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestTemplate; import org.junit.jupiter.api.io.TempDir; +import org.mockito.Mockito; import org.testcontainers.containers.Container; import org.testcontainers.containers.GenericContainer; import org.testcontainers.containers.output.Slf4jLogConsumer; @@ -67,6 +75,7 @@ import org.testcontainers.shaded.org.apache.commons.lang3.tuple.Pair; import org.testcontainers.utility.DockerImageName; import org.testcontainers.utility.DockerLoggerFactory; +import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; @@ -84,11 +93,13 @@ import java.time.LocalDateTime; import java.time.temporal.ChronoUnit; import java.util.ArrayList; import java.util.Arrays; +import java.util.Base64; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.UUID; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; @@ -150,6 +161,116 @@ public class RabbitmqIT extends TestSuiteBase implements TestResource { } } + /** + * Exercises the released checkpoint bytes through the production serializer and restore hook. + * + * <p>The restored queue is assigned once, consumed from the real broker, and checkpoint-acked; + * the resulting current state must not rediscover the already assigned split on restore. + */ + @Test + public void testLegacyCheckpointRestoreConsumesReassignedQueue() throws Exception { + // Captured from RabbitmqSourceState in SeaTunnel 2.3.13, before assigned-split state + // existed. + byte[] legacy = + Base64.getDecoder() + .decode( + "rO0ABXNyAE1vcmcuYXBhY2hlLnNlYXR1bm5lbC5jb25uZWN0b3JzLnNlYXR1bm5lbC5yYWJiaXRtcS5zb3VyY2UuUmFiYml0bXFTb3VyY2VTdGF0ZfAgVqbzFPa2AgAAeHA="); + String queue = "legacy-restore-" + UUID.randomUUID(); + try (Connection connection = dryRunConnection(); + Channel channel = connection.createChannel()) { + channel.queueDeclare(queue, DURABLE, EXCLUSIVE, AUTO_DELETE, null); + try { + channel.confirmSelect(); + channel.basicPublish( + "", + queue, + new AMQP.BasicProperties.Builder().correlationId(queue).build(), + "{\"name\":\"legacy-restored\"}".getBytes(StandardCharsets.UTF_8)); + channel.waitForConfirmsOrDie(5000); + RabbitmqSource source = + new RabbitmqSource(ReadonlyConfig.fromMap(dryRunOptions(queue))); + RabbitmqSplitEnumeratorState oldState = + source.getEnumeratorStateSerializer().deserialize(legacy); + Assertions.assertNotNull(oldState.getAssignedSplits()); + Assertions.assertTrue(oldState.getAssignedSplits().isEmpty()); + @SuppressWarnings("unchecked") + SourceSplitEnumerator.Context<RabbitmqSplit> context = + Mockito.mock(SourceSplitEnumerator.Context.class); + Mockito.when(context.registeredReaders()).thenReturn(Collections.singleton(0)); + List<RabbitmqSplit> assigned = new ArrayList<>(); + Mockito.doAnswer( + invocation -> { + assigned.add(invocation.getArgument(1)); + return null; + }) + .when(context) + .assignSplit(Mockito.eq(0), Mockito.any(RabbitmqSplit.class)); + try (SourceSplitEnumerator<RabbitmqSplit, RabbitmqSplitEnumeratorState> enumerator = + source.restoreEnumerator(context, oldState)) { + enumerator.open(); + enumerator.run(); + enumerator.registerReader(0); + Assertions.assertEquals(1, assigned.size()); + Assertions.assertEquals(queue, assigned.get(0).splitId()); + Assertions.assertEquals(0, enumerator.currentUnassignedSplitSize()); + List<SeaTunnelRow> rows = new ArrayList<>(); + Collector<SeaTunnelRow> collector = + new Collector<SeaTunnelRow>() { + @Override + public void collect(SeaTunnelRow row) { + rows.add(row); + } + + @Override + public Object getCheckpointLock() { + return rows; + } + }; + SourceReader.Context readerContext = Mockito.mock(SourceReader.Context.class); + Mockito.when(readerContext.getBoundedness()).thenReturn(Boundedness.UNBOUNDED); + try (SourceReader<SeaTunnelRow, RabbitmqSplit> reader = + source.createReader(readerContext)) { + reader.open(); + reader.addSplits(assigned); + Awaitility.await() + .atMost(Duration.ofSeconds(30)) + .untilAsserted( + () -> { + reader.pollNext(collector); + Assertions.assertEquals(1, rows.size()); + Assertions.assertEquals( + "legacy-restored", rows.get(0).getField(0)); + }); + reader.snapshotState(1L); + reader.notifyCheckpointComplete(1L); + } + Assertions.assertNull( + channel.basicGet(queue, true), + "checkpoint-acked data must not be redelivered after reader close"); + RabbitmqSplitEnumeratorState current = + source.getEnumeratorStateSerializer() + .deserialize( + source.getEnumeratorStateSerializer() + .serialize(enumerator.snapshotState(1L))); + Assertions.assertEquals( + Collections.singleton(queue), current.getAssignedSplits().keySet()); + assigned.clear(); + try (SourceSplitEnumerator<RabbitmqSplit, RabbitmqSplitEnumeratorState> + restored = source.restoreEnumerator(context, current)) { + restored.open(); + restored.run(); + Assertions.assertEquals(0, restored.currentUnassignedSplitSize()); + Assertions.assertTrue( + assigned.isEmpty(), + "current assigned state must not assign the queue twice"); + } + } + } finally { + channel.queueDelete(queue); + } + } + } + @Test public void testDryRunLeavesMessagesAndConsumersUnchanged() throws Exception { try (Connection connection = dryRunConnection();
