restore SaveModeHandler.open() method

apache · Sep 14, 2024 · 1f52ce1 · 1f52ce1
1 parent 9f94ead
commit 1f52ce1
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 0 deletions.
diff --git a/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java b/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java
@@ -153,6 +153,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);

diff --git a/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java b/...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java
@@ -155,6 +155,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);

diff --git a/...re/src/main/java/org/apache/seatunnel/engine/core/parse/MultipleTableJobConfigParser.java b/...re/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);