This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 33ffd48a49 [Fix][Connector-V2] Fix Paimon stream source state restore
(#11132)
33ffd48a49 is described below
commit 33ffd48a49b1f6074ff88818ce692c0c4db9b679
Author: QuakeWang <[email protected]>
AuthorDate: Fri Jun 19 18:28:02 2026 +0800
[Fix][Connector-V2] Fix Paimon stream source state restore (#11132)
Signed-off-by: QuakeWang <[email protected]>
---
.../seatunnel/paimon/source/PaimonSource.java | 1 +
.../seatunnel/paimon/source/PaimonSourceState.java | 43 +++++-
.../source/enumerator/AbstractSplitEnumerator.java | 34 ++++-
.../PaimonBatchSourceSplitEnumerator.java | 2 +-
.../PaimonStreamSourceSplitEnumerator.java | 17 +++
.../PaimonStreamSourceSplitEnumeratorTest.java | 167 +++++++++++++++++++++
6 files changed, 255 insertions(+), 9 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSource.java
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSource.java
index 501a1c46b9..aa16e457e1 100644
---
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSource.java
+++
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSource.java
@@ -176,6 +176,7 @@ public class PaimonSource
enumeratorContext,
checkpointState.getAssignedSplits(),
checkpointState.getCurrentSnapshotId(),
+ checkpointState.getCurrentSnapshotIds(),
readBuilders,
1);
}
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceState.java
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceState.java
index db6392520c..73991cdd9a 100644
---
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceState.java
+++
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/PaimonSourceState.java
@@ -17,17 +17,15 @@
package org.apache.seatunnel.connectors.seatunnel.paimon.source;
-import lombok.AllArgsConstructor;
-import lombok.Getter;
-
import javax.annotation.Nullable;
import java.io.Serializable;
+import java.util.Collections;
import java.util.Deque;
+import java.util.HashMap;
+import java.util.Map;
/** Paimon connector source state, saves the splits has assigned to readers. */
-@Getter
-@AllArgsConstructor
public class PaimonSourceState implements Serializable {
private static final long serialVersionUID = 1L;
@@ -35,4 +33,39 @@ public class PaimonSourceState implements Serializable {
private final Deque<PaimonSourceSplit> assignedSplits;
private final @Nullable Long currentSnapshotId;
+
+ private final Map<String, Long> currentSnapshotIds;
+
+ public PaimonSourceState(
+ Deque<PaimonSourceSplit> assignedSplits, @Nullable Long
currentSnapshotId) {
+ this.assignedSplits = assignedSplits;
+ this.currentSnapshotId = currentSnapshotId;
+ this.currentSnapshotIds = Collections.emptyMap();
+ }
+
+ public PaimonSourceState(
+ Deque<PaimonSourceSplit> assignedSplits, Map<String, Long>
currentSnapshotIds) {
+ this.assignedSplits = assignedSplits;
+ this.currentSnapshotIds = new HashMap<>(currentSnapshotIds);
+ this.currentSnapshotId = getSingleSnapshotId(this.currentSnapshotIds);
+ }
+
+ public Deque<PaimonSourceSplit> getAssignedSplits() {
+ return assignedSplits;
+ }
+
+ public @Nullable Long getCurrentSnapshotId() {
+ return currentSnapshotId;
+ }
+
+ public Map<String, Long> getCurrentSnapshotIds() {
+ return currentSnapshotIds == null ? Collections.emptyMap() :
currentSnapshotIds;
+ }
+
+ private static @Nullable Long getSingleSnapshotId(Map<String, Long>
currentSnapshotIds) {
+ if (currentSnapshotIds.size() != 1) {
+ return null;
+ }
+ return currentSnapshotIds.values().iterator().next();
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/AbstractSplitEnumerator.java
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/AbstractSplitEnumerator.java
index e18bda0e22..ed1356656a 100644
---
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/AbstractSplitEnumerator.java
+++
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/AbstractSplitEnumerator.java
@@ -42,6 +42,7 @@ import java.util.Collection;
import java.util.Deque;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
@@ -73,6 +74,8 @@ public abstract class AbstractSplitEnumerator
@Nullable protected Long nextSnapshotId;
+ protected final Map<String, Long> nextSnapshotIds;
+
private ExecutorService executorService;
public AbstractSplitEnumerator(
@@ -82,9 +85,28 @@ public abstract class AbstractSplitEnumerator
Map<String, ReadBuilder> readBuilders,
int splitMaxPerTask,
JobMode jobMode) {
+ this(
+ context,
+ pendingSplits,
+ nextSnapshotId,
+ new HashMap<>(),
+ readBuilders,
+ splitMaxPerTask,
+ jobMode);
+ }
+
+ public AbstractSplitEnumerator(
+ Context<PaimonSourceSplit> context,
+ Deque<PaimonSourceSplit> pendingSplits,
+ @Nullable Long nextSnapshotId,
+ Map<String, Long> nextSnapshotIds,
+ Map<String, ReadBuilder> readBuilders,
+ int splitMaxPerTask,
+ JobMode jobMode) {
this.context = context;
this.pendingSplits = new LinkedList<>(pendingSplits);
this.nextSnapshotId = nextSnapshotId;
+ this.nextSnapshotIds = new LinkedHashMap<>(nextSnapshotIds);
this.readersAwaitingSplit = new LinkedHashSet<>();
this.splitGenerator = new PaimonSourceSplitGenerator();
this.splitMaxNum = context.currentParallelism() * splitMaxPerTask;
@@ -101,8 +123,12 @@ public abstract class AbstractSplitEnumerator
? readBuilder.newScan()
: readBuilder.newStreamScan();
tableScans.put(tableId, scan);
- if (scan instanceof StreamTableScan && nextSnapshotId !=
null) {
- ((StreamTableScan) scan).restore(nextSnapshotId);
+ Long tableSnapshotId =
+ this.nextSnapshotIds.isEmpty()
+ ? nextSnapshotId
+ : this.nextSnapshotIds.get(tableId);
+ if (scan instanceof StreamTableScan && tableSnapshotId !=
null) {
+ ((StreamTableScan) scan).restore(tableSnapshotId);
}
});
}
@@ -146,7 +172,7 @@ public abstract class AbstractSplitEnumerator
@Override
public PaimonSourceState snapshotState(long checkpointId) throws Exception
{
synchronized (stateLock) {
- return new PaimonSourceState(pendingSplits, nextSnapshotId);
+ return new PaimonSourceState(pendingSplits, nextSnapshotIds);
}
}
@@ -245,6 +271,8 @@ public abstract class AbstractSplitEnumerator
for (PlanWithNextSnapshotId planWithNextSnapshotId :
planWithNextSnapshotIds) {
nextSnapshotId = planWithNextSnapshotId.nextSnapshotId;
+ nextSnapshotIds.put(
+ planWithNextSnapshotId.tableId,
planWithNextSnapshotId.nextSnapshotId);
TableScan.Plan plan = planWithNextSnapshotId.plan;
if (plan.splits().isEmpty()) {
continue;
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/PaimonBatchSourceSplitEnumerator.java
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/PaimonBatchSourceSplitEnumerator.java
index 43cf5c4e3e..6b777e7704 100644
---
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/PaimonBatchSourceSplitEnumerator.java
+++
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/PaimonBatchSourceSplitEnumerator.java
@@ -58,7 +58,7 @@ public class PaimonBatchSourceSplitEnumerator extends
AbstractSplitEnumerator {
@Override
public PaimonSourceState snapshotState(long checkpointId) throws Exception
{
synchronized (stateLock) {
- return new PaimonSourceState(pendingSplits, null);
+ return new PaimonSourceState(pendingSplits, (Long) null);
}
}
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/PaimonStreamSourceSplitEnumerator.java
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/PaimonStreamSourceSplitEnumerator.java
index 6852472f59..67706a8e51 100644
---
a/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/PaimonStreamSourceSplitEnumerator.java
+++
b/seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/PaimonStreamSourceSplitEnumerator.java
@@ -39,10 +39,27 @@ public class PaimonStreamSourceSplitEnumerator extends
AbstractSplitEnumerator {
@Nullable Long nextSnapshotId,
Map<String, ReadBuilder> readBuilders,
int splitMaxPerTask) {
+ this(
+ context,
+ pendingSplits,
+ nextSnapshotId,
+ java.util.Collections.emptyMap(),
+ readBuilders,
+ splitMaxPerTask);
+ }
+
+ public PaimonStreamSourceSplitEnumerator(
+ Context<PaimonSourceSplit> context,
+ Deque<PaimonSourceSplit> pendingSplits,
+ @Nullable Long nextSnapshotId,
+ Map<String, Long> nextSnapshotIds,
+ Map<String, ReadBuilder> readBuilders,
+ int splitMaxPerTask) {
super(
context,
pendingSplits,
nextSnapshotId,
+ nextSnapshotIds,
readBuilders,
splitMaxPerTask,
JobMode.STREAMING);
diff --git
a/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/PaimonStreamSourceSplitEnumeratorTest.java
b/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/PaimonStreamSourceSplitEnumeratorTest.java
new file mode 100644
index 0000000000..dce0b36919
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/enumerator/PaimonStreamSourceSplitEnumeratorTest.java
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.paimon.source.enumerator;
+
+import org.apache.seatunnel.api.source.SourceSplitEnumerator;
+import
org.apache.seatunnel.connectors.seatunnel.paimon.source.PaimonSourceSplit;
+import
org.apache.seatunnel.connectors.seatunnel.paimon.source.PaimonSourceState;
+
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.StreamTableScan;
+import org.apache.paimon.table.source.TableScan;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class PaimonStreamSourceSplitEnumeratorTest {
+
+ @Test
+ void shouldKeepIndependentSnapshotIdForEachTableWhenRestore() throws
Exception {
+ StreamTableScan firstScan = newStreamScan(10L);
+ StreamTableScan secondScan = newStreamScan(20L);
+
+ PaimonStreamSourceSplitEnumerator enumerator =
+ new PaimonStreamSourceSplitEnumerator(
+ context(),
+ new LinkedList<>(),
+ null,
+ readBuilders(firstScan, secondScan),
+ 1);
+
+ enumerator.processDiscoveredSplits(enumerator.scanNextSnapshot(),
null);
+ PaimonSourceState state = enumerator.snapshotState(1L);
+ StreamTableScan restoredFirstScan = newStreamScan(11L);
+ StreamTableScan restoredSecondScan = newStreamScan(21L);
+
+ PaimonStreamSourceSplitEnumerator restored =
+ new PaimonStreamSourceSplitEnumerator(
+ context(),
+ state.getAssignedSplits(),
+ state.getCurrentSnapshotId(),
+ state.getCurrentSnapshotIds(),
+ readBuilders(restoredFirstScan, restoredSecondScan),
+ 1);
+ restored.close();
+ enumerator.close();
+
+ verify(firstScan).checkpoint();
+ verify(secondScan).checkpoint();
+ assertEquals(10L, state.getCurrentSnapshotIds().get("db.table_a"));
+ assertEquals(20L, state.getCurrentSnapshotIds().get("db.table_b"));
+ verify(restoredFirstScan).restore(10L);
+ verify(restoredSecondScan).restore(20L);
+ }
+
+ @Test
+ void shouldNotFallbackToAnotherTableSnapshotWhenTableSnapshotIsNull()
throws Exception {
+ StreamTableScan firstScan = newStreamScan(10L);
+ StreamTableScan secondScan = newStreamScan(null);
+
+ PaimonStreamSourceSplitEnumerator enumerator =
+ new PaimonStreamSourceSplitEnumerator(
+ context(),
+ new LinkedList<>(),
+ null,
+ readBuilders(firstScan, secondScan),
+ 1);
+
+ enumerator.processDiscoveredSplits(enumerator.scanNextSnapshot(),
null);
+ PaimonSourceState state = enumerator.snapshotState(1L);
+ StreamTableScan restoredFirstScan = newStreamScan(11L);
+ StreamTableScan restoredSecondScan = newStreamScan(21L);
+
+ PaimonStreamSourceSplitEnumerator restored =
+ new PaimonStreamSourceSplitEnumerator(
+ context(),
+ state.getAssignedSplits(),
+ state.getCurrentSnapshotId(),
+ state.getCurrentSnapshotIds(),
+ readBuilders(restoredFirstScan, restoredSecondScan),
+ 1);
+ restored.close();
+ enumerator.close();
+
+ assertEquals(10L, state.getCurrentSnapshotIds().get("db.table_a"));
+ assertTrue(state.getCurrentSnapshotIds().containsKey("db.table_b"));
+ assertNull(state.getCurrentSnapshotIds().get("db.table_b"));
+ verify(restoredFirstScan).restore(10L);
+ verify(restoredSecondScan, never()).restore(10L);
+ }
+
+ @Test
+ void shouldRestoreLegacySingleSnapshotId() throws Exception {
+ StreamTableScan restoredScan = newStreamScan(11L);
+
+ PaimonStreamSourceSplitEnumerator restored =
+ new PaimonStreamSourceSplitEnumerator(
+ context(), new LinkedList<>(), 10L,
readBuilders(restoredScan), 1);
+ restored.close();
+
+ verify(restoredScan).restore(10L);
+ }
+
+ private static Map<String, ReadBuilder> readBuilders(
+ StreamTableScan firstScan, StreamTableScan secondScan) {
+ Map<String, ReadBuilder> readBuilders = new LinkedHashMap<>();
+ readBuilders.put("db.table_a", readBuilder(firstScan));
+ readBuilders.put("db.table_b", readBuilder(secondScan));
+ return readBuilders;
+ }
+
+ private static Map<String, ReadBuilder> readBuilders(StreamTableScan scan)
{
+ Map<String, ReadBuilder> readBuilders = new LinkedHashMap<>();
+ readBuilders.put("db.table_a", readBuilder(scan));
+ return readBuilders;
+ }
+
+ private static ReadBuilder readBuilder(StreamTableScan scan) {
+ ReadBuilder readBuilder = mock(ReadBuilder.class);
+ when(readBuilder.newStreamScan()).thenReturn(scan);
+ return readBuilder;
+ }
+
+ private static StreamTableScan newStreamScan(Long nextSnapshotId) {
+ StreamTableScan scan = mock(StreamTableScan.class);
+ TableScan.Plan plan = mock(TableScan.Plan.class);
+ when(plan.splits()).thenReturn(Collections.emptyList());
+ when(scan.plan()).thenReturn(plan);
+ when(scan.checkpoint()).thenReturn(nextSnapshotId);
+ return scan;
+ }
+
+ private static SourceSplitEnumerator.Context<PaimonSourceSplit> context() {
+ SourceSplitEnumerator.Context<PaimonSourceSplit> context =
+ mock(SourceSplitEnumerator.Context.class);
+ when(context.currentParallelism()).thenReturn(1);
+ when(context.registeredReaders()).thenReturn(new HashSet<>());
+ return context;
+ }
+}
