[INLONG-7787][Sort] Fix cannot output drop statement (#7788)

apache · Apr 8, 2023 · e68baeb · e68baeb
1 parent 4fb105c
commit e68baeb
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 0 deletions.
diff --git a/inlong-sort/sort-connectors/base/src/main/java/org/apache/inlong/sort/base/Constants.java b/inlong-sort/sort-connectors/base/src/main/java/org/apache/inlong/sort/base/Constants.java
@@ -136,6 +136,8 @@ public final class Constants {
 
     public static final String DDL_FIELD_NAME = "ddl";
 
+    public static final String DDL_OP_DROP = "DROP";
+
     public static final ConfigOption<String> INLONG_METRIC =
             ConfigOptions.key("inlong.metric.labels")
                     .stringType()

diff --git a/...-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/reader/MySqlRecordEmitter.java b/...-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/reader/MySqlRecordEmitter.java
@@ -23,6 +23,7 @@
 import io.debezium.document.Array;
 import io.debezium.relational.TableId;
 import io.debezium.relational.history.HistoryRecord;
+import io.debezium.relational.history.HistoryRecord.Fields;
 import io.debezium.relational.history.TableChanges;
 import io.debezium.relational.history.TableChanges.TableChange;
 import org.apache.flink.api.connector.source.SourceOutput;
@@ -42,6 +43,7 @@
 
 import java.util.Map;
 
+import static org.apache.inlong.sort.base.Constants.DDL_OP_DROP;
 import static org.apache.inlong.sort.cdc.mysql.source.utils.RecordUtils.getBinlogPosition;
 import static org.apache.inlong.sort.cdc.mysql.source.utils.RecordUtils.getFetchTimestamp;
 import static org.apache.inlong.sort.cdc.mysql.source.utils.RecordUtils.getHistoryRecord;
@@ -116,6 +118,18 @@ public void emitRecord(SourceRecord element, SourceOutput<T> output, MySqlSplitS
                 }
             }
 
+            // for drop table ddl, there's no table change events
+            if (tableChanges.isEmpty()) {
+                String ddl = historyRecord.document().getString(Fields.DDL_STATEMENTS);
+                if (ddl.toUpperCase().startsWith(DDL_OP_DROP)) {
+                    TableId tableId = RecordUtils.getTableId(element);
+                    // if this table is one of the captured tables, output the ddl element
+                    if (splitState.getMySQLSplit().getTableSchemas().containsKey(tableId)) {
+                        outputDdlElement(element, output, splitState, null);
+                    }
+                }
+            }
+
         } else if (isDataChangeRecord(element)) {
             if (splitState.isBinlogSplitState()) {
                 BinlogOffset position = getBinlogPosition(element);