[Improve][API] Move catalog open to SaveModeHandler

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

@@ -59,6 +59,11 @@ public DefaultSaveModeHandler(
                 customSql);
     }
 
+    @Override
+    public void open() {
+        catalog.open();
+    }
+
     @Override
     public void handleSchemaSaveMode() {
         switch (schemaSaveMode) {

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

@@ -22,6 +22,8 @@
 
 public interface SaveModeHandler extends AutoCloseable {
 
+    void open();
+
     void handleSchemaSaveMode();
 
     void handleDataSaveMode();

seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/DorisSink.java

@@ -125,7 +125,6 @@ public Optional<SaveModeHandler> getSaveModeHandler() {
         }
 
         Catalog catalog = catalogFactory.createCatalog(catalogFactory.factoryIdentifier(), config);
-        catalog.open();
         return Optional.of(
                 new DefaultSaveModeHandler(
                         config.get(DorisOptions.SCHEMA_SAVE_MODE),

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSink.java

@@ -91,7 +91,6 @@ public Optional<SaveModeHandler> getSaveModeHandler() {
         DataSaveMode dataSaveMode = config.get(SinkConfig.DATA_SAVE_MODE);
 
         TablePath tablePath = TablePath.of("", catalogTable.getTableId().getTableName());
-        catalog.open();
         return Optional.of(
                 new DefaultSaveModeHandler(
                         schemaSaveMode, dataSaveMode, catalog, tablePath, null, null));

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

@@ -130,7 +130,6 @@ public Optional<SaveModeHandler> getSaveModeHandler() {
         }
         Catalog catalog =
                 catalogFactory.createCatalog(catalogFactory.factoryIdentifier(), readonlyConfig);
-        catalog.open();
         return Optional.of(
                 new DefaultSaveModeHandler(
                         config.getSchemaSaveMode(),

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

@@ -198,7 +198,6 @@ public Optional<SaveModeHandler> getSaveModeHandler() {
             if (catalogOptional.isPresent()) {
                 try {
                     Catalog catalog = catalogOptional.get();
-                    catalog.open();
                     FieldIdeEnum fieldIdeEnumEnum = config.get(JdbcOptions.FIELD_IDE);
                     String fieldIde =
                             fieldIdeEnumEnum == null

seatunnel-connectors-v2/connector-milvus/src/main/java/org/apache/seatunnel/connectors/seatunnel/milvus/sink/MilvusSink.java

@@ -103,7 +103,6 @@ public Optional<SaveModeHandler> getSaveModeHandler() {
         SchemaSaveMode schemaSaveMode = config.get(MilvusSinkConfig.SCHEMA_SAVE_MODE);
         DataSaveMode dataSaveMode = config.get(MilvusSinkConfig.DATA_SAVE_MODE);
 
-        catalog.open();
         return Optional.of(
                 new DefaultSaveModeHandler(
                         schemaSaveMode,

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/sink/PaimonSink.java

@@ -144,7 +144,6 @@ public Optional<SaveModeHandler> getSaveModeHandler() {
         }
         org.apache.seatunnel.api.table.catalog.Catalog catalog =
                 catalogFactory.createCatalog(catalogFactory.factoryIdentifier(), readonlyConfig);
-        catalog.open();
         return Optional.of(
                 new PaimonSaveModeHandler(
                         this,

seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSink.java

@@ -79,7 +79,6 @@ public Optional<SaveModeHandler> getSaveModeHandler() {
                         sinkConfig.getPassword(),
                         sinkConfig.getJdbcUrl(),
                         sinkConfig.getSaveModeCreateTemplate());
-        catalog.open();
         return Optional.of(
                 new DefaultSaveModeHandler(
                         schemaSaveMode,

seatunnel-core/seatunnel-flink-starter/seatunnel-flink-13-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java

@@ -130,6 +130,7 @@ public List<DataStreamTableInfo> execute(List<DataStreamTableInfo> upstreamDataS
                 Optional<SaveModeHandler> saveModeHandler = saveModeSink.getSaveModeHandler();
                 if (saveModeHandler.isPresent()) {
                     try (SaveModeHandler handler = saveModeHandler.get()) {
+                        handler.open();
                         new SaveModeExecuteWrapper(handler).execute();
                     } catch (Exception e) {
                         throw new SeaTunnelRuntimeException(HANDLE_SAVE_MODE_FAILED, e);

seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java

@@ -131,6 +131,7 @@ public List<DataStreamTableInfo> execute(List<DataStreamTableInfo> upstreamDataS
                 Optional<SaveModeHandler> saveModeHandler = saveModeSink.getSaveModeHandler();
                 if (saveModeHandler.isPresent()) {
                     try (SaveModeHandler handler = saveModeHandler.get()) {
+                        handler.open();
                         new SaveModeExecuteWrapper(handler).execute();
                     } catch (Exception e) {
                         throw new SeaTunnelRuntimeException(HANDLE_SAVE_MODE_FAILED, e);

seatunnel-core/seatunnel-spark-starter/seatunnel-spark-2-starter/src/main/java/org/apache/seatunnel/core/starter/spark/execution/SinkExecuteProcessor.java

@@ -122,6 +122,7 @@ public List<DatasetTableInfo> execute(List<DatasetTableInfo> upstreamDataStreams
                 Optional<SaveModeHandler> saveModeHandler = saveModeSink.getSaveModeHandler();
                 if (saveModeHandler.isPresent()) {
                     try (SaveModeHandler handler = saveModeHandler.get()) {
+                        handler.open();
                         new SaveModeExecuteWrapper(handler).execute();
                     } catch (Exception e) {
                         throw new SeaTunnelRuntimeException(HANDLE_SAVE_MODE_FAILED, e);

seatunnel-core/seatunnel-spark-starter/seatunnel-spark-starter-common/src/main/java/org/apache/seatunnel/core/starter/spark/execution/SinkExecuteProcessor.java

@@ -122,6 +122,7 @@ public List<DatasetTableInfo> execute(List<DatasetTableInfo> upstreamDataStreams
                 Optional<SaveModeHandler> saveModeHandler = saveModeSink.getSaveModeHandler();
                 if (saveModeHandler.isPresent()) {
                     try (SaveModeHandler handler = saveModeHandler.get()) {
+                        handler.open();
                         new SaveModeExecuteWrapper(handler).execute();
                     } catch (Exception e) {
                         throw new SeaTunnelRuntimeException(HANDLE_SAVE_MODE_FAILED, e);

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisCatalogIT.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.e2e.connector.doris;
 
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.sink.SaveModeHandler;
 import org.apache.seatunnel.api.sink.SupportSaveMode;
 import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
@@ -245,7 +246,9 @@ private CatalogTable assertCreateTable(
                         Thread.currentThread().getContextClassLoader(),
                         Collections.emptyList());
         SupportSaveMode sink = (SupportSaveMode) dorisSinkFactory.createSink(context).createSink();
-        sink.getSaveModeHandler().get().handleSaveMode();
+        SaveModeHandler handler = sink.getSaveModeHandler().get();
+        handler.open();
+        handler.handleSaveMode();
         CatalogTable createdTable = catalog.getTable(TablePath.of(fullName));
         Assertions.assertEquals(
                 upstreamTable.getTableSchema().getColumns().size(),

seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/sink/inmemory/InMemorySaveModeHandler.java

@@ -42,6 +42,9 @@ public InMemorySaveModeHandler(CatalogTable catalogTable) {
         this.catalogTable = catalogTable;
     }
 
+    @Override
+    public void open() {}
+
     @Override
     public void handleSchemaSaveMode() {
         log.info("handle schema savemode with table path: {}", catalogTable.getTablePath());

seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/MultipleTableJobConfigParser.java

@@ -698,6 +698,7 @@ public void handleSaveMode(SeaTunnelSink<?, ?, ?, ?> sink) {
                 Optional<SaveModeHandler> saveModeHandler = saveModeSink.getSaveModeHandler();
                 if (saveModeHandler.isPresent()) {
                     try (SaveModeHandler handler = saveModeHandler.get()) {
+                        handler.open();
                         new SaveModeExecuteWrapper(handler).execute();
                     } catch (Exception e) {
                         throw new SeaTunnelRuntimeException(HANDLE_SAVE_MODE_FAILED, e);

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java

@@ -372,6 +372,7 @@ public static void handleSaveMode(SeaTunnelSink sink) {
                     ((SupportSaveMode) sink).getSaveModeHandler();
             if (saveModeHandler.isPresent()) {
                 try (SaveModeHandler handler = saveModeHandler.get()) {
+                    handler.open();
                     new SaveModeExecuteWrapper(handler).execute();
                 } catch (Exception e) {
                     throw new SeaTunnelRuntimeException(HANDLE_SAVE_MODE_FAILED, e);