[FLINK-36088][pipeline-connector][paimon] Fix NPE in BucketAssignOper…

…ator when job restoration This closes #3553
apache · Aug 19, 2024 · 7f08c6c · 7f08c6c
1 parent 0470fdb
commit 7f08c6c
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/...main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java b/...main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java
@@ -130,7 +130,7 @@ public void processElement(StreamRecord<Event> streamRecord) throws Exception {
 
         if (event instanceof DataChangeEvent) {
             DataChangeEvent dataChangeEvent = (DataChangeEvent) event;
-            if (schemaMaps.containsKey(dataChangeEvent.tableId())) {
+            if (!schemaMaps.containsKey(dataChangeEvent.tableId())) {
                 Optional<Schema> schema =
                         schemaEvolutionClient.getLatestEvolvedSchema(dataChangeEvent.tableId());
                 if (schema.isPresent()) {