remove Catalog.ColumnAlreadyExistException when apply applyAddColumnEventWithPosition in paimon. because use may create table already before with the target column.

Author: [email protected]

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);
+            }
         }
     }
 
@@ -293,11 +297,7 @@ private void applyDropColumn(DropColumnEvent event) throws SchemaEvolveException
         } catch (Catalog.TableNotExistException
                 | Catalog.ColumnAlreadyExistException
                 | Catalog.ColumnNotExistException e) {
-            if (e instanceof Catalog.ColumnAlreadyExistException) {
-                LOG.warn("{}, skip it.", e.getMessage());
-            } else {
-                throw new SchemaEvolveException(event, e.getMessage(), e);
-            }
+            throw new SchemaEvolveException(event, e.getMessage(), e);
         }
     }