rename configureFailFastOnScalaTypeResolutionWithClass() to copyWithExtra() and make it side-effect free

Author: arnaud-daroussin

modules/scala-api/src/main/scala/org/apache/flinkx/api/StreamExecutionEnvironment.scala

@@ -827,9 +827,8 @@ object StreamExecutionEnvironment {
     * cluster.
     */
   def getExecutionEnvironment: StreamExecutionEnvironment = {
-    val configuration = new Configuration
-    configureFailFastOnScalaTypeResolutionWithClass(configuration)
-    new StreamExecutionEnvironment(JavaEnv.getExecutionEnvironment(configuration))
+    val extraConfig = copyWithExtra(new Configuration)
+    new StreamExecutionEnvironment(JavaEnv.getExecutionEnvironment(extraConfig))
   }
 
   /** Creates an execution environment that represents the context in which the program is currently executed.
@@ -838,8 +837,8 @@ object StreamExecutionEnvironment {
     *   Pass a custom configuration into the cluster.
     */
   def getExecutionEnvironment(configuration: Configuration): StreamExecutionEnvironment = {
-    configureFailFastOnScalaTypeResolutionWithClass(configuration)
-    new StreamExecutionEnvironment(JavaEnv.getExecutionEnvironment(configuration))
+    val extraConfig = copyWithExtra(configuration)
+    new StreamExecutionEnvironment(JavaEnv.getExecutionEnvironment(extraConfig))
   }
 
   // --------------------------------------------------------------------------
@@ -853,9 +852,8 @@ object StreamExecutionEnvironment {
     * [[setDefaultLocalParallelism(Int)]].
     */
   def createLocalEnvironment(parallelism: Int = JavaEnv.getDefaultLocalParallelism): StreamExecutionEnvironment = {
-    val configuration = new Configuration
-    configureFailFastOnScalaTypeResolutionWithClass(configuration)
-    new StreamExecutionEnvironment(JavaEnv.createLocalEnvironment(parallelism, configuration))
+    val extraConfig = copyWithExtra(new Configuration)
+    new StreamExecutionEnvironment(JavaEnv.createLocalEnvironment(parallelism, extraConfig))
   }
 
   /** Creates a local execution environment. The local execution environment will run the program in a multi-threaded
@@ -867,8 +865,8 @@ object StreamExecutionEnvironment {
     *   Pass a custom configuration into the cluster.
     */
   def createLocalEnvironment(parallelism: Int, configuration: Configuration): StreamExecutionEnvironment = {
-    configureFailFastOnScalaTypeResolutionWithClass(configuration)
-    new StreamExecutionEnvironment(JavaEnv.createLocalEnvironment(parallelism, configuration))
+    val extraConfig = copyWithExtra(configuration)
+    new StreamExecutionEnvironment(JavaEnv.createLocalEnvironment(parallelism, extraConfig))
   }
 
   /** Creates a [[StreamExecutionEnvironment]] for local program execution that also starts the web monitoring UI.
@@ -886,9 +884,8 @@ object StreamExecutionEnvironment {
     */
   @PublicEvolving
   def createLocalEnvironmentWithWebUI(config: Configuration = null): StreamExecutionEnvironment = {
-    val conf: Configuration = if (config == null) new Configuration() else config
-    configureFailFastOnScalaTypeResolutionWithClass(conf)
-    new StreamExecutionEnvironment(JavaEnv.createLocalEnvironmentWithWebUI(conf))
+    val extraConfig = copyWithExtra(if (config == null) new Configuration() else config)
+    new StreamExecutionEnvironment(JavaEnv.createLocalEnvironmentWithWebUI(extraConfig))
   }
 
   // --------------------------------------------------------------------------
@@ -909,9 +906,8 @@ object StreamExecutionEnvironment {
     *   user-defined input formats, or any libraries, those must be provided in the JAR files.
     */
   def createRemoteEnvironment(host: String, port: Int, jarFiles: String*): StreamExecutionEnvironment = {
-    val configuration = new Configuration
-    configureFailFastOnScalaTypeResolutionWithClass(configuration)
-    new StreamExecutionEnvironment(JavaEnv.createRemoteEnvironment(host, port, configuration, jarFiles: _*))
+    val extraConfig = copyWithExtra(new Configuration)
+    new StreamExecutionEnvironment(JavaEnv.createRemoteEnvironment(host, port, extraConfig, jarFiles: _*))
   }
 
   /** Creates a remote execution environment. The remote environment sends (parts of) the program to a cluster for
@@ -934,9 +930,8 @@ object StreamExecutionEnvironment {
       parallelism: Int,
       jarFiles: String*
   ): StreamExecutionEnvironment = {
-    val configuration = new Configuration
-    configureFailFastOnScalaTypeResolutionWithClass(configuration)
-    val javaEnv = JavaEnv.createRemoteEnvironment(host, port, configuration, jarFiles: _*)
+    val extraConfig = copyWithExtra(new Configuration)
+    val javaEnv = JavaEnv.createRemoteEnvironment(host, port, extraConfig, jarFiles: _*)
     javaEnv.setParallelism(parallelism)
     new StreamExecutionEnvironment(javaEnv)
   }
@@ -949,7 +944,7 @@ object StreamExecutionEnvironment {
     *   The host name or address of the master (JobManager), where the program should be executed.
     * @param port
     *   The port of the master (JobManager), where the program should be executed.
-    * @param configuration
+    * @param config
     *   Pass a custom configuration into the cluster.
     * @param jarFiles
     *   The JAR files with code that needs to be shipped to the cluster. If the program uses user-defined functions,
@@ -961,22 +956,30 @@ object StreamExecutionEnvironment {
       config: Configuration,
       jarFiles: String*
   ): StreamExecutionEnvironment = {
-    configureFailFastOnScalaTypeResolutionWithClass(config)
-    val javaEnv = JavaEnv.createRemoteEnvironment(host, port, config, jarFiles: _*)
+    val extraConfig = copyWithExtra(config)
+    val javaEnv = JavaEnv.createRemoteEnvironment(host, port, extraConfig, jarFiles: _*)
     new StreamExecutionEnvironment(javaEnv)
   }
 
-  private def configureFailFastOnScalaTypeResolutionWithClass(configuration: Configuration): Unit = {
+  /**
+   * Copy input config and add extra configuration:
+   *  - register type info factories to fail-fast on Scala type resolution with Class
+   * @param config input configuration
+   * @return a copy of input config with extra configuration
+   */
+  private def copyWithExtra(config: Configuration): Configuration = {
     if (!isFailFastOnScalaTypeResolutionWithClassConfigured && !isFailFastOnScalaTypeResolutionWithClassDisabled) {
       isFailFastOnScalaTypeResolutionWithClassConfigured = true
       val serializationOption = ConfigOptions.key("pipeline.serialization-config").stringType().asList().noDefaultValue()
-      val serializationConfig = configuration.getOptional(serializationOption).orElse(new util.ArrayList[String])
+      val serializationConfig = config.getOptional(serializationOption).orElse(new util.ArrayList[String])
       serializationConfig.add("scala.Product: {type: typeinfo, class: org.apache.flinkx.api.typeinfo.FailFastTypeInfoFactory}")
       serializationConfig.add("scala.Option: {type: typeinfo, class: org.apache.flinkx.api.typeinfo.FailFastTypeInfoFactory}")
       serializationConfig.add("scala.util.Either: {type: typeinfo, class: org.apache.flinkx.api.typeinfo.FailFastTypeInfoFactory}")
       serializationConfig.add("scala.Array: {type: typeinfo, class: org.apache.flinkx.api.typeinfo.FailFastTypeInfoFactory}")
       serializationConfig.add("scala.collection.Iterable: {type: typeinfo, class: org.apache.flinkx.api.typeinfo.FailFastTypeInfoFactory}")
-      configuration.set(serializationOption, serializationConfig)
+      config.clone().set(serializationOption, serializationConfig)
+    } else {
+      config
     }
   }