From 7f08c6caba082915b9684ea384363f1df85dab8d Mon Sep 17 00:00:00 2001
From: Kunni <lvyanquan.lyq@alibaba-inc.com>
Date: Tue, 20 Aug 2024 00:50:22 +0800
Subject: [PATCH] [FLINK-36088][pipeline-connector][paimon] Fix NPE in
 BucketAssignOperator when job restoration

This closes #3553
---
 .../connectors/paimon/sink/v2/bucket/BucketAssignOperator.java  | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java
index 07509574c6..358d7c36ce 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketAssignOperator.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/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()) {