[FLINK-36985][pipeline-connector/paimon] Tolerante ColumnAlreadyExistException when apply AddColumnEvent in paimon

hiliuxg · web-flow · commit 32934393cefc · 2025-01-14T11:50:49.000+08:00
This closes #3828.
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/PaimonMetadataApplier.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplier.java
@@ -201,7 +201,11 @@ private void applyAddColumn(AddColumnEvent event) throws SchemaEvolveException {
         } catch (Catalog.TableNotExistException
                 | Catalog.ColumnAlreadyExistException
                 | Catalog.ColumnNotExistException e) {
-            throw new SchemaEvolveException(event, e.getMessage(), e);
+            if (e instanceof Catalog.ColumnAlreadyExistException) {
+                LOG.warn("{}, skip it.", e.getMessage());
+            } else {
+                throw new SchemaEvolveException(event, e.getMessage(), e);
+            }
         }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -201,7 +201,11 @@ private void applyAddColumn(AddColumnEvent event) throws SchemaEvolveException {`
`201`	`201`	`} catch (Catalog.TableNotExistException`
`202`	`202`	`\| Catalog.ColumnAlreadyExistException`
`203`	`203`	`\| Catalog.ColumnNotExistException e) {`
`204`		`- throw new SchemaEvolveException(event, e.getMessage(), e);`
	`204`	`+ if (e instanceof Catalog.ColumnAlreadyExistException) {`
	`205`	`+ LOG.warn("{}, skip it.", e.getMessage());`
	`206`	`+ } else {`
	`207`	`+ throw new SchemaEvolveException(event, e.getMessage(), e);`
	`208`	`+ }`
`205`	`209`	`}`
`206`	`210`	`}`
`207`	`211`