(seatunnel) branch dev updated: [Fix][API] Forward restoreCommit to each table in MultiTableSinkAggregatedCommitter (#12807)

[zhiweiniu]

This is an automated email from the ASF dual-hosted git repository.

nzw921rx 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 ebb2cc3033 [Fix][API] Forward restoreCommit to each table in 
MultiTableSinkAggregatedCommitter (#12807)
ebb2cc3033 is described below

commit ebb2cc3033dd1689c535615242820150174c1c1e
Author: Ricky Makhija <96081477+ricky2...@users.noreply.github.com>
AuthorDate: Wed Oct 28 09:00:49 2026 +0530

    [Fix][API] Forward restoreCommit to each table in 
MultiTableSinkAggregatedCommitter (#12807)
    
    Co-authored-by: zhiweiniu <zniu70...@gmail.com>
---
 .../MultiTableSinkAggregatedCommitter.java         | 19 +++++++++-
 .../MultiTableSinkAggregatedCommitterTest.java     | 44 ++++++++++++++++++++++
 2 files changed, 62 insertions(+), 1 deletion(-)

diff --git 
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkAggregatedCommitter.java
 
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkAggregatedCommitter.java
index 863237f78e..0136dc5ece 100644
--- 
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkAggregatedCommitter.java
+++ 
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkAggregatedCommitter.java
@@ -72,6 +72,20 @@ public class MultiTableSinkAggregatedCommitter
     @Override
     public List<MultiTableAggregatedCommitInfo> commit(
             List<MultiTableAggregatedCommitInfo> aggregatedCommitInfo) throws 
IOException {
+        return commitPerTable(aggregatedCommitInfo, false);
+    }
+
+    // Restore must reach each table's restoreCommit, which may treat 
already-resolved commits
+    // differently from commit; the inherited default would route it to commit 
instead.
+    @Override
+    public List<MultiTableAggregatedCommitInfo> restoreCommit(
+            List<MultiTableAggregatedCommitInfo> aggregatedCommitInfo) throws 
IOException {
+        return commitPerTable(aggregatedCommitInfo, true);
+    }
+
+    private List<MultiTableAggregatedCommitInfo> commitPerTable(
+            List<MultiTableAggregatedCommitInfo> aggregatedCommitInfo, boolean 
restore)
+            throws IOException {
         List<MultiTableAggregatedCommitInfo> errorList = new ArrayList<>();
         for (String sinkIdentifier : aggCommitters.keySet()) {
             SinkAggregatedCommitter<?, ?> sinkCommitter = 
aggCommitters.get(sinkIdentifier);
@@ -85,7 +99,10 @@ public class MultiTableSinkAggregatedCommitter
                                                         .get(sinkIdentifier))
                                 .filter(Objects::nonNull)
                                 .collect(Collectors.toList());
-                List errCommitList = sinkCommitter.commit(commitInfo);
+                List errCommitList =
+                        restore
+                                ? sinkCommitter.restoreCommit(commitInfo)
+                                : sinkCommitter.commit(commitInfo);
                 if (errCommitList.size() == 0) {
                     continue;
                 }
diff --git 
a/seatunnel-api/src/test/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkAggregatedCommitterTest.java
 
b/seatunnel-api/src/test/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkAggregatedCommitterTest.java
index b01c04699f..ecae2d98a9 100644
--- 
a/seatunnel-api/src/test/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkAggregatedCommitterTest.java
+++ 
b/seatunnel-api/src/test/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkAggregatedCommitterTest.java
@@ -70,4 +70,48 @@ public class MultiTableSinkAggregatedCommitterTest {
         committer.close();
         Assertions.assertIterableEquals(Arrays.asList("init", "close"), 
methodInvoked);
     }
+
+    @Test
+    void testRestoreCommitRoutesToTableRestoreCommit() throws IOException {
+        List<String> calls = new ArrayList<>();
+        Map<String, SinkAggregatedCommitter<?, ?>> aggCommitters = new 
HashMap<>();
+        aggCommitters.put(
+                "table1",
+                new SinkAggregatedCommitter<Object, Object>() {
+                    @Override
+                    public List<Object> commit(List<Object> 
aggregatedCommitInfo) {
+                        calls.add("commit" + aggregatedCommitInfo);
+                        return Collections.emptyList();
+                    }
+
+                    @Override
+                    public List<Object> restoreCommit(List<Object> 
aggregatedCommitInfo) {
+                        calls.add("restoreCommit" + aggregatedCommitInfo);
+                        return Collections.emptyList();
+                    }
+
+                    @Override
+                    public Object combine(List<Object> commitInfos) {
+                        return null;
+                    }
+
+                    @Override
+                    public void abort(List<Object> aggregatedCommitInfo) {}
+
+                    @Override
+                    public void close() {}
+                });
+        MultiTableSinkAggregatedCommitter committer =
+                new MultiTableSinkAggregatedCommitter(aggCommitters);
+        Map<String, Object> info = new HashMap<>();
+        info.put("table1", "xid-1");
+        List<MultiTableAggregatedCommitInfo> infos =
+                Collections.singletonList(new 
MultiTableAggregatedCommitInfo(info));
+
+        committer.restoreCommit(infos);
+        committer.commit(infos);
+
+        Assertions.assertIterableEquals(
+                Arrays.asList("restoreCommit[xid-1]", "commit[xid-1]"), calls);
+    }
 }