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

Author: lvyanquan

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

@@ -20,17 +20,21 @@
 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;
@@ -83,7 +87,7 @@ void testGettingFlinkConfiguration() {
         FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
         PipelineDef pipelineDef =
                 new PipelineDef(
-                        new SourceDef(ValuesDataFactory.IDENTIFIER, null, new Configuration()),
+                        new SourceDef(TestDataSourceFactory.IDENTIFIER, null, new Configuration()),
                         new SinkDef(TestDataSinkFactory.IDENTIFIER, null, new Configuration()),
                         Collections.emptyList(),
                         Collections.emptyList(),
@@ -93,7 +97,7 @@ void testGettingFlinkConfiguration() {
         composer.compose(pipelineDef);
     }
 
-    /** A dummy {@link DataSinkFactory} for testing. */
+    /** A dummy {@link DataSinkFactory} that validates the execution target. */
     public static class TestDataSinkFactory implements DataSinkFactory {
 
         public static final String IDENTIFIER = "test-sink-factory";
@@ -135,6 +139,39 @@ public Set<ConfigOption<?>> optionalOptions() {
         }
     }
 
+    /** 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) {

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

@@ -17,4 +17,5 @@ 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.flink.FlinkPipelineComposerTest$TestDataSinkFactory
+org.apache.flink.cdc.composer.flink.FlinkPipelineComposerTest$TestDataSinkFactory
+org.apache.flink.cdc.composer.flink.FlinkPipelineComposerTest$TestDataSourceFactory