[FLINK-38723][common/composer] Add getFlinkConf method to Context.

Author: lvyanquan

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();
+        }
     }
 }

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;
+        }
     }
 }

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(

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);
     }
 

flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerTest.java

@@ -17,17 +17,21 @@
 
 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.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.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.configuration.DeploymentOptions;
 
 import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
 
@@ -38,6 +42,8 @@
 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.assertThatThrownBy;
@@ -72,6 +78,63 @@ void testCreateDataSinkFromSinkDef() {
                 .isEqualTo("0.0.0.0");
     }
 
+    @Test
+    void testGettingFlinkConfiguration() {
+        FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
+        PipelineDef pipelineDef =
+                new PipelineDef(
+                        new SourceDef(ValuesDataFactory.IDENTIFIER, null, new Configuration()),
+                        new SinkDef(TestDataSinkFactory.IDENTIFIER, null, new Configuration()),
+                        Collections.emptyList(),
+                        Collections.emptyList(),
+                        Collections.emptyList(),
+                        new Configuration());
+        // No exception will be thrown.
+        composer.compose(pipelineDef);
+    }
+
+    /** A dummy {@link DataSinkFactory} for testing. */
+    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<>();
+        }
+    }
+
     @ParameterizedTest
     @MethodSource
     void testInvalidPipelineConfiguration(Configuration pipelineConfig) {

flink-cdc-composer/src/test/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory

@@ -16,4 +16,5 @@ 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