[FLINK-38723][common/composer] Add getFlinkConf method to Context. (apache#4179)

lvyanquan · web-flow · commit 7ad789edf9aa · 2025-11-26T16:04:39.000+08:00
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/factories/Factory.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/factories/Factory.java
@@ -20,6 +20,7 @@
 import org.apache.flink.cdc.common.annotation.PublicEvolving;
 import org.apache.flink.cdc.common.configuration.ConfigOption;
 import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
 
 import java.util.Set;
 
@@ -78,5 +79,10 @@ interface Context {
          * <p>The class loader is in particular useful for discovering factories.
          */
         ClassLoader getClassLoader();
+
+        /** Returns the flink configuration of the current session. */
+        default ReadableConfig getFlinkConf() {
+            return new org.apache.flink.configuration.Configuration();
+        }
     }
 }
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/factories/FactoryHelper.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/factories/FactoryHelper.java
@@ -168,14 +168,28 @@ public static class DefaultContext implements Factory.Context {
         private final Configuration factoryConfiguration;
         private final ClassLoader classLoader;
         private final Configuration pipelineConfiguration;
+        private final ReadableConfig flinkConf;
 
         public DefaultContext(
                 Configuration factoryConfiguration,
                 Configuration pipelineConfiguration,
                 ClassLoader classLoader) {
+            this(
+                    factoryConfiguration,
+                    pipelineConfiguration,
+                    classLoader,
+                    new org.apache.flink.configuration.Configuration());
+        }
+
+        public DefaultContext(
+                Configuration factoryConfiguration,
+                Configuration pipelineConfiguration,
+                ClassLoader classLoader,
+                ReadableConfig flinkConf) {
             this.factoryConfiguration = factoryConfiguration;
             this.pipelineConfiguration = pipelineConfiguration;
             this.classLoader = classLoader;
+            this.flinkConf = flinkConf;
         }
 
         @Override
@@ -192,5 +206,10 @@ public Configuration getPipelineConfiguration() {
         public ClassLoader getClassLoader() {
             return classLoader;
         }
+
+        @Override
+        public ReadableConfig getFlinkConf() {
+            return flinkConf;
+        }
     }
 }
diff --git a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSinkTranslator.java b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSinkTranslator.java
@@ -76,7 +76,8 @@ public DataSink createDataSink(
                 new FactoryHelper.DefaultContext(
                         sinkDef.getConfig(),
                         pipelineConfig,
-                        Thread.currentThread().getContextClassLoader()));
+                        Thread.currentThread().getContextClassLoader(),
+                        env.getConfiguration()));
     }
 
     public void translate(
diff --git a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSourceTranslator.java b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/DataSourceTranslator.java
@@ -97,7 +97,8 @@ public DataSource createDataSource(
                 new FactoryHelper.DefaultContext(
                         sourceDef.getConfig(),
                         pipelineConfig,
-                        Thread.currentThread().getContextClassLoader());
+                        Thread.currentThread().getContextClassLoader(),
+                        env.getConfiguration());
         return sourceFactory.createDataSource(context);
     }
 
diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerTest.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerTest.java
@@ -17,17 +17,25 @@
 
 package org.apache.flink.cdc.composer.flink;
 
+import org.apache.flink.cdc.common.configuration.ConfigOption;
 import org.apache.flink.cdc.common.configuration.Configuration;
 import org.apache.flink.cdc.common.factories.DataSinkFactory;
+import org.apache.flink.cdc.common.factories.DataSourceFactory;
 import org.apache.flink.cdc.common.factories.FactoryHelper;
 import org.apache.flink.cdc.common.pipeline.PipelineOptions;
 import org.apache.flink.cdc.common.sink.DataSink;
+import org.apache.flink.cdc.common.sink.EventSinkProvider;
+import org.apache.flink.cdc.common.sink.MetadataApplier;
+import org.apache.flink.cdc.common.source.DataSource;
 import org.apache.flink.cdc.composer.definition.PipelineDef;
 import org.apache.flink.cdc.composer.definition.SinkDef;
 import org.apache.flink.cdc.composer.definition.SourceDef;
 import org.apache.flink.cdc.composer.utils.FactoryDiscoveryUtils;
 import org.apache.flink.cdc.composer.utils.factory.DataSinkFactory1;
 import org.apache.flink.cdc.connectors.values.factory.ValuesDataFactory;
+import org.apache.flink.cdc.connectors.values.source.ValuesDataSource;
+import org.apache.flink.cdc.connectors.values.source.ValuesDataSourceHelper;
+import org.apache.flink.configuration.DeploymentOptions;
 
 import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
 
@@ -38,8 +46,11 @@
 import org.junit.jupiter.params.provider.MethodSource;
 
 import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.stream.Stream;
 
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** A test for the {@link FlinkPipelineComposer}. */
@@ -72,6 +83,96 @@ void testCreateDataSinkFromSinkDef() {
                 .isEqualTo("0.0.0.0");
     }
 
+    @Test
+    void testGettingFlinkConfiguration() {
+        FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+        PipelineDef pipelineDef =
+                new PipelineDef(
+                        new SourceDef(TestDataSourceFactory.IDENTIFIER, null, new Configuration()),
+                        new SinkDef(TestDataSinkFactory.IDENTIFIER, null, new Configuration()),
+                        Collections.emptyList(),
+                        Collections.emptyList(),
+                        Collections.emptyList(),
+                        new Configuration());
+
+        assertThatCode(() -> composer.compose(pipelineDef)).doesNotThrowAnyException();
+    }
+
+    /** A dummy {@link DataSinkFactory} that validates the execution target. */
+    public static class TestDataSinkFactory implements DataSinkFactory {
+
+        public static final String IDENTIFIER = "test-sink-factory";
+
+        @Override
+        public DataSink createDataSink(Context context) {
+            // This option has no default value.
+            String target = context.getFlinkConf().get(DeploymentOptions.TARGET);
+            if (!"local".equals(target)) {
+                throw new IllegalArgumentException(
+                        "The flink configuration is invalid. Please check the pipeline configuration.");
+            }
+            return new DataSink() {
+                @Override
+                public EventSinkProvider getEventSinkProvider() {
+                    return null;
+                }
+
+                @Override
+                public MetadataApplier getMetadataApplier() {
+                    return schemaChangeEvent -> {};
+                }
+            };
+        }
+
+        @Override
+        public String identifier() {
+            return IDENTIFIER;
+        }
+
+        @Override
+        public Set<ConfigOption<?>> requiredOptions() {
+            return new HashSet<>();
+        }
+
+        @Override
+        public Set<ConfigOption<?>> optionalOptions() {
+            return new HashSet<>();
+        }
+    }
+
+    /** A dummy {@link DataSourceFactory} that validates the execution target. */
+    public static class TestDataSourceFactory implements DataSourceFactory {
+
+        public static final String IDENTIFIER = "test-source-factory";
+
+        @Override
+        public DataSource createDataSource(Context context) {
+            // This option has no default value.
+            String target = context.getFlinkConf().get(DeploymentOptions.TARGET);
+            if (!"local".equals(target)) {
+                throw new IllegalArgumentException(
+                        "The flink configuration is invalid. Please check the pipeline configuration.");
+            }
+            return new ValuesDataSource(
+                    ValuesDataSourceHelper.EventSetId.SINGLE_SPLIT_SINGLE_TABLE, Integer.MAX_VALUE);
+        }
+
+        @Override
+        public String identifier() {
+            return IDENTIFIER;
+        }
+
+        @Override
+        public Set<ConfigOption<?>> requiredOptions() {
+            return new HashSet<>();
+        }
+
+        @Override
+        public Set<ConfigOption<?>> optionalOptions() {
+            return new HashSet<>();
+        }
+    }
+
     @ParameterizedTest
     @MethodSource
     void testInvalidPipelineConfiguration(Configuration pipelineConfig) {
diff --git a/flink-cdc-composer/src/test/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory b/flink-cdc-composer/src/test/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
@@ -16,4 +16,6 @@ org.apache.flink.cdc.composer.utils.factory.DataSinkFactory1
 org.apache.flink.cdc.composer.utils.factory.DataSinkFactory2
 org.apache.flink.cdc.composer.utils.factory.DataSourceFactory1
 org.apache.flink.cdc.composer.utils.factory.DataSourceFactory2
-org.apache.flink.cdc.composer.testsource.factory.DistributedDataSourceFactory
+org.apache.flink.cdc.composer.testsource.factory.DistributedDataSourceFactory
+org.apache.flink.cdc.composer.flink.FlinkPipelineComposerTest$TestDataSinkFactory
+org.apache.flink.cdc.composer.flink.FlinkPipelineComposerTest$TestDataSourceFactory

Original file line number	Diff line number	Diff line change
`@@ -76,7 +76,8 @@ public DataSink createDataSink(`
`76`	`76`	`new FactoryHelper.DefaultContext(`
`77`	`77`	`sinkDef.getConfig(),`
`78`	`78`	`pipelineConfig,`
`79`		`- Thread.currentThread().getContextClassLoader()));`
	`79`	`+ Thread.currentThread().getContextClassLoader(),`
	`80`	`+ env.getConfiguration()));`
`80`	`81`	`}`
`81`	`82`
`82`	`83`	`public void translate(`
Original file line number	Diff line number	Diff line change
`@@ -97,7 +97,8 @@ public DataSource createDataSource(`
`97`	`97`	`new FactoryHelper.DefaultContext(`
`98`	`98`	`sourceDef.getConfig(),`
`99`	`99`	`pipelineConfig,`
`100`		`- Thread.currentThread().getContextClassLoader());`
	`100`	`+ Thread.currentThread().getContextClassLoader(),`
	`101`	`+ env.getConfiguration());`
`101`	`102`	`return sourceFactory.createDataSource(context);`
`102`	`103`	`}`
`103`	`104`