hotfix: transform with schema evolution issue

apache · Aug 19, 2024 · e137bdf · e137bdf
1 parent 0b341b6
commit e137bdf
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 2 deletions.
diff --git a/...ne-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaEvolveE2eITCase.java b/...ne-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaEvolveE2eITCase.java
@@ -192,8 +192,12 @@ public void testLenientSchemaEvolution() throws Exception {
                         "AddColumnEvent{tableId=%s.members, addedColumns=[ColumnWithPosition{column=`biological_sex` TINYINT, position=LAST, existedColumnName=null}]}",
                         "DataChangeEvent{tableId=%s.members, before=[], after=[1013, Fiona, null, null, 16.0, null], op=INSERT, meta=()}",
                         "TruncateTableEvent{tableId=%s.members}",
-                        "DataChangeEvent{tableId=%s.members, before=[], after=[1014, Gem, null, null, 17.0, null], op=INSERT, meta=()}",
-                        "DropTableEvent{tableId=%s.members}"));
+                        "DataChangeEvent{tableId=%s.members, before=[], after=[1014, Gem, null, null, 17.0, null], op=INSERT, meta=()}"));
+
+        assertNotExists(
+                Collections.singletonList(
+                        "Applied schema change event DropTableEvent{tableId=%s.members}"),
+                taskManagerConsumer);
     }
 
     @Test
@@ -404,6 +408,17 @@ private void validateResult(List<String> expectedEvents, ToStringConsumer consum
         }
     }
 
+    private void assertNotExists(List<String> unexpectedEvents, ToStringConsumer consumer) {
+        String consumerLog = consumer.toUtf8String();
+        System.out.println(consumerLog);
+        for (String event : unexpectedEvents) {
+            System.out.println("Assuming there's no " + event + " in logs...");
+            Assert.assertFalse(
+                    consumerLog.contains(
+                            String.format(event, schemaEvolveDatabase.getDatabaseName())));
+        }
+    }
+
     private void waitUntilSpecificEvent(String event, ToStringConsumer consumer) throws Exception {
         boolean result = false;
         long endTimeout = System.currentTimeMillis() + SchemaEvolveE2eITCase.EVENT_WAITING_TIMEOUT;

diff --git a/...g/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java b/...g/apache/flink/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java
@@ -428,6 +428,9 @@ private List<SchemaChangeEvent> lenientizeSchemaChangeEvent(SchemaChangeEvent ev
                     }
                     return events;
                 }
+            case DROP_TABLE:
+                // We don't drop any tables in Lenient mode.
+                return Collections.emptyList();
             default:
                 return Collections.singletonList(event);
         }