[hotfix] Optimize SchemaChangeEventVisitor: use generic visitor handler class & introduce voidVisitors

Author: yuxiqian

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

@@ -33,14 +33,14 @@
 public class SchemaChangeEventVisitor {
     public static <T, E extends Throwable> T visit(
             SchemaChangeEvent event,
-            AddColumnEventVisitor<T, E> addColumnVisitor,
-            AlterColumnTypeEventVisitor<T, E> alterColumnTypeEventVisitor,
-            CreateTableEventVisitor<T, E> createTableEventVisitor,
-            DropColumnEventVisitor<T, E> dropColumnEventVisitor,
-            DropTableEventVisitor<T, E> dropTableEventVisitor,
-            RenameColumnEventVisitor<T, E> renameColumnEventVisitor,
-            TruncateTableEventVisitor<T, E> truncateTableEventVisitor,
-            AlterTableCommentEventVisitor<T, E> alterTableCommentEventVisitor)
+            VisitorHandler<AddColumnEvent, T, E> addColumnVisitor,
+            VisitorHandler<AlterColumnTypeEvent, T, E> alterColumnTypeEventVisitor,
+            VisitorHandler<CreateTableEvent, T, E> createTableEventVisitor,
+            VisitorHandler<DropColumnEvent, T, E> dropColumnEventVisitor,
+            VisitorHandler<DropTableEvent, T, E> dropTableEventVisitor,
+            VisitorHandler<RenameColumnEvent, T, E> renameColumnEventVisitor,
+            VisitorHandler<TruncateTableEvent, T, E> truncateTableEventVisitor,
+            VisitorHandler<AlterTableCommentEvent, T, E> alterTableCommentEventVisitor)
             throws E {
         if (event instanceof AddColumnEvent) {
             if (addColumnVisitor == null) {
@@ -87,4 +87,53 @@ public static <T, E extends Throwable> T visit(
                     "Unknown schema change event type " + event.getType());
         }
     }
+
+    public static <E extends Throwable> void voidVisit(
+            SchemaChangeEvent event,
+            VoidVisitorHandler<AddColumnEvent, E> addColumnVisitor,
+            VoidVisitorHandler<AlterColumnTypeEvent, E> alterColumnTypeEventVisitor,
+            VoidVisitorHandler<CreateTableEvent, E> createTableEventVisitor,
+            VoidVisitorHandler<DropColumnEvent, E> dropColumnEventVisitor,
+            VoidVisitorHandler<DropTableEvent, E> dropTableEventVisitor,
+            VoidVisitorHandler<RenameColumnEvent, E> renameColumnEventVisitor,
+            VoidVisitorHandler<TruncateTableEvent, E> truncateTableEventVisitor,
+            VoidVisitorHandler<AlterTableCommentEvent, E> alterTableCommentEventHandler)
+            throws E {
+        if (event instanceof AddColumnEvent) {
+            if (addColumnVisitor != null) {
+                addColumnVisitor.visit((AddColumnEvent) event);
+            }
+        } else if (event instanceof AlterColumnTypeEvent) {
+            if (alterColumnTypeEventVisitor != null) {
+                alterColumnTypeEventVisitor.visit((AlterColumnTypeEvent) event);
+            }
+        } else if (event instanceof CreateTableEvent) {
+            if (createTableEventVisitor != null) {
+                createTableEventVisitor.visit((CreateTableEvent) event);
+            }
+        } else if (event instanceof DropColumnEvent) {
+            if (dropColumnEventVisitor != null) {
+                dropColumnEventVisitor.visit((DropColumnEvent) event);
+            }
+        } else if (event instanceof DropTableEvent) {
+            if (dropTableEventVisitor != null) {
+                dropTableEventVisitor.visit((DropTableEvent) event);
+            }
+        } else if (event instanceof RenameColumnEvent) {
+            if (renameColumnEventVisitor != null) {
+                renameColumnEventVisitor.visit((RenameColumnEvent) event);
+            }
+        } else if (event instanceof TruncateTableEvent) {
+            if (truncateTableEventVisitor != null) {
+                truncateTableEventVisitor.visit((TruncateTableEvent) event);
+            }
+        } else if (event instanceof AlterTableCommentEvent) {
+            if (alterTableCommentEventHandler != null) {
+                alterTableCommentEventHandler.visit((AlterTableCommentEvent) event);
+            }
+        } else {
+            throw new IllegalArgumentException(
+                    "Unknown schema change event type " + event.getType());
+        }
+    }
 }

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

@@ -18,11 +18,12 @@
 package org.apache.flink.cdc.common.event.visitor;
 
 import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
 import org.apache.flink.cdc.common.event.TruncateTableEvent;
 
 /** Visitor for {@link TruncateTableEvent}s. */
 @Internal
 @FunctionalInterface
-public interface TruncateTableEventVisitor<T, E extends Throwable> {
-    T visit(TruncateTableEvent event) throws E;
+public interface VisitorHandler<EVT extends SchemaChangeEvent, T, E extends Throwable> {
+    T visit(EVT event) throws E;
 }

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

@@ -18,11 +18,12 @@
 package org.apache.flink.cdc.common.event.visitor;
 
 import org.apache.flink.cdc.common.annotation.Internal;
-import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TruncateTableEvent;
 
-/** Visitor for {@link AlterColumnTypeEvent}s. */
+/** Visitor for {@link TruncateTableEvent}s. */
 @Internal
 @FunctionalInterface
-public interface AlterColumnTypeEventVisitor<T, E extends Throwable> {
-    T visit(AlterColumnTypeEvent event) throws E;
+public interface VoidVisitorHandler<EVT extends SchemaChangeEvent, E extends Throwable> {
+    void visit(EVT event) throws E;
 }

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

@@ -112,40 +112,16 @@ public Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
 
     @Override
     public void applySchemaChange(SchemaChangeEvent event) {
-        SchemaChangeEventVisitor.<Void, SchemaEvolveException>visit(
+        SchemaChangeEventVisitor.voidVisit(
                 event,
-                addColumnEvent -> {
-                    applyAddColumnEvent(addColumnEvent);
-                    return null;
-                },
-                alterColumnTypeEvent -> {
-                    applyAlterColumnTypeEvent(alterColumnTypeEvent);
-                    return null;
-                },
-                createTableEvent -> {
-                    applyCreateTableEvent(createTableEvent);
-                    return null;
-                },
-                dropColumnEvent -> {
-                    applyDropColumnEvent(dropColumnEvent);
-                    return null;
-                },
-                dropTableEvent -> {
-                    applyDropTableEvent(dropTableEvent);
-                    return null;
-                },
-                renameColumnEvent -> {
-                    applyRenameColumnEvent(renameColumnEvent);
-                    return null;
-                },
-                truncateTableEvent -> {
-                    applyTruncateTableEvent(truncateTableEvent);
-                    return null;
-                },
-                alterTableCommentEvent -> {
-                    applyAlterTableCommentEvent(alterTableCommentEvent);
-                    return null;
-                });
+                this::applyAddColumnEvent,
+                this::applyAlterColumnTypeEvent,
+                this::applyCreateTableEvent,
+                this::applyDropColumnEvent,
+                this::applyDropTableEvent,
+                this::applyRenameColumnEvent,
+                this::applyTruncateTableEvent,
+                this::applyAlterTableCommentEvent);
     }
 
     private void applyCreateTableEvent(CreateTableEvent event) throws SchemaEvolveException {