[FLINK-26537][statefun] Allow disabling StatefulFunctionsConfigValidator validation for classloader.parent-first-patterns.additional

Author: FilKarnicki

docs/content/docs/deployment/configurations.md

@@ -75,5 +75,14 @@ These may be set through your job's ``flink-conf.yaml``.
             <td>Integer</td>
             <td>The max number of async operations per task before backpressure is applied.</td>
         </tr>
+        <tr>
+            <td><h5>statefun.embedded</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Set to 'true' if Flink is running this job from an uber jar, rather than using statefun-specific docker images.
+                This disables the validation of whether 'classloader.parent-first-patterns.additional' 
+                contains 'org.apache.flink.statefun', 'org.apache.kafka' and 'com.google.protobuf' patterns.
+                It is then up to the creator of the uber jar to ensure that the three dependencies (statefun, kafka and protobuf) don't have version conflicts.</td>
+        </tr>
 	</tbody>
 </table>

statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfig.java

@@ -107,6 +107,13 @@ public class StatefulFunctionsConfig implements Serializable {
           .withDescription(
               "The name of the remote module entity to look for. Also supported, file:///...");
 
+  public static final ConfigOption<Boolean> EMBEDDED =
+      ConfigOptions.key("statefun.embedded")
+          .booleanType()
+          .defaultValue(false)
+          .withDescription(
+              "True if Flink is running this job from an uber jar, rather than using statefun-specific docker images");
+
   /**
    * Creates a new {@link StatefulFunctionsConfig} based on the default configurations in the
    * current environment set via the {@code flink-conf.yaml}.
@@ -134,7 +141,9 @@ public static StatefulFunctionsConfig fromFlinkConfiguration(Configuration flink
 
   private String remoteModuleName;
 
-  private Map<String, String> globalConfigurations = new HashMap<>();
+  private boolean embedded;
+
+  private final Map<String, String> globalConfigurations = new HashMap<>();
 
   /**
    * Create a new configuration object based on the values set in flink-conf.
@@ -149,6 +158,7 @@ private StatefulFunctionsConfig(Configuration configuration) {
     this.feedbackBufferSize = configuration.get(TOTAL_MEMORY_USED_FOR_FEEDBACK_CHECKPOINTING);
     this.maxAsyncOperationsPerTask = configuration.get(ASYNC_MAX_OPERATIONS_PER_TASK);
     this.remoteModuleName = configuration.get(REMOTE_MODULE_NAME);
+    this.embedded = configuration.getBoolean(EMBEDDED);
 
     for (String key : configuration.keySet()) {
       if (key.startsWith(MODULE_CONFIG_PREFIX)) {
@@ -234,6 +244,19 @@ public void setRemoteModuleName(String remoteModuleName) {
     this.remoteModuleName = Objects.requireNonNull(remoteModuleName);
   }
 
+  /** Returns whether the job was launched in embedded mode (see {@linkplain #EMBEDDED}). */
+  public boolean isEmbedded() {
+    return embedded;
+  }
+
+  /**
+   * Sets the embedded mode. If true, disables certain validation steps. See documentation:
+   * Configurations.
+   */
+  public void setEmbedded(boolean embedded) {
+    this.embedded = embedded;
+  }
+
   /**
    * Retrieves the universe provider for loading modules.
    *

statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java

@@ -18,12 +18,7 @@
 
 package org.apache.flink.statefun.flink.core;
 
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Locale;
-import java.util.Set;
+import java.util.*;
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
 import org.apache.flink.configuration.Configuration;
@@ -42,8 +37,10 @@ private StatefulFunctionsConfigValidator() {}
 
   public static final int MAX_CONCURRENT_CHECKPOINTS = 1;
 
-  static void validate(Configuration configuration) {
-    validateParentFirstClassloaderPatterns(configuration);
+  static void validate(boolean isEmbedded, Configuration configuration) {
+    if (!isEmbedded) {
+      validateParentFirstClassloaderPatterns(configuration);
+    }
     validateCustomPayloadSerializerClassName(configuration);
     validateNoHeapBackedTimers(configuration);
     validateUnalignedCheckpointsDisabled(configuration);
@@ -70,10 +67,9 @@ private static Set<String> parentFirstClassloaderPatterns(Configuration configur
   }
 
   private static void validateCustomPayloadSerializerClassName(Configuration configuration) {
-
-    MessageFactoryType factoryType =
+    final MessageFactoryType factoryType =
         configuration.get(StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER);
-    String customPayloadSerializerClassName =
+    final String customPayloadSerializerClassName =
         configuration.get(StatefulFunctionsConfig.USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS);
 
     if (factoryType == MessageFactoryType.WITH_CUSTOM_PAYLOADS) {

statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsJob.java

@@ -19,7 +19,6 @@
 
 import java.net.URL;
 import java.net.URLClassLoader;
-import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.flink.api.java.utils.ParameterTool;
@@ -36,19 +35,21 @@ public class StatefulFunctionsJob {
   private static final AtomicInteger FEEDBACK_INVOCATION_ID_SEQ = new AtomicInteger();
 
   public static void main(String... args) throws Exception {
-    ParameterTool parameterTool = ParameterTool.fromArgs(args);
-    Map<String, String> globalConfigurations = parameterTool.toMap();
-
+    ParameterTool argsParameterTool = ParameterTool.fromArgs(args);
     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
     Configuration flinkConfig = FlinkConfigExtractor.reflectivelyExtractFromEnv(env);
-    StatefulFunctionsConfigValidator.validate(flinkConfig);
 
     StatefulFunctionsConfig stateFunConfig =
-        StatefulFunctionsConfig.fromFlinkConfiguration(flinkConfig);
-    stateFunConfig.addAllGlobalConfigurations(globalConfigurations);
+        StatefulFunctionsConfig.fromFlinkConfiguration(
+            ParameterTool.fromMap(flinkConfig.toMap())
+                .mergeWith(argsParameterTool)
+                .getConfiguration());
+
+    stateFunConfig.addAllGlobalConfigurations(argsParameterTool.toMap());
     stateFunConfig.setProvider(new StatefulFunctionsUniverses.ClassPathUniverseProvider());
 
+    StatefulFunctionsConfigValidator.validate(stateFunConfig.isEmbedded(), flinkConfig);
+
     main(env, stateFunConfig);
   }
 

statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigTest.java

@@ -91,7 +91,7 @@ public void invalidCustomSerializerThrows() {
     Configuration configuration = baseConfiguration();
     configuration.set(
         StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER, MessageFactoryType.WITH_CUSTOM_PAYLOADS);
-    StatefulFunctionsConfigValidator.validate(configuration);
+    StatefulFunctionsConfigValidator.validate(false, configuration);
   }
 
   @Test(expected = StatefulFunctionsInvalidConfigException.class)
@@ -101,6 +101,6 @@ public void invalidNonCustomSerializerThrows() {
         StatefulFunctionsConfig.USER_MESSAGE_SERIALIZER, MessageFactoryType.WITH_KRYO_PAYLOADS);
     configuration.set(
         StatefulFunctionsConfig.USER_MESSAGE_CUSTOM_PAYLOAD_SERIALIZER_CLASS, serializerClassName);
-    StatefulFunctionsConfigValidator.validate(configuration);
+    StatefulFunctionsConfigValidator.validate(false, configuration);
   }
 }