[Fix][API] Fixed not invoke the SinkAggregatedCommitter's init method (#9070)

Author: Hisoka-X

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/SinkAggregatedCommitter.java

@@ -32,7 +32,10 @@
  */
 public interface SinkAggregatedCommitter<CommitInfoT, AggregatedCommitInfoT> extends Serializable {
 
-    /** init sink aggregated committer */
+    /**
+     * init sink aggregated committer, this method will be called not once. Each retry will call
+     * this.
+     */
     default void init() {};
 
     /** Re-commit message to third party data receiver, The method need to achieve idempotency. */

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkAggregatedCommitter.java

@@ -53,17 +53,16 @@ private void initResourceManager() {
         for (String tableIdentifier : aggCommitters.keySet()) {
             SinkAggregatedCommitter<?, ?> aggCommitter = aggCommitters.get(tableIdentifier);
             if (!(aggCommitter instanceof SupportMultiTableSinkAggregatedCommitter)) {
-                return;
+                break;
             }
             resourceManager =
                     ((SupportMultiTableSinkAggregatedCommitter<?>) aggCommitter)
                             .initMultiTableResourceManager(aggCommitters.size(), 1);
             break;
         }
-        if (resourceManager != null) {
-            for (String tableIdentifier : aggCommitters.keySet()) {
-                SinkAggregatedCommitter<?, ?> aggCommitter = aggCommitters.get(tableIdentifier);
-                aggCommitter.init();
+        for (SinkAggregatedCommitter<?, ?> aggCommitter : aggCommitters.values()) {
+            aggCommitter.init();
+            if (resourceManager != null) {
                 ((SupportMultiTableSinkAggregatedCommitter<?>) aggCommitter)
                         .setMultiTableResourceManager(resourceManager, 0);
             }

seatunnel-api/src/test/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkAggregatedCommitterTest.java

@@ -0,0 +1,73 @@
+/*
+ * 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.api.sink.multitablesink;
+
+import org.apache.seatunnel.api.sink.SinkAggregatedCommitter;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class MultiTableSinkAggregatedCommitterTest {
+
+    @Test
+    void testInitBeInvoked() throws IOException {
+        Map<String, SinkAggregatedCommitter<?, ?>> aggCommitters = new HashMap<>();
+        List<String> methodInvoked = new ArrayList<>();
+        aggCommitters.put(
+                "table1",
+                new SinkAggregatedCommitter<Object, Object>() {
+
+                    @Override
+                    public void init() {
+                        methodInvoked.add("init");
+                    }
+
+                    @Override
+                    public List<Object> commit(List<Object> aggregatedCommitInfo)
+                            throws IOException {
+                        return Collections.emptyList();
+                    }
+
+                    @Override
+                    public Object combine(List<Object> commitInfos) {
+                        return null;
+                    }
+
+                    @Override
+                    public void abort(List<Object> aggregatedCommitInfo) throws Exception {}
+
+                    @Override
+                    public void close() throws IOException {
+                        methodInvoked.add("close");
+                    }
+                });
+        MultiTableSinkAggregatedCommitter committer =
+                new MultiTableSinkAggregatedCommitter(aggCommitters);
+        committer.init();
+        committer.close();
+        Assertions.assertIterableEquals(Arrays.asList("init", "close"), methodInvoked);
+    }
+}

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/file/ClickhouseFileSinkAggCommitter.java

@@ -41,18 +41,24 @@ public class ClickhouseFileSinkAggCommitter
         implements SinkAggregatedCommitter<CKFileCommitInfo, CKFileAggCommitInfo> {
 
     private transient ClickhouseProxy proxy;
-    private final ClickhouseTable clickhouseTable;
+    private ClickhouseTable clickhouseTable;
 
     private final FileReaderOption fileReaderOption;
 
     public ClickhouseFileSinkAggCommitter(FileReaderOption readerOption) {
         fileReaderOption = readerOption;
-        proxy = new ClickhouseProxy(readerOption.getShardMetadata().getDefaultShard().getNode());
+    }
+
+    @Override
+    public void init() {
+        proxy =
+                new ClickhouseProxy(
+                        fileReaderOption.getShardMetadata().getDefaultShard().getNode());
         clickhouseTable =
                 proxy.getClickhouseTable(
                         proxy.getClickhouseConnection(),
-                        readerOption.getShardMetadata().getDatabase(),
-                        readerOption.getShardMetadata().getTable());
+                        fileReaderOption.getShardMetadata().getDatabase(),
+                        fileReaderOption.getShardMetadata().getTable());
     }
 
     @Override

seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/sink/SinkFlowTestUtils.java

@@ -20,10 +20,12 @@
 import org.apache.seatunnel.api.common.JobContext;
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.sink.DefaultSinkWriterContext;
+import org.apache.seatunnel.api.sink.MultiTableResourceManager;
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
 import org.apache.seatunnel.api.sink.SinkAggregatedCommitter;
 import org.apache.seatunnel.api.sink.SinkCommitter;
 import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.sink.SupportMultiTableSinkAggregatedCommitter;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.factory.TableSinkFactory;
 import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
@@ -108,16 +110,30 @@ private static void runWithContext(
         }
 
         Optional<? extends SinkCommitter<?>> sinkCommitter = sink.createCommitter();
-        Optional<? extends SinkAggregatedCommitter<?, ?>> aggregatedCommitter =
+        Optional<? extends SinkAggregatedCommitter<?, ?>> aggregatedCommitterOptional =
                 sink.createAggregatedCommitter();
 
         if (!commitInfos.isEmpty()) {
-            if (aggregatedCommitter.isPresent()) {
+            if (aggregatedCommitterOptional.isPresent()) {
+                SinkAggregatedCommitter<?, ?> aggregatedCommitter =
+                        aggregatedCommitterOptional.get();
+                MultiTableResourceManager resourceManager = null;
+                if (aggregatedCommitter instanceof SupportMultiTableSinkAggregatedCommitter) {
+                    resourceManager =
+                            ((SupportMultiTableSinkAggregatedCommitter<?>) aggregatedCommitter)
+                                    .initMultiTableResourceManager(1, 1);
+                }
+                aggregatedCommitter.init();
+                if (resourceManager != null) {
+                    ((SupportMultiTableSinkAggregatedCommitter<?>) aggregatedCommitter)
+                            .setMultiTableResourceManager(resourceManager, 0);
+                }
+
                 Object aggregatedCommitInfoT =
-                        ((SinkAggregatedCommitter) aggregatedCommitter.get()).combine(commitInfos);
-                ((SinkAggregatedCommitter) aggregatedCommitter.get())
+                        ((SinkAggregatedCommitter) aggregatedCommitter).combine(commitInfos);
+                ((SinkAggregatedCommitter) aggregatedCommitter)
                         .commit(Collections.singletonList(aggregatedCommitInfoT));
-                aggregatedCommitter.get().close();
+                aggregatedCommitter.close();
             } else if (sinkCommitter.isPresent()) {
                 ((SinkCommitter) sinkCommitter.get()).commit(commitInfos);
             } else {

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/commit/FileSinkAggregatedCommitter.java

@@ -33,8 +33,14 @@
 public class FileSinkAggregatedCommitter
         implements SinkAggregatedCommitter<FileCommitInfo, FileAggregatedCommitInfo> {
     protected HadoopFileSystemProxy hadoopFileSystemProxy;
+    private final HadoopConf hadoopConf;
 
     public FileSinkAggregatedCommitter(HadoopConf hadoopConf) {
+        this.hadoopConf = hadoopConf;
+    }
+
+    @Override
+    public void init() {
         this.hadoopFileSystemProxy = new HadoopFileSystemProxy(hadoopConf);
     }
 

seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/sink/commit/IcebergAggregatedCommitter.java

@@ -33,10 +33,18 @@
 public class IcebergAggregatedCommitter
         implements SinkAggregatedCommitter<IcebergCommitInfo, IcebergAggregatedCommitInfo> {
 
-    private final IcebergTableLoader tableLoader;
-    private final IcebergFilesCommitter filesCommitter;
+    private IcebergTableLoader tableLoader;
+    private IcebergFilesCommitter filesCommitter;
+    private final IcebergSinkConfig config;
+    private final CatalogTable catalogTable;
 
     public IcebergAggregatedCommitter(IcebergSinkConfig config, CatalogTable catalogTable) {
+        this.config = config;
+        this.catalogTable = catalogTable;
+    }
+
+    @Override
+    public void init() {
         this.tableLoader = IcebergTableLoader.create(config, catalogTable);
         this.filesCommitter = IcebergFilesCommitter.of(config, tableLoader);
     }

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkAggregatedCommitter.java

@@ -39,15 +39,19 @@
 public class JdbcSinkAggregatedCommitter
         implements SinkAggregatedCommitter<XidInfo, JdbcAggregatedCommitInfo> {
 
-    private final XaFacade xaFacade;
-    private final XaGroupOps xaGroupOps;
+    private XaFacade xaFacade;
+    private XaGroupOps xaGroupOps;
     private final JdbcSinkConfig jdbcSinkConfig;
 
     public JdbcSinkAggregatedCommitter(JdbcSinkConfig jdbcSinkConfig) {
+        this.jdbcSinkConfig = jdbcSinkConfig;
+    }
+
+    @Override
+    public void init() {
         this.xaFacade =
                 XaFacade.fromJdbcConnectionOptions(jdbcSinkConfig.getJdbcConnectionConfig());
         this.xaGroupOps = new XaGroupOpsImpl(xaFacade);
-        this.jdbcSinkConfig = jdbcSinkConfig;
     }
 
     private void tryOpen() throws IOException {