This is an automated email from the ASF dual-hosted git repository.
leonard pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 7f08c6cab [FLINK-36088][pipeline-connector][paimon] Fix NPE in
BucketAssignOperator when job restoration
7f08c6cab is described below
commit 7f08c6caba082915b9684ea384363f1df85dab8d
Author: Kunni <[email protected]>
AuthorDate: Tue Aug 20 00:50:22 2024 +0800
[FLINK-36088][pipeline-connector][paimon] Fix NPE in BucketAssignOperator
when job restoration
This closes #3553
---
.../cdc/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 07509574c..358d7c36c 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 class BucketAssignOperator extends
AbstractStreamOperator<Event>
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()) {
