[FLINK-36964][pipeline-connector/paimon] Fix potential exception when SchemaChange in parallel with Paimon Sink

This closes #3818

Co-authored-by: yuxiqian.yxq <[email protected]>

Author: lvyanquan

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/pom.xml

@@ -285,6 +285,19 @@ limitations under the License.
                     </execution>
                 </executions>
             </plugin>
+
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>test-jar</id>
+                        <goals>
+                            <goal>test-jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonDataSinkFactory.java

@@ -24,12 +24,9 @@
 import org.apache.flink.cdc.common.factories.FactoryHelper;
 import org.apache.flink.cdc.common.pipeline.PipelineOptions;
 import org.apache.flink.cdc.common.sink.DataSink;
-import org.apache.flink.cdc.common.utils.Preconditions;
 import org.apache.flink.cdc.connectors.paimon.sink.v2.PaimonRecordEventSerializer;
 import org.apache.flink.cdc.connectors.paimon.sink.v2.PaimonRecordSerializer;
 
-import org.apache.paimon.catalog.Catalog;
-import org.apache.paimon.flink.FlinkCatalogFactory;
 import org.apache.paimon.options.Options;
 
 import java.time.ZoneId;
@@ -71,12 +68,6 @@ public DataSink createDataSink(Context context) {
         Options options = Options.fromMap(catalogOptions);
         // Avoid using previous table schema.
         options.setString("cache-enabled", "false");
-        try (Catalog catalog = FlinkCatalogFactory.createPaimonCatalog(options)) {
-            Preconditions.checkNotNull(
-                    catalog.listDatabases(), "catalog option of Paimon is invalid.");
-        } catch (Exception e) {
-            throw new RuntimeException("failed to create or use paimon catalog", e);
-        }
         ZoneId zoneId = ZoneId.systemDefault();
         if (!Objects.equals(
                 context.getPipelineConfiguration().get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE),

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonEventSink.java

@@ -21,6 +21,7 @@
 import org.apache.flink.cdc.connectors.paimon.sink.v2.bucket.BucketAssignOperator;
 import org.apache.flink.cdc.connectors.paimon.sink.v2.bucket.BucketWrapper;
 import org.apache.flink.cdc.connectors.paimon.sink.v2.bucket.BucketWrapperEventTypeInfo;
+import org.apache.flink.cdc.connectors.paimon.sink.v2.bucket.FlushEventAlignmentOperator;
 import org.apache.flink.core.io.SimpleVersionedSerializer;
 import org.apache.flink.streaming.api.connector.sink2.WithPreWriteTopology;
 import org.apache.flink.streaming.api.datastream.DataStream;
@@ -63,7 +64,12 @@ public DataStream<Event> addPreWriteTopology(DataStream<Event> dataStream) {
                 // All Events after BucketAssignOperator are decorated with BucketWrapper.
                 .partitionCustom(
                         (bucket, numPartitions) -> bucket % numPartitions,
-                        (event) -> ((BucketWrapper) event).getBucket());
+                        (event) -> ((BucketWrapper) event).getBucket())
+                // Avoid disorder of FlushEvent and DataChangeEvent.
+                .transform(
+                        "FlushEventAlignment",
+                        new BucketWrapperEventTypeInfo(),
+                        new FlushEventAlignmentOperator());
     }
 
     @Override

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonRecordEventSerializer.java

@@ -23,6 +23,7 @@
 import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.common.event.SchemaChangeEvent;
 import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
 import org.apache.flink.cdc.common.utils.SchemaUtils;
 import org.apache.flink.cdc.connectors.paimon.sink.v2.bucket.BucketWrapperChangeEvent;
 
@@ -66,13 +67,14 @@ public PaimonEvent serialize(Event event) {
                         new TableSchemaInfo(createTableEvent.getSchema(), zoneId));
             } else {
                 SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event;
-                schemaMaps.put(
-                        schemaChangeEvent.tableId(),
-                        new TableSchemaInfo(
-                                SchemaUtils.applySchemaChangeEvent(
-                                        schemaMaps.get(schemaChangeEvent.tableId()).getSchema(),
-                                        schemaChangeEvent),
-                                zoneId));
+                Schema schema = schemaMaps.get(schemaChangeEvent.tableId()).getSchema();
+                if (!SchemaUtils.isSchemaChangeEventRedundant(schema, schemaChangeEvent)) {
+                    schemaMaps.put(
+                            schemaChangeEvent.tableId(),
+                            new TableSchemaInfo(
+                                    SchemaUtils.applySchemaChangeEvent(schema, schemaChangeEvent),
+                                    zoneId));
+                }
             }
             return new PaimonEvent(tableId, null, true);
         } else if (event instanceof DataChangeEvent) {

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSink.java

@@ -94,7 +94,10 @@ public DataStream<CommittableMessage<MultiTableCommittable>> addPreCommitTopolog
 
         // add correct checkpointId to MultiTableCommittable and recreate CommittableSummary.
         return partitioned
-                .transform("preCommit", typeInformation, new PreCommitOperator())
+                .transform(
+                        "preCommit",
+                        typeInformation,
+                        new PreCommitOperator(catalogOptions, commitUser))
                 .setParallelism(committables.getParallelism());
     }
 }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PreCommitOperator.java

@@ -17,16 +17,25 @@
 
 package org.apache.flink.cdc.connectors.paimon.sink.v2;
 
+import org.apache.flink.runtime.state.StateSnapshotContext;
 import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
-import org.apache.flink.streaming.api.connector.sink2.CommittableSummary;
 import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
 import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.flink.sink.Committer;
 import org.apache.paimon.flink.sink.MultiTableCommittable;
+import org.apache.paimon.flink.sink.StoreMultiCommitter;
+import org.apache.paimon.manifest.WrappedManifestCommittable;
+import org.apache.paimon.options.Options;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 
 /** An Operator to add checkpointId to MultiTableCommittable and generate CommittableSummary. */
@@ -35,12 +44,23 @@ public class PreCommitOperator
         implements OneInputStreamOperator<
                 CommittableMessage<MultiTableCommittable>,
                 CommittableMessage<MultiTableCommittable>> {
+    protected static final Logger LOGGER = LoggerFactory.getLogger(PreCommitOperator.class);
+
+    private final String commitUser;
+
+    private final Options catalogOptions;
+
+    private Catalog catalog;
+
+    private StoreMultiCommitter storeMultiCommitter;
 
     /** store a list of MultiTableCommittable in one checkpoint. */
-    private final List<MultiTableCommittable> results;
+    private final List<MultiTableCommittable> multiTableCommittables;
 
-    public PreCommitOperator() {
-        results = new ArrayList<>();
+    public PreCommitOperator(Options catalogOptions, String commitUser) {
+        multiTableCommittables = new ArrayList<>();
+        this.catalogOptions = catalogOptions;
+        this.commitUser = commitUser;
     }
 
     @Override
@@ -50,8 +70,16 @@ public void open() throws Exception {
 
     @Override
     public void processElement(StreamRecord<CommittableMessage<MultiTableCommittable>> element) {
+        if (catalog == null) {
+            this.catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
+            this.storeMultiCommitter =
+                    new StoreMultiCommitter(
+                            () -> FlinkCatalogFactory.createPaimonCatalog(catalogOptions),
+                            Committer.createContext(
+                                    commitUser, getMetricGroup(), true, false, null));
+        }
         if (element.getValue() instanceof CommittableWithLineage) {
-            results.add(
+            multiTableCommittables.add(
                     ((CommittableWithLineage<MultiTableCommittable>) element.getValue())
                             .getCommittable());
         }
@@ -64,34 +92,44 @@ public void finish() {
 
     @Override
     public void prepareSnapshotPreBarrier(long checkpointId) {
-        // CommittableSummary should be sent before all CommittableWithLineage.
-        CommittableMessage<MultiTableCommittable> summary =
-                new CommittableSummary<>(
-                        getRuntimeContext().getIndexOfThisSubtask(),
-                        getRuntimeContext().getNumberOfParallelSubtasks(),
-                        checkpointId,
-                        results.size(),
-                        results.size(),
-                        0);
-        output.collect(new StreamRecord<>(summary));
-
-        results.forEach(
-                committable -> {
-                    // update the right checkpointId for MultiTableCommittable
-                    MultiTableCommittable committableWithCheckPointId =
-                            new MultiTableCommittable(
-                                    committable.getDatabase(),
-                                    committable.getTable(),
-                                    checkpointId,
-                                    committable.kind(),
-                                    committable.wrappedCommittable());
-                    CommittableMessage<MultiTableCommittable> message =
-                            new CommittableWithLineage<>(
-                                    committableWithCheckPointId,
-                                    checkpointId,
-                                    getRuntimeContext().getIndexOfThisSubtask());
-                    output.collect(new StreamRecord<>(message));
-                });
-        results.clear();
+        for (int i = 0; i < multiTableCommittables.size(); i++) {
+            MultiTableCommittable multiTableCommittable = multiTableCommittables.get(i);
+            multiTableCommittables.set(
+                    i,
+                    new MultiTableCommittable(
+                            multiTableCommittable.getDatabase(),
+                            multiTableCommittable.getTable(),
+                            checkpointId,
+                            multiTableCommittable.kind(),
+                            multiTableCommittable.wrappedCommittable()));
+        }
+    }
+
+    @Override
+    public void snapshotState(StateSnapshotContext context) throws Exception {
+        super.snapshotState(context);
+        long checkpointId = context.getCheckpointId();
+        if (!multiTableCommittables.isEmpty()) {
+            multiTableCommittables.forEach(
+                    (multiTableCommittable) ->
+                            LOGGER.debug(
+                                    "Try to commit for {}.{} : {} in checkpoint {}",
+                                    multiTableCommittable.getDatabase(),
+                                    multiTableCommittable.getTable(),
+                                    multiTableCommittables,
+                                    checkpointId));
+            WrappedManifestCommittable wrappedManifestCommittable =
+                    storeMultiCommitter.combine(checkpointId, checkpointId, multiTableCommittables);
+            storeMultiCommitter.commit(Collections.singletonList(wrappedManifestCommittable));
+            multiTableCommittables.clear();
+        }
+    }
+
+    @Override
+    public void close() throws Exception {
+        super.close();
+        if (storeMultiCommitter != null) {
+            storeMultiCommitter.close();
+        }
     }
 }

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

@@ -19,7 +19,6 @@
 
 import org.apache.flink.api.java.tuple.Tuple4;
 import org.apache.flink.cdc.common.event.ChangeEvent;
-import org.apache.flink.cdc.common.event.CreateTableEvent;
 import org.apache.flink.cdc.common.event.DataChangeEvent;
 import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.common.event.FlushEvent;
@@ -53,6 +52,8 @@
 import org.apache.paimon.table.sink.RowKeyExtractor;
 import org.apache.paimon.table.sink.RowPartitionKeyExtractor;
 import org.apache.paimon.utils.MathUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.time.ZoneId;
 import java.util.HashMap;
@@ -63,6 +64,8 @@
 public class BucketAssignOperator extends AbstractStreamOperator<Event>
         implements OneInputStreamOperator<Event, Event> {
 
+    protected static final Logger LOGGER = LoggerFactory.getLogger(BucketAssignOperator.class);
+
     public final String commitUser;
 
     private final Options catalogOptions;
@@ -99,8 +102,8 @@ public void open() throws Exception {
         super.open();
         this.catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
         this.bucketAssignerMap = new HashMap<>();
-        this.totalTasksNumber = getRuntimeContext().getNumberOfParallelSubtasks();
-        this.currentTaskNumber = getRuntimeContext().getIndexOfThisSubtask();
+        this.totalTasksNumber = getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks();
+        this.currentTaskNumber = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask();
         this.schemaMaps = new HashMap<>();
     }
 
@@ -121,13 +124,16 @@ public void setup(
     public void processElement(StreamRecord<Event> streamRecord) throws Exception {
         Event event = streamRecord.getValue();
         if (event instanceof FlushEvent) {
-            output.collect(
-                    new StreamRecord<>(
-                            new BucketWrapperFlushEvent(
-                                    currentTaskNumber,
-                                    ((FlushEvent) event).getSourceSubTaskId(),
-                                    ((FlushEvent) event).getTableIds(),
-                                    ((FlushEvent) event).getSchemaChangeEventType())));
+            for (int i = 0; i < totalTasksNumber; i++) {
+                output.collect(
+                        new StreamRecord<>(
+                                new BucketWrapperFlushEvent(
+                                        i,
+                                        ((FlushEvent) event).getSourceSubTaskId(),
+                                        currentTaskNumber,
+                                        ((FlushEvent) event).getTableIds(),
+                                        ((FlushEvent) event).getSchemaChangeEventType())));
+            }
             return;
         }
 
@@ -181,24 +187,21 @@ public void processElement(StreamRecord<Event> streamRecord) throws Exception {
             }
             output.collect(
                     new StreamRecord<>(new BucketWrapperChangeEvent(bucket, (ChangeEvent) event)));
-        } else if (event instanceof CreateTableEvent) {
-            CreateTableEvent createTableEvent = (CreateTableEvent) event;
-            schemaMaps.put(
-                    createTableEvent.tableId(),
-                    new TableSchemaInfo(createTableEvent.getSchema(), zoneId));
-            output.collect(
-                    new StreamRecord<>(
-                            new BucketWrapperChangeEvent(currentTaskNumber, (ChangeEvent) event)));
         } else if (event instanceof SchemaChangeEvent) {
             SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event;
             Schema schema =
                     SchemaUtils.applySchemaChangeEvent(
-                            schemaMaps.get(schemaChangeEvent.tableId()).getSchema(),
+                            Optional.ofNullable(schemaMaps.get(schemaChangeEvent.tableId()))
+                                    .map(TableSchemaInfo::getSchema)
+                                    .orElse(null),
                             schemaChangeEvent);
             schemaMaps.put(schemaChangeEvent.tableId(), new TableSchemaInfo(schema, zoneId));
-            output.collect(
-                    new StreamRecord<>(
-                            new BucketWrapperChangeEvent(currentTaskNumber, (ChangeEvent) event)));
+            // Broadcast SchemachangeEvent.
+            for (int index = 0; index < totalTasksNumber; index++) {
+                output.collect(
+                        new StreamRecord<>(
+                                new BucketWrapperChangeEvent(index, (ChangeEvent) event)));
+            }
         }
     }
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/bucket/BucketWrapperEventSerializer.java

@@ -87,6 +87,7 @@ public void serialize(Event event, DataOutputView dataOutputView) throws IOExcep
             BucketWrapperFlushEvent bucketWrapperFlushEvent = (BucketWrapperFlushEvent) event;
             dataOutputView.writeInt(bucketWrapperFlushEvent.getBucket());
             dataOutputView.writeInt(bucketWrapperFlushEvent.getSourceSubTaskId());
+            dataOutputView.writeInt(bucketWrapperFlushEvent.getBucketAssignTaskId());
             tableIdListSerializer.serialize(bucketWrapperFlushEvent.getTableIds(), dataOutputView);
             schemaChangeEventTypeEnumSerializer.serialize(
                     bucketWrapperFlushEvent.getSchemaChangeEventType(), dataOutputView);
@@ -98,6 +99,7 @@ public Event deserialize(DataInputView source) throws IOException {
         EventClass eventClass = enumSerializer.deserialize(source);
         if (eventClass.equals(EventClass.BUCKET_WRAPPER_FLUSH_EVENT)) {
             return new BucketWrapperFlushEvent(
+                    source.readInt(),
                     source.readInt(),
                     source.readInt(),
                     tableIdListSerializer.deserialize(source),