fix review

ruanhang1993 · ruanhang1993 · commit e939f3bbc85a · 2024-12-17T17:19:09.000+08:00
diff --git a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java
@@ -78,8 +78,7 @@ public class YamlPipelineDefinitionParser implements PipelineDefinitionParser {
     private static final String TRANSFORM_PROJECTION_KEY = "projection";
     private static final String TRANSFORM_FILTER_KEY = "filter";
     private static final String TRANSFORM_DESCRIPTION_KEY = "description";
-    private static final String TRANSFORM_CONVERTOR_AFTER_TRANSFORM_KEY =
-            "convertor-after-transform";
+    private static final String TRANSFORM_POST_TRANSFORM_CONVERTER_KEY = "post-transform-converter";
 
     // UDF related keys
     private static final String UDF_KEY = "user-defined-function";
@@ -318,8 +317,8 @@ private TransformDef toTransformDef(JsonNode transformNode) {
                 Optional.ofNullable(transformNode.get(TRANSFORM_DESCRIPTION_KEY))
                         .map(JsonNode::asText)
                         .orElse(null);
-        String convertorAfterTransform =
-                Optional.ofNullable(transformNode.get(TRANSFORM_CONVERTOR_AFTER_TRANSFORM_KEY))
+        String postTransformConverter =
+                Optional.ofNullable(transformNode.get(TRANSFORM_POST_TRANSFORM_CONVERTER_KEY))
                         .map(JsonNode::asText)
                         .orElse(null);
 
@@ -331,7 +330,7 @@ private TransformDef toTransformDef(JsonNode transformNode) {
                 partitionKeys,
                 tableOptions,
                 description,
-                convertorAfterTransform);
+                postTransformConverter);
     }
 
     private Configuration toPipelineConfig(JsonNode pipelineConfigNode) {
diff --git a/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java b/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java
@@ -404,7 +404,7 @@ void testParsingFullDefinitionFromString() throws Exception {
                         + "    partition-keys: product_name\n"
                         + "    table-options: comment=app order\n"
                         + "    description: project fields from source table\n"
-                        + "    convertor-after-transform: SOFT_DELETE\n"
+                        + "    post-transform-converter: SOFT_DELETE\n"
                         + "  - source-table: mydb.web_order_.*\n"
                         + "    projection: CONCAT(id, order_id) as uniq_id, *\n"
                         + "    filter: uniq_id > 10\n"
diff --git a/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full-with-repsym.yaml b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full-with-repsym.yaml
@@ -49,7 +49,7 @@ transform:
     partition-keys: product_name
     table-options: comment=app order
     description: project fields from source table
-    convertor-after-transform: SOFT_DELETE
+    converter-after-transform: SOFT_DELETE
   - source-table: mydb.web_order_.*
     projection: CONCAT(id, order_id) as uniq_id, *
     filter: uniq_id > 10
diff --git a/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml b/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml
@@ -47,7 +47,7 @@ transform:
     partition-keys: product_name
     table-options: comment=app order
     description: project fields from source table
-    convertor-after-transform: SOFT_DELETE
+    converter-after-transform: SOFT_DELETE
   - source-table: mydb.web_order_.*
     projection: CONCAT(id, order_id) as uniq_id, *
     filter: uniq_id > 10
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/transform/converter/PostTransformConverter.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/transform/converter/PostTransformConverter.java
@@ -15,47 +15,48 @@
  * limitations under the License.
  */
 
-package org.apache.flink.cdc.runtime.operators.transform.convertor;
+package org.apache.flink.cdc.common.transform.converter;
 
+import org.apache.flink.cdc.common.annotation.Experimental;
 import org.apache.flink.cdc.common.event.DataChangeEvent;
 import org.apache.flink.cdc.common.utils.StringUtils;
-import org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator;
-import org.apache.flink.cdc.runtime.operators.transform.TransformRule;
 
 import java.io.Serializable;
 import java.util.Optional;
 
 /**
- * The TransformConvertor applies to convert the {@link DataChangeEvent} after other part of {@link
- * TransformRule} in {@link PostTransformOperator}.
+ * The PostTransformConverter applies to convert the {@link DataChangeEvent} after other part of
+ * TransformRule in PostTransformOperator.
  */
-public interface TransformConvertor extends Serializable {
-    String SOFT_DELETE_CONVERTOR = "SOFT_DELETE";
+@Experimental
+public interface PostTransformConverter extends Serializable {
 
     Optional<DataChangeEvent> convert(DataChangeEvent dataChangeEvent);
 
-    static Optional<TransformConvertor> of(String classPath) {
+    static Optional<PostTransformConverter> of(String classPath) {
         if (StringUtils.isNullOrWhitespaceOnly(classPath)) {
             return Optional.empty();
         }
 
-        if (SOFT_DELETE_CONVERTOR.equals(classPath)) {
-            return Optional.of(new SoftDeleteConvertor());
+        Optional<PostTransformConverter> postTransformConverter =
+                PostTransformConverters.getInternalPostTransformConverter(classPath);
+        if (postTransformConverter.isPresent()) {
+            return postTransformConverter;
         }
 
         ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
         try {
-            Class<?> convertorClass = classLoader.loadClass(classPath);
-            Object convertor = convertorClass.newInstance();
-            if (convertor instanceof TransformConvertor) {
-                return Optional.of((TransformConvertor) convertor);
+            Class<?> converterClass = classLoader.loadClass(classPath);
+            Object converter = converterClass.newInstance();
+            if (converter instanceof PostTransformConverter) {
+                return Optional.of((PostTransformConverter) converter);
             }
             throw new IllegalArgumentException(
                     String.format(
                             "%s is not an instance of %s",
-                            classPath, TransformConvertor.class.getName()));
+                            classPath, PostTransformConverter.class.getName()));
         } catch (Exception e) {
-            throw new RuntimeException("Create transform convertor failed.", e);
+            throw new RuntimeException("Create post transform converter failed.", e);
         }
     }
 }
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/transform/converter/PostTransformConverters.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/transform/converter/PostTransformConverters.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.transform.converter;
+
+import org.apache.flink.cdc.common.annotation.Experimental;
+
+import java.util.Optional;
+
+/** The {@link PostTransformConverter} utils. */
+@Experimental
+public class PostTransformConverters {
+    public static final String SOFT_DELETE_CONVERTER = "SOFT_DELETE";
+
+    public static Optional<PostTransformConverter> getInternalPostTransformConverter(
+            String identifier) {
+        switch (identifier) {
+            case SOFT_DELETE_CONVERTER:
+                return Optional.of(new SoftDeleteConverter());
+            default:
+                return Optional.empty();
+        }
+    }
+}
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/transform/converter/SoftDeleteConverter.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/transform/converter/SoftDeleteConverter.java
@@ -15,16 +15,16 @@
  * limitations under the License.
  */
 
-package org.apache.flink.cdc.runtime.operators.transform.convertor;
+package org.apache.flink.cdc.common.transform.converter;
 
 import org.apache.flink.cdc.common.event.DataChangeEvent;
 
 import java.util.Optional;
 
 import static org.apache.flink.cdc.common.event.OperationType.DELETE;
 
-/** This {@link TransformConvertor} convert delete events to insert events. */
-public class SoftDeleteConvertor implements TransformConvertor {
+/** This {@link PostTransformConverter} convert delete events to insert events. */
+public class SoftDeleteConverter implements PostTransformConverter {
 
     @Override
     public Optional<DataChangeEvent> convert(DataChangeEvent dataChangeEvent) {
diff --git a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/TransformDef.java b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/TransformDef.java
@@ -50,7 +50,7 @@ public class TransformDef {
     private final String primaryKeys;
     private final String partitionKeys;
     private final String tableOptions;
-    private final String convertorAfterTransform;
+    private final String postTransformConverter;
 
     public TransformDef(
             String sourceTable,
@@ -60,15 +60,15 @@ public TransformDef(
             String partitionKeys,
             String tableOptions,
             String description,
-            String convertorAfterTransform) {
+            String postTransformConverter) {
         this.sourceTable = sourceTable;
         this.projection = projection;
         this.filter = filter;
         this.primaryKeys = primaryKeys;
         this.partitionKeys = partitionKeys;
         this.tableOptions = tableOptions;
         this.description = description;
-        this.convertorAfterTransform = convertorAfterTransform;
+        this.postTransformConverter = postTransformConverter;
     }
 
     public String getSourceTable() {
@@ -107,8 +107,8 @@ public String getTableOptions() {
         return tableOptions;
     }
 
-    public String getConvertorAfterTransform() {
-        return convertorAfterTransform;
+    public String getPostTransformConverter() {
+        return postTransformConverter;
     }
 
     @Override
@@ -126,8 +126,8 @@ public String toString() {
                 + ", description='"
                 + description
                 + '\''
-                + ", convertorAfterTransform='"
-                + convertorAfterTransform
+                + ", postTransformConverter='"
+                + postTransformConverter
                 + '\''
                 + '}';
     }
@@ -148,7 +148,7 @@ public boolean equals(Object o) {
                 && Objects.equals(primaryKeys, that.primaryKeys)
                 && Objects.equals(partitionKeys, that.partitionKeys)
                 && Objects.equals(tableOptions, that.tableOptions)
-                && Objects.equals(convertorAfterTransform, that.convertorAfterTransform);
+                && Objects.equals(postTransformConverter, that.postTransformConverter);
     }
 
     @Override
@@ -161,6 +161,6 @@ public int hashCode() {
                 primaryKeys,
                 partitionKeys,
                 tableOptions,
-                convertorAfterTransform);
+                postTransformConverter);
     }
 }
diff --git a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
@@ -61,7 +61,7 @@ public DataStream<Event> translatePreTransform(
                     transform.getPrimaryKeys(),
                     transform.getPartitionKeys(),
                     transform.getTableOptions(),
-                    transform.getConvertorAfterTransform());
+                    transform.getPostTransformConverter());
         }
 
         preTransformFunctionBuilder.addUdfFunctions(
@@ -93,7 +93,7 @@ public DataStream<Event> translatePostTransform(
                         transform.getPrimaryKeys(),
                         transform.getPartitionKeys(),
                         transform.getTableOptions(),
-                        transform.getConvertorAfterTransform());
+                        transform.getPostTransformConverter());
             }
         }
         postTransformFunctionBuilder.addTimezone(timezone);
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eITCase.java
@@ -351,7 +351,7 @@ public void testSoftDelete() throws Exception {
                                 + "transform:\n"
                                 + "  - source-table: \\.*.\\.*\n"
                                 + "    projection: \\*, __data_event_type__ AS op_type\n"
-                                + "    convertor-after-transform: SOFT_DELETE\n"
+                                + "    post-transform-converter: SOFT_DELETE\n"
                                 + "\n"
                                 + "sink:\n"
                                 + "  type: values\n"
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java
@@ -31,9 +31,9 @@
 import org.apache.flink.cdc.common.pipeline.PipelineOptions;
 import org.apache.flink.cdc.common.schema.Schema;
 import org.apache.flink.cdc.common.schema.Selectors;
+import org.apache.flink.cdc.common.transform.converter.PostTransformConverter;
 import org.apache.flink.cdc.common.udf.UserDefinedFunctionContext;
 import org.apache.flink.cdc.common.utils.SchemaUtils;
-import org.apache.flink.cdc.runtime.operators.transform.convertor.TransformConvertor;
 import org.apache.flink.cdc.runtime.parser.TransformParser;
 import org.apache.flink.streaming.api.graph.StreamConfig;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
@@ -103,7 +103,7 @@ public PostTransformOperator.Builder addTransform(
                 String primaryKey,
                 String partitionKey,
                 String tableOptions,
-                String convertorAfterTransform) {
+                String postTransformConverter) {
             transformRules.add(
                     new TransformRule(
                             tableInclusions,
@@ -112,7 +112,7 @@ public PostTransformOperator.Builder addTransform(
                             primaryKey,
                             partitionKey,
                             tableOptions,
-                            convertorAfterTransform));
+                            postTransformConverter));
             return this;
         }
 
@@ -190,8 +190,8 @@ public void open() throws Exception {
                                             TransformProjection.of(projection).orElse(null),
                                             TransformFilter.of(filterExpression, udfDescriptors)
                                                     .orElse(null),
-                                            TransformConvertor.of(
-                                                    transformRule.getConvertorAfterTransform()));
+                                            PostTransformConverter.of(
+                                                    transformRule.getPostTransformConverter()));
                                 })
                         .collect(Collectors.toList());
         this.transformProjectionProcessorMap = new ConcurrentHashMap<>();
@@ -438,11 +438,11 @@ private Optional<DataChangeEvent> processDataChangeEvent(DataChangeEvent dataCha
                                     epochTime);
                 }
                 if (dataChangeEventOptional.isPresent()
-                        && transform.getConvertorAfterTransform().isPresent()) {
+                        && transform.getPostTransformConverter().isPresent()) {
                     dataChangeEventOptional =
                             convertDataChangeEvent(
                                     dataChangeEventOptional.get(),
-                                    transform.getConvertorAfterTransform().get());
+                                    transform.getPostTransformConverter().get());
                 }
                 transformedDataChangeEventOptionalList.add(dataChangeEventOptional);
             }
@@ -462,8 +462,8 @@ private Optional<DataChangeEvent> processDataChangeEvent(DataChangeEvent dataCha
     }
 
     private Optional<DataChangeEvent> convertDataChangeEvent(
-            DataChangeEvent dataChangeEvent, TransformConvertor convertor) {
-        return convertor.convert(dataChangeEvent);
+            DataChangeEvent dataChangeEvent, PostTransformConverter postTransformConverter) {
+        return postTransformConverter.convert(dataChangeEvent);
     }
 
     private Optional<DataChangeEvent> processFilter(
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformer.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformer.java
@@ -18,7 +18,7 @@
 package org.apache.flink.cdc.runtime.operators.transform;
 
 import org.apache.flink.cdc.common.schema.Selectors;
-import org.apache.flink.cdc.runtime.operators.transform.convertor.TransformConvertor;
+import org.apache.flink.cdc.common.transform.converter.PostTransformConverter;
 
 import javax.annotation.Nullable;
 
@@ -30,17 +30,17 @@ public class PostTransformer {
 
     private final Optional<TransformProjection> projection;
     private final Optional<TransformFilter> filter;
-    private final Optional<TransformConvertor> convertorAfterTransform;
+    private final Optional<PostTransformConverter> postTransformConverter;
 
     public PostTransformer(
             Selectors selectors,
             @Nullable TransformProjection projection,
             @Nullable TransformFilter filter,
-            Optional<TransformConvertor> convertorAfterTransform) {
+            Optional<PostTransformConverter> postTransformConverter) {
         this.selectors = selectors;
         this.projection = projection != null ? Optional.of(projection) : Optional.empty();
         this.filter = filter != null ? Optional.of(filter) : Optional.empty();
-        this.convertorAfterTransform = convertorAfterTransform;
+        this.postTransformConverter = postTransformConverter;
     }
 
     public Selectors getSelectors() {
@@ -55,7 +55,7 @@ public Optional<TransformFilter> getFilter() {
         return filter;
     }
 
-    public Optional<TransformConvertor> getConvertorAfterTransform() {
-        return convertorAfterTransform;
+    public Optional<PostTransformConverter> getPostTransformConverter() {
+        return postTransformConverter;
     }
 }
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PreTransformOperator.java
@@ -100,7 +100,7 @@ public PreTransformOperator.Builder addTransform(
                 String primaryKey,
                 String partitionKey,
                 String tableOption,
-                @Nullable String convertorAfterTransform) {
+                @Nullable String postTransformConverter) {
             transformRules.add(
                     new TransformRule(
                             tableInclusions,
@@ -109,7 +109,7 @@ public PreTransformOperator.Builder addTransform(
                             primaryKey,
                             partitionKey,
                             tableOption,
-                            convertorAfterTransform));
+                            postTransformConverter));
             return this;
         }
 
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformRule.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformRule.java

Original file line number	Diff line number	Diff line change
`@@ -61,7 +61,7 @@ public DataStream<Event> translatePreTransform(`
`61`	`61`	`transform.getPrimaryKeys(),`
`62`	`62`	`transform.getPartitionKeys(),`
`63`	`63`	`transform.getTableOptions(),`
`64`		`- transform.getConvertorAfterTransform());`
	`64`	`+ transform.getPostTransformConverter());`
`65`	`65`	`}`
`66`	`66`
`67`	`67`	`preTransformFunctionBuilder.addUdfFunctions(`
`@@ -93,7 +93,7 @@ public DataStream<Event> translatePostTransform(`
`93`	`93`	`transform.getPrimaryKeys(),`
`94`	`94`	`transform.getPartitionKeys(),`
`95`	`95`	`transform.getTableOptions(),`
`96`		`- transform.getConvertorAfterTransform());`
	`96`	`+ transform.getPostTransformConverter());`
`97`	`97`	`}`
`98`	`98`	`}`
`99`	`99`	`postTransformFunctionBuilder.addTimezone(timezone);`