[cdc-common][hotfix] Remove redundant prefix of pipeline options (apache#2821)

This closes apache#2821.

Author: PatrickRen

flink-cdc-common/src/main/java/com/ververica/cdc/common/pipeline/PipelineOptions.java

@@ -29,19 +29,19 @@
 public class PipelineOptions {
 
     public static final ConfigOption<String> PIPELINE_NAME =
-            ConfigOptions.key("pipeline.name")
+            ConfigOptions.key("name")
                     .stringType()
                     .defaultValue("Flink CDC Pipeline Job")
                     .withDescription("The name of the pipeline");
 
-    public static final ConfigOption<Integer> GLOBAL_PARALLELISM =
-            ConfigOptions.key("pipeline.global.parallelism")
+    public static final ConfigOption<Integer> PIPELINE_PARALLELISM =
+            ConfigOptions.key("parallelism")
                     .intType()
                     .noDefaultValue()
-                    .withDescription("The global parallelism of the pipeline");
+                    .withDescription("Parallelism of the pipeline");
 
-    public static final ConfigOption<SchemaChangeBehavior> SCHEMA_CHANGE_BEHAVIOR =
-            ConfigOptions.key("pipeline.schema.change.behavior")
+    public static final ConfigOption<SchemaChangeBehavior> PIPELINE_SCHEMA_CHANGE_BEHAVIOR =
+            ConfigOptions.key("schema.change.behavior")
                     .enumType(SchemaChangeBehavior.class)
                     .defaultValue(SchemaChangeBehavior.EVOLVE)
                     .withDescription(
@@ -58,7 +58,7 @@ public class PipelineOptions {
                                     .build());
 
     public static final ConfigOption<String> PIPELINE_LOCAL_TIME_ZONE =
-            ConfigOptions.key("pipeline.local-time-zone")
+            ConfigOptions.key("local-time-zone")
                     .stringType()
                     // "systemDefault" is a special value to decide whether to use
                     // ZoneId.systemDefault() in
@@ -77,8 +77,8 @@ public class PipelineOptions {
                                                     + "such as \"America/Los_Angeles\", or a custom timezone id such as \"GMT-08:00\".")
                                     .build());
 
-    public static final ConfigOption<String> SCHEMA_OPERATOR_UID =
-            ConfigOptions.key("pipeline.schema.operator.uid")
+    public static final ConfigOption<String> PIPELINE_SCHEMA_OPERATOR_UID =
+            ConfigOptions.key("schema.operator.uid")
                     .stringType()
                     .defaultValue("$$_schema_operator_$$")
                     .withDescription(

flink-cdc-composer/src/main/java/com/ververica/cdc/composer/flink/FlinkPipelineComposer.java

@@ -92,7 +92,7 @@ private FlinkPipelineComposer(StreamExecutionEnvironment env, boolean isBlocking
 
     @Override
     public PipelineExecution compose(PipelineDef pipelineDef) {
-        int parallelism = pipelineDef.getConfig().get(PipelineOptions.GLOBAL_PARALLELISM);
+        int parallelism = pipelineDef.getConfig().get(PipelineOptions.PIPELINE_PARALLELISM);
         env.getConfig().setParallelism(parallelism);
 
         // Source
@@ -110,8 +110,10 @@ public PipelineExecution compose(PipelineDef pipelineDef) {
         // Schema operator
         SchemaOperatorTranslator schemaOperatorTranslator =
                 new SchemaOperatorTranslator(
-                        pipelineDef.getConfig().get(PipelineOptions.SCHEMA_CHANGE_BEHAVIOR),
-                        pipelineDef.getConfig().get(PipelineOptions.SCHEMA_OPERATOR_UID));
+                        pipelineDef
+                                .getConfig()
+                                .get(PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR),
+                        pipelineDef.getConfig().get(PipelineOptions.PIPELINE_SCHEMA_OPERATOR_UID));
         stream =
                 schemaOperatorTranslator.translate(
                         stream, parallelism, dataSink.getMetadataApplier());

flink-cdc-composer/src/main/java/com/ververica/cdc/composer/flink/translator/DataSourceTranslator.java

@@ -62,7 +62,7 @@ public DataStreamSource<Event> translate(
                 .ifPresent(jar -> FlinkEnvironmentUtils.addJar(env, jar));
 
         // Get source provider
-        final int sourceParallelism = pipelineConfig.get(PipelineOptions.GLOBAL_PARALLELISM);
+        final int sourceParallelism = pipelineConfig.get(PipelineOptions.PIPELINE_PARALLELISM);
         EventSourceProvider eventSourceProvider = dataSource.getEventSourceProvider();
         if (eventSourceProvider instanceof FlinkSourceProvider) {
             // Source

flink-cdc-composer/src/test/java/com/ververica/cdc/composer/flink/FlinkPipelineComposerITCase.java

@@ -109,7 +109,7 @@ void testSingleSplitSingleTable() throws Exception {
 
         // Setup pipeline
         Configuration pipelineConfig = new Configuration();
-        pipelineConfig.set(PipelineOptions.GLOBAL_PARALLELISM, 1);
+        pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
         PipelineDef pipelineDef =
                 new PipelineDef(
                         sourceDef,
@@ -163,7 +163,7 @@ void testSingleSplitMultipleTables() throws Exception {
 
         // Setup pipeline
         Configuration pipelineConfig = new Configuration();
-        pipelineConfig.set(PipelineOptions.GLOBAL_PARALLELISM, 1);
+        pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
         PipelineDef pipelineDef =
                 new PipelineDef(
                         sourceDef,
@@ -227,7 +227,7 @@ void testMultiSplitsSingleTable() throws Exception {
 
         // Setup pipeline
         Configuration pipelineConfig = new Configuration();
-        pipelineConfig.set(PipelineOptions.GLOBAL_PARALLELISM, MAX_PARALLELISM);
+        pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, MAX_PARALLELISM);
         PipelineDef pipelineDef =
                 new PipelineDef(
                         sourceDef,