fix review

Author: ruanhang1993

flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java

@@ -78,7 +78,8 @@ 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_POST_TRANSFORM_CONVERTER_KEY = "post-transform-converter";
+    private static final String TRANSFORM_CONVERTER_AFTER_TRANSFORM_KEY =
+            "converter-after-transform";
 
     // UDF related keys
     private static final String UDF_KEY = "user-defined-function";
@@ -318,7 +319,7 @@ private TransformDef toTransformDef(JsonNode transformNode) {
                         .map(JsonNode::asText)
                         .orElse(null);
         String postTransformConverter =
-                Optional.ofNullable(transformNode.get(TRANSFORM_POST_TRANSFORM_CONVERTER_KEY))
+                Optional.ofNullable(transformNode.get(TRANSFORM_CONVERTER_AFTER_TRANSFORM_KEY))
                         .map(JsonNode::asText)
                         .orElse(null);
 

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"
-                        + "    post-transform-converter: SOFT_DELETE\n"
+                        + "    converter-after-transform: SOFT_DELETE\n"
                         + "  - source-table: mydb.web_order_.*\n"
                         + "    projection: CONCAT(id, order_id) as uniq_id, *\n"
                         + "    filter: uniq_id > 10\n"

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
-    post-transform-converter: SOFT_DELETE
+    converter-after-transform: SOFT_DELETE
   - source-table: mydb.web_order_.*
     projection: CONCAT(id, order_id) as uniq_id, *
     filter: uniq_id > 10

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
-    post-transform-converter: SOFT_DELETE
+    converter-after-transform: SOFT_DELETE
   - source-table: mydb.web_order_.*
     projection: CONCAT(id, order_id) as uniq_id, *
     filter: uniq_id > 10

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"
-                                + "    post-transform-converter: SOFT_DELETE\n"
+                                + "    converter-after-transform: SOFT_DELETE\n"
                                 + "\n"
                                 + "sink:\n"
                                 + "  type: values\n"

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java

@@ -31,9 +31,10 @@
 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.converter.PostTransformConverter;
+import org.apache.flink.cdc.runtime.operators.transform.converter.PostTransformConverters;
 import org.apache.flink.cdc.runtime.parser.TransformParser;
 import org.apache.flink.streaming.api.graph.StreamConfig;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
@@ -190,7 +191,7 @@ public void open() throws Exception {
                                             TransformProjection.of(projection).orElse(null),
                                             TransformFilter.of(filterExpression, udfDescriptors)
                                                     .orElse(null),
-                                            PostTransformConverter.of(
+                                            PostTransformConverters.of(
                                                     transformRule.getPostTransformConverter()));
                                 })
                         .collect(Collectors.toList());

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.common.transform.converter.PostTransformConverter;
+import org.apache.flink.cdc.runtime.operators.transform.converter.PostTransformConverter;
 
 import javax.annotation.Nullable;
 

…rm/converter/PostTransformConverter.java …rm/converter/PostTransformConverter.javaflink-cdc-common/src/main/java/org/apache/flink/cdc/common/transform/converter/PostTransformConverter.java renamed to flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/converter/PostTransformConverter.java flink-cdc-common/src/main/java/org/apache/flink/cdc/common/transform/converter/PostTransformConverter.java renamed to flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/converter/PostTransformConverter.java

@@ -15,11 +15,10 @@
  * limitations under the License.
  */
 
-package org.apache.flink.cdc.common.transform.converter;
+package org.apache.flink.cdc.runtime.operators.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 java.io.Serializable;
 import java.util.Optional;
@@ -31,32 +30,6 @@
 @Experimental
 public interface PostTransformConverter extends Serializable {
 
+    /** Change some parts of {@link DataChangeEvent}, like op or meta. */
     Optional<DataChangeEvent> convert(DataChangeEvent dataChangeEvent);
-
-    static Optional<PostTransformConverter> of(String classPath) {
-        if (StringUtils.isNullOrWhitespaceOnly(classPath)) {
-            return Optional.empty();
-        }
-
-        Optional<PostTransformConverter> postTransformConverter =
-                PostTransformConverters.getInternalPostTransformConverter(classPath);
-        if (postTransformConverter.isPresent()) {
-            return postTransformConverter;
-        }
-
-        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
-        try {
-            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, PostTransformConverter.class.getName()));
-        } catch (Exception e) {
-            throw new RuntimeException("Create post transform converter failed.", e);
-        }
-    }
 }

…m/converter/PostTransformConverters.java …m/converter/PostTransformConverters.javaflink-cdc-common/src/main/java/org/apache/flink/cdc/common/transform/converter/PostTransformConverters.java renamed to flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/converter/PostTransformConverters.java flink-cdc-common/src/main/java/org/apache/flink/cdc/common/transform/converter/PostTransformConverters.java renamed to flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/converter/PostTransformConverters.java

@@ -15,9 +15,10 @@
  * limitations under the License.
  */
 
-package org.apache.flink.cdc.common.transform.converter;
+package org.apache.flink.cdc.runtime.operators.transform.converter;
 
 import org.apache.flink.cdc.common.annotation.Experimental;
+import org.apache.flink.cdc.common.utils.StringUtils;
 
 import java.util.Optional;
 
@@ -26,13 +27,18 @@
 public class PostTransformConverters {
     public static final String SOFT_DELETE_CONVERTER = "SOFT_DELETE";
 
-    public static Optional<PostTransformConverter> getInternalPostTransformConverter(
-            String identifier) {
+    /** Get the {@link PostTransformConverter} by given identifier. */
+    public static Optional<PostTransformConverter> of(String identifier) {
+        if (StringUtils.isNullOrWhitespaceOnly(identifier)) {
+            return Optional.empty();
+        }
+
         switch (identifier) {
             case SOFT_DELETE_CONVERTER:
                 return Optional.of(new SoftDeleteConverter());
             default:
-                return Optional.empty();
+                throw new IllegalArgumentException(
+                        String.format("Failed to find the converter %s.", identifier));
         }
     }
 }

…sform/converter/SoftDeleteConverter.java …sform/converter/SoftDeleteConverter.javaflink-cdc-common/src/main/java/org/apache/flink/cdc/common/transform/converter/SoftDeleteConverter.java renamed to flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/converter/SoftDeleteConverter.java flink-cdc-common/src/main/java/org/apache/flink/cdc/common/transform/converter/SoftDeleteConverter.java renamed to flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/converter/SoftDeleteConverter.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.cdc.common.transform.converter;
+package org.apache.flink.cdc.runtime.operators.transform.converter;
 
 import org.apache.flink.cdc.common.event.DataChangeEvent;