[FLINK-23498][streaming-java] Expose an easier StreamExecutionEnviron…

…ment.configure()
Treydone · Aug 4, 2021 · 5b2535b · 5b2535b
1 parent bf55428
commit 5b2535b
Show file tree

Hide file tree

Showing 2 changed files with 33 additions and 0 deletions.
diff --git a/.../src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/.../src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -930,6 +930,21 @@ public TimeCharacteristic getStreamTimeCharacteristic() {
         return timeCharacteristic;
     }
 
+    /**
+     * Sets all relevant options contained in the {@link ReadableConfig} such as e.g. {@link
+     * StreamPipelineOptions#TIME_CHARACTERISTIC}. It will reconfigure {@link
+     * StreamExecutionEnvironment}, {@link ExecutionConfig} and {@link CheckpointConfig}.
+     *
+     * <p>It will change the value of a setting only if a corresponding option was set in the {@code
+     * configuration}. If a key is not present, the current value of a field will remain untouched.
+     *
+     * @param configuration a configuration to read the values from
+     */
+    @PublicEvolving
+    public void configure(ReadableConfig configuration) {
+        configure(configuration, userClassloader);
+    }
+
     /**
      * Sets all relevant options contained in the {@link ReadableConfig} such as e.g. {@link
      * StreamPipelineOptions#TIME_CHARACTERISTIC}. It will reconfigure {@link

diff --git a/...cala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala b/...cala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
@@ -572,6 +572,24 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
     javaEnv.configure(configuration, classLoader)
   }
 
+  /**
+   * Sets all relevant options contained in the [[ReadableConfig]] such as e.g.
+   * [[org.apache.flink.streaming.api.environment.StreamPipelineOptions#TIME_CHARACTERISTIC]].
+   * It will reconfigure [[StreamExecutionEnvironment]],
+   * [[org.apache.flink.api.common.ExecutionConfig]] and
+   * [[org.apache.flink.streaming.api.environment.CheckpointConfig]].
+   *
+   * It will change the value of a setting only if a corresponding option was set in the
+   * `configuration`. If a key is not present, the current value of a field will remain
+   * untouched.
+   *
+   * @param configuration a configuration to read the values from
+   */
+  @PublicEvolving
+  def configure(configuration: ReadableConfig): Unit = {
+    javaEnv.configure(configuration)
+  }
+
   // --------------------------------------------------------------------------------------------
   // Data stream creations
   // --------------------------------------------------------------------------------------------