[FLINK-35242] Supports per-SE type configuration & "lenient" evolution behavior

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

@@ -18,6 +18,7 @@
 package org.apache.flink.cdc.cli.parser;
 
 import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.event.SchemaChangeEventType;
 import org.apache.flink.cdc.common.utils.StringUtils;
 import org.apache.flink.cdc.composer.definition.PipelineDef;
 import org.apache.flink.cdc.composer.definition.RouteDef;
@@ -28,14 +29,17 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.type.TypeReference;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
 
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 
+import static org.apache.flink.cdc.common.utils.ChangeEventUtils.resolveSchemaEvolutionOptions;
 import static org.apache.flink.cdc.common.utils.Preconditions.checkNotNull;
 
 /** Parser for converting YAML formatted pipeline definition to {@link PipelineDef}. */
@@ -51,6 +55,8 @@ public class YamlPipelineDefinitionParser implements PipelineDefinitionParser {
     // Source / sink keys
     private static final String TYPE_KEY = "type";
     private static final String NAME_KEY = "name";
+    private static final String INCLUDE_SCHEMA_EVOLUTION_TYPES = "include.schema.changes";
+    private static final String EXCLUDE_SCHEMA_EVOLUTION_TYPES = "exclude.schema.changes";
 
     // Route keys
     private static final String ROUTE_SOURCE_TABLE_KEY = "source-table";
@@ -136,6 +142,23 @@ private SourceDef toSourceDef(JsonNode sourceNode) {
     }
 
     private SinkDef toSinkDef(JsonNode sinkNode) {
+        List<String> includedSETypes = new ArrayList<>();
+        List<String> excludedSETypes = new ArrayList<>();
+
+        Optional.ofNullable(sinkNode.get(INCLUDE_SCHEMA_EVOLUTION_TYPES))
+                .ifPresent(e -> e.forEach(tag -> includedSETypes.add(tag.asText())));
+
+        Optional.ofNullable(sinkNode.get(EXCLUDE_SCHEMA_EVOLUTION_TYPES))
+                .ifPresent(e -> e.forEach(tag -> excludedSETypes.add(tag.asText())));
+
+        Set<SchemaChangeEventType> declaredSETypes =
+                resolveSchemaEvolutionOptions(includedSETypes, excludedSETypes);
+
+        if (sinkNode instanceof ObjectNode) {
+            ((ObjectNode) sinkNode).remove(INCLUDE_SCHEMA_EVOLUTION_TYPES);
+            ((ObjectNode) sinkNode).remove(EXCLUDE_SCHEMA_EVOLUTION_TYPES);
+        }
+
         Map<String, String> sinkMap =
                 mapper.convertValue(sinkNode, new TypeReference<Map<String, String>>() {});
 
@@ -149,7 +172,7 @@ private SinkDef toSinkDef(JsonNode sinkNode) {
         // "name" field is optional
         String name = sinkMap.remove(NAME_KEY);
 
-        return new SinkDef(type, name, Configuration.fromMap(sinkMap));
+        return new SinkDef(type, name, Configuration.fromMap(sinkMap), declaredSETypes);
     }
 
     private RouteDef toRouteDef(JsonNode routeNode) {

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/AddColumnEvent.java

@@ -173,4 +173,9 @@ public String toString() {
     public TableId tableId() {
         return tableId;
     }
+
+    @Override
+    public SchemaChangeEventType getType() {
+        return SchemaChangeEventType.ADD_COLUMN;
+    }
 }

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/AlterColumnTypeEvent.java

@@ -77,4 +77,9 @@ public String toString() {
     public TableId tableId() {
         return tableId;
     }
+
+    @Override
+    public SchemaChangeEventType getType() {
+        return SchemaChangeEventType.ALTER_COLUMN_TYPE;
+    }
 }

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/CreateTableEvent.java

@@ -72,4 +72,9 @@ public String toString() {
     public TableId tableId() {
         return tableId;
     }
+
+    @Override
+    public SchemaChangeEventType getType() {
+        return SchemaChangeEventType.CREATE_TABLE;
+    }
 }

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/DropColumnEvent.java

@@ -76,4 +76,9 @@ public String toString() {
     public TableId tableId() {
         return tableId;
     }
+
+    @Override
+    public SchemaChangeEventType getType() {
+        return SchemaChangeEventType.DROP_COLUMN;
+    }
 }

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/RenameColumnEvent.java

@@ -70,4 +70,9 @@ public String toString() {
     public TableId tableId() {
         return tableId;
     }
+
+    @Override
+    public SchemaChangeEventType getType() {
+        return SchemaChangeEventType.RENAME_COLUMN;
+    }
 }

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/SchemaChangeEvent.java

@@ -26,4 +26,7 @@
  * system, such as CREATE, DROP, RENAME and so on.
  */
 @PublicEvolving
-public interface SchemaChangeEvent extends ChangeEvent, Serializable {}
+public interface SchemaChangeEvent extends ChangeEvent, Serializable {
+    /** Returns its {@link SchemaChangeEventType}. */
+    SchemaChangeEventType getType();
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/SchemaChangeEventType.java

@@ -0,0 +1,63 @@
+/*
+ * 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.event;
+
+import org.apache.flink.cdc.common.annotation.PublicEvolving;
+
+/** An enumeration of schema change event types for {@link SchemaChangeEvent}. */
+@PublicEvolving
+public enum SchemaChangeEventType {
+    ADD_COLUMN,
+    ALTER_COLUMN_TYPE,
+    CREATE_TABLE,
+    DROP_COLUMN,
+    RENAME_COLUMN;
+
+    public static SchemaChangeEventType ofEvent(SchemaChangeEvent event) {
+        if (event instanceof AddColumnEvent) {
+            return ADD_COLUMN;
+        } else if (event instanceof AlterColumnTypeEvent) {
+            return ALTER_COLUMN_TYPE;
+        } else if (event instanceof CreateTableEvent) {
+            return CREATE_TABLE;
+        } else if (event instanceof DropColumnEvent) {
+            return DROP_COLUMN;
+        } else if (event instanceof RenameColumnEvent) {
+            return RENAME_COLUMN;
+        } else {
+            throw new RuntimeException("Unknown schema change event type: " + event.getClass());
+        }
+    }
+
+    public static SchemaChangeEventType ofTag(String tag) {
+        switch (tag) {
+            case "add.column":
+                return ADD_COLUMN;
+            case "alter.column.type":
+                return ALTER_COLUMN_TYPE;
+            case "create.table":
+                return CREATE_TABLE;
+            case "drop.column":
+                return DROP_COLUMN;
+            case "rename.column":
+                return RENAME_COLUMN;
+            default:
+                throw new RuntimeException("Unknown schema change event type: " + tag);
+        }
+    }
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/SchemaChangeEventTypeFamily.java

@@ -0,0 +1,80 @@
+/*
+ * 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.event;
+
+import org.apache.flink.cdc.common.annotation.PublicEvolving;
+
+/**
+ * An enumeration of schema change event families for clustering {@link SchemaChangeEvent}s into
+ * categories.
+ */
+@PublicEvolving
+public class SchemaChangeEventTypeFamily {
+
+    public static final SchemaChangeEventType[] ADD = {SchemaChangeEventType.ADD_COLUMN};
+
+    public static final SchemaChangeEventType[] ALTER = {SchemaChangeEventType.ALTER_COLUMN_TYPE};
+
+    public static final SchemaChangeEventType[] CREATE = {SchemaChangeEventType.CREATE_TABLE};
+
+    public static final SchemaChangeEventType[] DROP = {SchemaChangeEventType.DROP_COLUMN};
+
+    public static final SchemaChangeEventType[] RENAME = {SchemaChangeEventType.RENAME_COLUMN};
+
+    public static final SchemaChangeEventType[] TABLE = {SchemaChangeEventType.CREATE_TABLE};
+
+    public static final SchemaChangeEventType[] COLUMN = {
+        SchemaChangeEventType.ADD_COLUMN,
+        SchemaChangeEventType.ALTER_COLUMN_TYPE,
+        SchemaChangeEventType.DROP_COLUMN,
+        SchemaChangeEventType.RENAME_COLUMN
+    };
+
+    public static final SchemaChangeEventType[] ALL = {
+        SchemaChangeEventType.ADD_COLUMN,
+        SchemaChangeEventType.CREATE_TABLE,
+        SchemaChangeEventType.ALTER_COLUMN_TYPE,
+        SchemaChangeEventType.DROP_COLUMN,
+        SchemaChangeEventType.RENAME_COLUMN
+    };
+
+    public static final SchemaChangeEventType[] NONE = {};
+
+    public static SchemaChangeEventType[] ofTag(String tag) {
+        switch (tag) {
+            case "add":
+                return ADD;
+            case "alter":
+                return ALTER;
+            case "create":
+                return CREATE;
+            case "drop":
+                return DROP;
+            case "rename":
+                return RENAME;
+            case "table":
+                return TABLE;
+            case "column":
+                return COLUMN;
+            case "all":
+                return ALL;
+            default:
+                return NONE;
+        }
+    }
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/exceptions/SchemaEvolveException.java

@@ -0,0 +1,69 @@
+/*
+ * 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.exceptions;
+
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+/** An exception occurred during schema evolution. */
+public class SchemaEvolveException extends FlinkRuntimeException {
+    private final SchemaChangeEvent applyingEvent;
+    private final String exceptionMessage;
+    private final @Nullable Throwable cause;
+
+    public SchemaEvolveException(SchemaChangeEvent applyingEvent, String exceptionMessage) {
+        this(applyingEvent, exceptionMessage, null);
+    }
+
+    public SchemaEvolveException(
+            SchemaChangeEvent applyingEvent, String exceptionMessage, @Nullable Throwable cause) {
+        super(cause);
+        this.applyingEvent = applyingEvent;
+        this.exceptionMessage = exceptionMessage;
+        this.cause = cause;
+    }
+
+    public SchemaChangeEvent getApplyingEvent() {
+        return applyingEvent;
+    }
+
+    public String getExceptionMessage() {
+        return exceptionMessage;
+    }
+
+    @Nullable
+    public Throwable getCause() {
+        return cause;
+    }
+
+    @Override
+    public String toString() {
+        return "SchemaEvolveException{"
+                + "applyingEvent="
+                + applyingEvent
+                + ", exceptionMessage='"
+                + exceptionMessage
+                + '\''
+                + ", cause='"
+                + cause
+                + '\''
+                + '}';
+    }
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/exceptions/UnsupportedSchemaChangeEventException.java

@@ -0,0 +1,28 @@
+/*
+ * 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.exceptions;
+
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+
+/** A special kind of {@link SchemaEvolveException} that sink doesn't support such event type. */
+public class UnsupportedSchemaChangeEventException extends SchemaEvolveException {
+
+    public UnsupportedSchemaChangeEventException(SchemaChangeEvent applyingEvent) {
+        super(applyingEvent, "Sink doesn't support such schema change event.", null);
+    }
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java

@@ -55,9 +55,13 @@ public class PipelineOptions {
                                     .linebreak()
                                     .add(
                                             ListElement.list(
+                                                    text("IGNORE: Drop all schema change events."),
+                                                    text(
+                                                            "LENIENT: Apply schema changes to downstream tolerantly, and keeps executing if applying fails."),
+                                                    text(
+                                                            "TRY_EVOLVE: Apply schema changes to downstream, but keeps executing if applying fails."),
                                                     text(
                                                             "EVOLVE: Apply schema changes to downstream. This requires sink to support handling schema changes."),
-                                                    text("IGNORE: Drop all schema change events."),
                                                     text(
                                                             "EXCEPTION: Throw an exception to terminate the sync pipeline.")))
                                     .build());