merge 1.8

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/environment/StreamEnvConfigManager.java

@@ -91,25 +91,15 @@ public static void streamExecutionEnvironmentConfig(StreamExecutionEnvironment s
             ((Configuration) exeConfig.getGlobalJobParameters()).addAll(globalJobParameters);
         }
 
-        Optional<Integer> envParallelism = getEnvParallelism(confProperties);
-        if (envParallelism.isPresent()) {
-            streamEnv.setParallelism(envParallelism.get());
-        }
-
-        Optional<Integer> maxParallelism = getMaxEnvParallelism(confProperties);
-        if (maxParallelism.isPresent()) {
-            streamEnv.setMaxParallelism(maxParallelism.get());
-        }
-
-        Optional<Long> bufferTimeoutMillis = getBufferTimeoutMillis(confProperties);
-        if (bufferTimeoutMillis.isPresent()) {
-            streamEnv.setBufferTimeout(bufferTimeoutMillis.get());
-        }
-
-        Optional<TimeCharacteristic> streamTimeCharacteristic = getStreamTimeCharacteristic(confProperties);
-        if (streamTimeCharacteristic.isPresent()) {
-            streamEnv.setStreamTimeCharacteristic(streamTimeCharacteristic.get());
-        }
+        getEnvParallelism(confProperties).ifPresent(streamEnv::setParallelism);
+        getMaxEnvParallelism(confProperties).ifPresent(streamEnv::setMaxParallelism);
+        getBufferTimeoutMillis(confProperties).ifPresent(streamEnv::setBufferTimeout);
+        getStreamTimeCharacteristic(confProperties).ifPresent(streamEnv::setStreamTimeCharacteristic);
+        getAutoWatermarkInterval(confProperties).ifPresent(op -> {
+            if (streamEnv.getStreamTimeCharacteristic().equals(TimeCharacteristic.EventTime)) {
+                streamEnv.getConfig().setAutoWatermarkInterval(op);
+            }
+        });
 
         streamEnv.setRestartStrategy(RestartStrategies.failureRateRestart(
                 ConfigConstrant.failureRate,
@@ -120,32 +110,12 @@ public static void streamExecutionEnvironmentConfig(StreamExecutionEnvironment s
         // checkpoint config
         Optional<Boolean> checkpointingEnabled = isCheckpointingEnabled(confProperties);
         if (checkpointingEnabled.get()) {
-            Optional<Long> checkpointInterval = getCheckpointInterval(confProperties);
-            streamEnv.enableCheckpointing(checkpointInterval.get());
-
-            Optional<CheckpointingMode> checkpointingMode = getCheckpointingMode(confProperties);
-            if (checkpointingMode.isPresent()) {
-                streamEnv.getCheckpointConfig().setCheckpointingMode(checkpointingMode.get());
-            }
-            Optional<Long> checkpointTimeout = getCheckpointTimeout(confProperties);
-            if (checkpointTimeout.isPresent()) {
-                streamEnv.getCheckpointConfig().setCheckpointTimeout(checkpointTimeout.get());
-            }
-
-            Optional<Integer> maxConcurrentCheckpoints = getMaxConcurrentCheckpoints(confProperties);
-            if (maxConcurrentCheckpoints.isPresent()) {
-                streamEnv.getCheckpointConfig().setMaxConcurrentCheckpoints(maxConcurrentCheckpoints.get());
-            }
-
-            Optional<CheckpointConfig.ExternalizedCheckpointCleanup> checkpointCleanup = getCheckpointCleanup(confProperties);
-            if (checkpointCleanup.isPresent()) {
-                streamEnv.getCheckpointConfig().enableExternalizedCheckpoints(checkpointCleanup.get());
-            }
-
-            Optional<StateBackend> stateBackend = getStateBackend(confProperties);
-            if (stateBackend.isPresent()) {
-                streamEnv.setStateBackend(stateBackend.get());
-            }
+            getCheckpointInterval(confProperties).ifPresent(streamEnv::enableCheckpointing);
+            getCheckpointingMode(confProperties).ifPresent(streamEnv.getCheckpointConfig()::setCheckpointingMode);
+            getCheckpointTimeout(confProperties).ifPresent(streamEnv.getCheckpointConfig()::setCheckpointTimeout);
+            getMaxConcurrentCheckpoints(confProperties).ifPresent(streamEnv.getCheckpointConfig()::setMaxConcurrentCheckpoints);
+            getCheckpointCleanup(confProperties).ifPresent(streamEnv.getCheckpointConfig()::enableExternalizedCheckpoints);
+            getStateBackend(confProperties).ifPresent(streamEnv::setStateBackend);
         }
     }
 
@@ -181,6 +151,11 @@ public static Optional<Long> getBufferTimeoutMillis(Properties properties) {
         return StringUtils.isNotBlank(mills) ? Optional.of(Long.valueOf(mills)) : Optional.empty();
     }
 
+    public static Optional<Long> getAutoWatermarkInterval(Properties properties) {
+        String autoWatermarkInterval = properties.getProperty(ConfigConstrant.AUTO_WATERMARK_INTERVAL_KEY);
+        return StringUtils.isNotBlank(autoWatermarkInterval) ? Optional.of(Long.valueOf(autoWatermarkInterval)) : Optional.empty();
+    }
+
     /**
      * #ProcessingTime(默认), IngestionTime, EventTime
      * @param properties
@@ -191,7 +166,7 @@ public static Optional<TimeCharacteristic> getStreamTimeCharacteristic(Propertie
         }
         String characteristicStr = properties.getProperty(ConfigConstrant.FLINK_TIME_CHARACTERISTIC_KEY);
         Optional<TimeCharacteristic> characteristic = Arrays.stream(TimeCharacteristic.values())
-                .filter(tc -> !characteristicStr.equalsIgnoreCase(tc.toString())).findAny();
+                .filter(tc -> characteristicStr.equalsIgnoreCase(tc.toString())).findAny();
 
         if (!characteristic.isPresent()) {
             throw new RuntimeException("illegal property :" + ConfigConstrant.FLINK_TIME_CHARACTERISTIC_KEY);
@@ -207,9 +182,9 @@ public static Optional<Boolean> isCheckpointingEnabled(Properties properties) {
 
     public static Optional<Long> getCheckpointInterval(Properties properties) {
         // 两个参数主要用来做上层兼容
-        Long sql_interval = Long.valueOf(properties.getProperty(ConfigConstrant.SQL_CHECKPOINT_INTERVAL_KEY, "0"));
-        Long flink_interval = Long.valueOf(properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_INTERVAL_KEY, "0"));
-        long checkpointInterval = Math.max(sql_interval, flink_interval);
+        Long sqlInterval = Long.valueOf(properties.getProperty(ConfigConstrant.SQL_CHECKPOINT_INTERVAL_KEY, "0"));
+        Long flinkInterval = Long.valueOf(properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_INTERVAL_KEY, "0"));
+        long checkpointInterval = Math.max(sqlInterval, flinkInterval);
         return Optional.of(checkpointInterval);
     }
 

launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/PerJobSubmitter.java

@@ -62,10 +62,9 @@ public static String submit(Options launcherOptions, JobGraph jobGraph, Configur
         ClusterSpecification clusterSpecification = FLinkPerJobResourceUtil.createClusterSpecification(confProperties);
 
         PerJobClusterClientBuilder perJobClusterClientBuilder = new PerJobClusterClientBuilder();
-        perJobClusterClientBuilder.init(launcherOptions.getYarnconf(), flinkConfig, confProperties);
+        perJobClusterClientBuilder.init(launcherOptions.getYarnconf(), confProperties);
 
         String flinkJarPath = launcherOptions.getFlinkJarPath();
-        AbstractYarnClusterDescriptor yarnClusterDescriptor = perJobClusterClientBuilder.createPerJobClusterDescriptor(flinkJarPath, launcherOptions, jobGraph);
 
         AbstractYarnClusterDescriptor yarnClusterDescriptor = perJobClusterClientBuilder.createPerJobClusterDescriptor(confProperties, flinkJarPath, launcherOptions, jobGraph, flinkConfig);
         ClusterClient<ApplicationId> clusterClient = yarnClusterDescriptor.deployJobCluster(clusterSpecification, jobGraph,true);