[Bug][Connector] Fixed ConsoleSink submit job twice. (#1710)

* fix ConsoleSink submit job twice
apache · Apr 18, 2022 · ec62ef0 · ec62ef0
1 parent 34090bf
commit ec62ef0
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 8 deletions.
diff --git a/...atunnel-api-flink/src/main/java/org/apache/seatunnel/flink/batch/FlinkBatchExecution.java b/...atunnel-api-flink/src/main/java/org/apache/seatunnel/flink/batch/FlinkBatchExecution.java
@@ -72,13 +72,15 @@ public void start(List<FlinkBatchSource> sources, List<FlinkBatchTransform> tran
             sink.outputBatch(flinkEnvironment, dataSet);
         }
 
-        try {
-            LOGGER.info("Flink Execution Plan:{}", flinkEnvironment.getBatchEnvironment().getExecutionPlan());
-            JobExecutionResult execute = flinkEnvironment.getBatchEnvironment().execute(flinkEnvironment.getJobName());
-            LOGGER.info(execute.toString());
-        } catch (Exception e) {
-            LOGGER.warn("Flink with job name [{}] execute failed", flinkEnvironment.getJobName());
-            throw e;
+        if (whetherExecute(sinks)) {
+            try {
+                LOGGER.info("Flink Execution Plan:{}", flinkEnvironment.getBatchEnvironment().getExecutionPlan());
+                JobExecutionResult execute = flinkEnvironment.getBatchEnvironment().execute(flinkEnvironment.getJobName());
+                LOGGER.info(execute.toString());
+            } catch (Exception e) {
+                LOGGER.warn("Flink with job name [{}] execute failed", flinkEnvironment.getJobName());
+                throw e;
+            }
         }
     }
 
@@ -116,4 +118,7 @@ public Config getConfig() {
         return config;
     }
 
+    private boolean whetherExecute(List<FlinkBatchSink> sinks) {
+        return sinks.stream().anyMatch(s -> !"ConsoleSink".equals(s.getPluginName()));
+    }
 }
diff --git a/...ctor-flink-console/src/main/java/org/apache/seatunnel/flink/console/sink/ConsoleSink.java b/...ctor-flink-console/src/main/java/org/apache/seatunnel/flink/console/sink/ConsoleSink.java
@@ -48,7 +48,6 @@ public void outputBatch(FlinkEnvironment env, DataSet<Row> rowDataSet) {
         } catch (Exception e) {
             LOGGER.error("Failed to print result! ", e);
         }
-        rowDataSet.output(this);
     }
 
     @Override