diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/ReflectiveFlinkConfigExtractor.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/FlinkConfigExtractor.java
similarity index 80%
rename from statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/ReflectiveFlinkConfigExtractor.java
rename to statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/FlinkConfigExtractor.java
index 72bc44c..30dfba0 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/ReflectiveFlinkConfigExtractor.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/FlinkConfigExtractor.java
@@ -23,9 +23,14 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
-class ReflectiveFlinkConfigExtractor {
+final class FlinkConfigExtractor {
 
-  static Configuration extractFromEnv(StreamExecutionEnvironment env) {
+  /**
+   * Reflectively extracts Flink {@link Configuration} from a {@link StreamExecutionEnvironment}.
+   * The Flink configuration contains Stateful Functions specific configurations.
+   * This is currently a private method in the {@code StreamExecutionEnvironment} class.
+   */
+  static Configuration reflectivelyExtractFromEnv(StreamExecutionEnvironment env) {
     try {
       return (Configuration) getConfigurationMethod().invoke(env);
     } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfig.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfig.java
index 9c96bce..4982571 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfig.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfig.java
@@ -104,7 +104,7 @@
    * current environment set via the {@code flink-conf.yaml}.
    */
   public static StatefulFunctionsConfig fromEnvironment(StreamExecutionEnvironment env) {
-    Configuration configuration = ReflectiveFlinkConfigExtractor.extractFromEnv(env);
+    Configuration configuration = FlinkConfigExtractor.reflectivelyExtractFromEnv(env);
     return new StatefulFunctionsConfig(configuration);
   }
 
@@ -131,7 +131,7 @@
    *
    * @param configuration a configuration to read the values from
    */
-  public StatefulFunctionsConfig(Configuration configuration) {
+  private StatefulFunctionsConfig(Configuration configuration) {
     this.factoryType = configuration.get(USER_MESSAGE_SERIALIZER);
     this.flinkJobName = configuration.get(FLINK_JOB_NAME);
     this.feedbackBufferSize = configuration.get(TOTAL_MEMORY_USED_FOR_FEEDBACK_CHECKPOINTING);
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsJob.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsJob.java
index 803bde4..5f56786 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsJob.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsJob.java
@@ -36,7 +36,7 @@
 
     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
-    Configuration flinkConfig = ReflectiveFlinkConfigExtractor.extractFromEnv(env);
+    Configuration flinkConfig = FlinkConfigExtractor.reflectivelyExtractFromEnv(env);
     StatefulFunctionsConfigValidator.validate(flinkConfig);
 
     StatefulFunctionsConfig stateFunConfig =
diff --git a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigTest.java b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigTest.java
index 5a93dd7..baa704b 100644
--- a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigTest.java
+++ b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigTest.java
@@ -20,7 +20,6 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.configuration.MemorySize;
-import org.apache.flink.statefun.flink.core.exceptions.StatefulFunctionsInvalidConfigException;
 import org.apache.flink.statefun.flink.core.message.MessageFactoryType;
 import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
 import org.hamcrest.Matchers;
@@ -48,7 +47,8 @@
     configuration.setString("statefun.module.global-config.key1", "value1");
     configuration.setString("statefun.module.global-config.key2", "value2");
 
-    StatefulFunctionsConfig stateFunConfig = new StatefulFunctionsConfig(configuration);
+    StatefulFunctionsConfig stateFunConfig =
+        StatefulFunctionsConfig.fromFlinkConfiguration(configuration);
 
     Assert.assertEquals(stateFunConfig.getFlinkJobName(), testName);
     Assert.assertEquals(stateFunConfig.getFactoryType(), MessageFactoryType.WITH_KRYO_PAYLOADS);
@@ -60,12 +60,6 @@
         stateFunConfig.getGlobalConfigurations(), Matchers.hasEntry("key2", "value2"));
   }
 
-  @Test(expected = StatefulFunctionsInvalidConfigException.class)
-  public void invalidStrictFlinkConfigsThrows() {
-    Configuration configuration = new Configuration();
-    new StatefulFunctionsConfig(configuration);
-  }
-
   private static Configuration validConfiguration() {
     Configuration configuration = new Configuration();
     configuration.set(StatefulFunctionsConfig.FLINK_JOB_NAME, "name");
diff --git a/statefun-flink/statefun-flink-harness/src/main/java/org/apache/flink/statefun/flink/harness/Harness.java b/statefun-flink/statefun-flink-harness/src/main/java/org/apache/flink/statefun/flink/harness/Harness.java
index 6fe3856..9eb7def 100644
--- a/statefun-flink/statefun-flink-harness/src/main/java/org/apache/flink/statefun/flink/harness/Harness.java
+++ b/statefun-flink/statefun-flink-harness/src/main/java/org/apache/flink/statefun/flink/harness/Harness.java
@@ -120,7 +120,8 @@
     // untouched.
     env.configure(flinkConfig, Thread.currentThread().getContextClassLoader());
 
-    StatefulFunctionsConfig stateFunConfig = new StatefulFunctionsConfig(flinkConfig);
+    StatefulFunctionsConfig stateFunConfig =
+        StatefulFunctionsConfig.fromFlinkConfiguration(flinkConfig);
     stateFunConfig.addAllGlobalConfigurations(globalConfigurations);
     stateFunConfig.setProvider(new HarnessProvider(overrideIngress, overrideEgress));
     StatefulFunctionsJob.main(env, stateFunConfig);