[hotfix] Fix job failure due to dangling DropTableEvent (without a prior CreateTableEvent)

Author: yuxiqian

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eITCase.java

@@ -331,6 +331,64 @@ public void testSchemaChangeEvents() throws Exception {
                 "DropTableEvent{tableId=%s.products}");
     }
 
+    @Test
+    public void testDroppingTable() throws Exception {
+        Thread.sleep(5000);
+        LOG.info("Sleep 5 seconds to distinguish initial DDL events with dropping table events...");
+        long ddlTimestamp = System.currentTimeMillis();
+        Thread.sleep(5000);
+        LOG.info("Going to drop tables after timestamp {}", ddlTimestamp);
+
+        try (Connection connection = mysqlInventoryDatabase.getJdbcConnection();
+                Statement statement = connection.createStatement()) {
+            statement.execute("DROP TABLE products;");
+        }
+
+        String pipelineJob =
+                String.format(
+                        "source:\n"
+                                + "  type: mysql\n"
+                                + "  hostname: %s\n"
+                                + "  port: 3306\n"
+                                + "  username: %s\n"
+                                + "  password: %s\n"
+                                + "  tables: %s.\\.*\n"
+                                + "  server-id: 5400-5404\n"
+                                + "  server-time-zone: UTC\n"
+                                + "  scan.startup.mode: timestamp\n"
+                                + "  scan.startup.timestamp-millis: %d\n"
+                                + "  scan.binlog.newly-added-table.enabled: true\n"
+                                + "\n"
+                                + "sink:\n"
+                                + "  type: values\n"
+                                + "\n"
+                                + "pipeline:\n"
+                                + "  parallelism: %d\n"
+                                + "  schema.change.behavior: evolve",
+                        INTER_CONTAINER_MYSQL_ALIAS,
+                        MYSQL_TEST_USER,
+                        MYSQL_TEST_PASSWORD,
+                        mysqlInventoryDatabase.getDatabaseName(),
+                        ddlTimestamp,
+                        parallelism);
+        Path mysqlCdcJar = TestUtils.getResource("mysql-cdc-pipeline-connector.jar");
+        Path valuesCdcJar = TestUtils.getResource("values-cdc-pipeline-connector.jar");
+        Path mysqlDriverJar = TestUtils.getResource("mysql-driver.jar");
+        submitPipelineJob(pipelineJob, mysqlCdcJar, valuesCdcJar, mysqlDriverJar);
+        waitUntilJobRunning(Duration.ofSeconds(30));
+        LOG.info("Pipeline job is running");
+        waitUntilSpecificEvent(
+                String.format(
+                        "Table %s.products received SchemaChangeEvent DropTableEvent{tableId=%s.products} and start to be blocked.",
+                        mysqlInventoryDatabase.getDatabaseName(),
+                        mysqlInventoryDatabase.getDatabaseName()));
+
+        waitUntilSpecificEvent(
+                String.format(
+                        "Schema change event DropTableEvent{tableId=%s.products} has been handled in another subTask already.",
+                        mysqlInventoryDatabase.getDatabaseName()));
+    }
+
     private void validateResult(String... expectedEvents) throws Exception {
         String dbName = mysqlInventoryDatabase.getDatabaseName();
         for (String event : expectedEvents) {

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/SchemaOperator.java

@@ -244,7 +244,13 @@ private void processSchemaChangeEvents(SchemaChangeEvent event)
                 tableId,
                 event);
         handleSchemaChangeEvent(tableId, event);
-        // Update caches
+
+        if (event instanceof DropTableEvent) {
+            // Update caches unless event is a Drop table event. In that case, no schema will be
+            // available / necessary.
+            return;
+        }
+
         originalSchema.put(tableId, getLatestOriginalSchema(tableId));
         schemaDivergesMap.put(tableId, checkSchemaDiverges(tableId));