You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2025/06/09 07:14:01 UTC
(flink-cdc) branch master updated: [FLINK-37860][pipeline-connector/paimon]Fix Committer.createContext not passing OperatorStateStore correctly (#4029)
This is an automated email from the ASF dual-hosted git repository.
kunni 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 2fcf6f216 [FLINK-37860][pipeline-connector/paimon]Fix Committer.createContext not passing OperatorStateStore correctly (#4029)
2fcf6f216 is described below
commit 2fcf6f216aa74da4e0309d261e95f3462c2d1cc5
Author: MOBIN <18...@163.com>
AuthorDate: Mon Jun 9 15:13:58 2025 +0800
[FLINK-37860][pipeline-connector/paimon]Fix Committer.createContext not passing OperatorStateStore correctly (#4029)
---
.../paimon/sink/v2/PreCommitOperator.java | 30 ++++-
.../paimon/sink/v2/PaimonSinkITCase.java | 137 ++++++++++++++++++++-
2 files changed, 161 insertions(+), 6 deletions(-)
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/PreCommitOperator.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/PreCommitOperator.java
index 87338690b..927ef1268 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/PreCommitOperator.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/PreCommitOperator.java
@@ -17,10 +17,12 @@
package org.apache.flink.cdc.connectors.paimon.sink.v2;
+import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -34,6 +36,7 @@ import org.apache.paimon.options.Options;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -42,8 +45,9 @@ import java.util.List;
public class PreCommitOperator
extends AbstractStreamOperator<CommittableMessage<MultiTableCommittable>>
implements OneInputStreamOperator<
- CommittableMessage<MultiTableCommittable>,
- CommittableMessage<MultiTableCommittable>> {
+ CommittableMessage<MultiTableCommittable>,
+ CommittableMessage<MultiTableCommittable>>,
+ BoundedOneInput {
protected static final Logger LOGGER = LoggerFactory.getLogger(PreCommitOperator.class);
private final String commitUser;
@@ -69,15 +73,24 @@ public class PreCommitOperator
}
@Override
- public void processElement(StreamRecord<CommittableMessage<MultiTableCommittable>> element) {
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
if (catalog == null) {
this.catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
this.storeMultiCommitter =
new StoreMultiCommitter(
() -> FlinkCatalogFactory.createPaimonCatalog(catalogOptions),
Committer.createContext(
- commitUser, getMetricGroup(), true, false, null));
+ commitUser,
+ getMetricGroup(),
+ true,
+ context.isRestored(),
+ context.getOperatorStateStore()));
}
+ }
+
+ @Override
+ public void processElement(StreamRecord<CommittableMessage<MultiTableCommittable>> element) {
if (element.getValue() instanceof CommittableWithLineage) {
multiTableCommittables.add(
((CommittableWithLineage<MultiTableCommittable>) element.getValue())
@@ -109,6 +122,10 @@ public class PreCommitOperator
public void snapshotState(StateSnapshotContext context) throws Exception {
super.snapshotState(context);
long checkpointId = context.getCheckpointId();
+ commitUpToCheckpoint(checkpointId);
+ }
+
+ private void commitUpToCheckpoint(long checkpointId) throws IOException, InterruptedException {
if (!multiTableCommittables.isEmpty()) {
WrappedManifestCommittable wrappedManifestCommittable =
storeMultiCommitter.combine(checkpointId, checkpointId, multiTableCommittables);
@@ -130,4 +147,9 @@ public class PreCommitOperator
storeMultiCommitter.close();
}
}
+
+ @Override
+ public void endInput() throws Exception {
+ commitUpToCheckpoint(Long.MAX_VALUE);
+ }
}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java
index 08b1774e5..f28fd97c2 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java
@@ -19,14 +19,17 @@ package org.apache.flink.cdc.connectors.paimon.sink.v2;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobInfo;
+import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.common.operators.MailboxExecutor;
import org.apache.flink.api.common.operators.ProcessingTimeService;
import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.connector.sink2.Committer;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.cdc.common.configuration.Configuration;
import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
import org.apache.flink.cdc.common.data.binary.BinaryStringData;
import org.apache.flink.cdc.common.event.AddColumnEvent;
@@ -35,27 +38,45 @@ import org.apache.flink.cdc.common.event.DataChangeEvent;
import org.apache.flink.cdc.common.event.DropColumnEvent;
import org.apache.flink.cdc.common.event.DropTableEvent;
import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.SchemaChangeEventTypeFamily;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.event.TruncateTableEvent;
import org.apache.flink.cdc.common.exceptions.SchemaEvolveException;
import org.apache.flink.cdc.common.exceptions.UnsupportedSchemaChangeEventException;
+import org.apache.flink.cdc.common.factories.DataSinkFactory;
+import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.sink.DataSink;
import org.apache.flink.cdc.common.types.DataType;
import org.apache.flink.cdc.common.types.RowType;
+import org.apache.flink.cdc.composer.definition.SinkDef;
+import org.apache.flink.cdc.composer.flink.coordination.OperatorIDGenerator;
+import org.apache.flink.cdc.composer.flink.translator.DataSinkTranslator;
+import org.apache.flink.cdc.composer.flink.translator.OperatorUidGenerator;
+import org.apache.flink.cdc.composer.flink.translator.PartitioningTranslator;
+import org.apache.flink.cdc.composer.flink.translator.SchemaOperatorTranslator;
+import org.apache.flink.cdc.composer.utils.FactoryDiscoveryUtils;
+import org.apache.flink.cdc.connectors.paimon.sink.PaimonDataSinkFactory;
+import org.apache.flink.cdc.connectors.paimon.sink.PaimonDataSinkOptions;
import org.apache.flink.cdc.connectors.paimon.sink.PaimonMetadataApplier;
import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.metrics.groups.InternalSinkCommitterMetricGroup;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.runtime.operators.sink.committables.CommitRequestImpl;
-import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.UserCodeClassLoader;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
+
import org.apache.calcite.sql.validate.SqlValidatorException;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.flink.FlinkCatalogFactory;
@@ -81,7 +102,9 @@ import java.util.Set;
import java.util.UUID;
import java.util.stream.Collectors;
+import static org.apache.flink.cdc.common.pipeline.PipelineOptions.DEFAULT_SCHEMA_OPERATOR_RPC_TIMEOUT;
import static org.apache.flink.cdc.common.types.DataTypes.STRING;
+import static org.apache.flink.configuration.ConfigConstants.DEFAULT_PARALLELISM;
/** An ITCase for {@link PaimonWriter} and {@link PaimonCommitter}. */
public class PaimonSinkITCase {
@@ -92,6 +115,8 @@ public class PaimonSinkITCase {
private TableEnvironment tEnv;
+ private StreamExecutionEnvironment env;
+
private String warehouse;
private final TableId table1 = TableId.tableId("test", "table1");
@@ -116,7 +141,11 @@ public class PaimonSinkITCase {
private void initialize(String metastore)
throws Catalog.DatabaseNotEmptyException, Catalog.DatabaseNotExistException {
- tEnv = TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build());
+ env =
+ StreamExecutionEnvironment.getExecutionEnvironment()
+ .setRuntimeMode(RuntimeExecutionMode.BATCH)
+ .setParallelism(DEFAULT_PARALLELISM);
+ tEnv = StreamTableEnvironment.create(env);
warehouse = new File(temporaryFolder.toFile(), UUID.randomUUID().toString()).toString();
catalogOptions = new Options();
catalogOptions.setString("metastore", metastore);
@@ -580,6 +609,110 @@ public class PaimonSinkITCase {
Row.ofKind(RowKind.INSERT, "8", "8"));
}
+ @ParameterizedTest
+ @CsvSource({"hive"})
+ public void testSinkWithHiveMetastorePartitionedTable(String metastore) throws Exception {
+ initialize(metastore);
+ List<Event> testEvents = new ArrayList<>();
+ Schema.Builder builder = Schema.newBuilder();
+ builder.option("metastore.partitioned-table", "true");
+ // create table
+ Schema schema =
+ builder.physicalColumn("pt", STRING())
+ .physicalColumn("pk", STRING())
+ .physicalColumn("name", STRING())
+ .primaryKey("pk")
+ .partitionKey("pt")
+ .build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(table1, schema);
+ testEvents.add(createTableEvent);
+ testEvents.add(
+ generateInsert(
+ table1,
+ Arrays.asList(
+ Tuple2.of(STRING(), "20250604"),
+ Tuple2.of(STRING(), "1"),
+ Tuple2.of(STRING(), "Alice"))));
+ testEvents.add(
+ generateInsert(
+ table1,
+ Arrays.asList(
+ Tuple2.of(STRING(), "20250604"),
+ Tuple2.of(STRING(), "2"),
+ Tuple2.of(STRING(), "Bob"))));
+
+ runJobWithEvents(testEvents, true);
+
+ Assertions.assertThat(fetchResults(table1))
+ .containsExactlyInAnyOrder(
+ Row.ofKind(RowKind.INSERT, "20250604", "1", "Alice"),
+ Row.ofKind(RowKind.INSERT, "20250604", "2", "Bob"));
+ }
+
+ private void runJobWithEvents(List<Event> events, boolean isBatchMode) throws Exception {
+ DataStream<Event> stream = env.fromCollection(events, TypeInformation.of(Event.class));
+
+ DataSinkFactory sinkFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("paimon", DataSinkFactory.class);
+ Assertions.assertThat(sinkFactory).isInstanceOf(PaimonDataSinkFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.<String, String>builder()
+ .put(PaimonDataSinkOptions.METASTORE.key(), "hive")
+ .put(PaimonDataSinkOptions.WAREHOUSE.key(), warehouse)
+ .put("catalog.properties.cache-enabled", "false")
+ .build());
+
+ DataSink paimonSink =
+ sinkFactory.createDataSink(
+ new FactoryHelper.DefaultContext(
+ conf, conf, Thread.currentThread().getContextClassLoader()));
+
+ PartitioningTranslator partitioningTranslator = new PartitioningTranslator();
+
+ SchemaOperatorTranslator schemaOperatorTranslator =
+ new SchemaOperatorTranslator(
+ SchemaChangeBehavior.EVOLVE,
+ "$$_schema_operator_$$",
+ DEFAULT_SCHEMA_OPERATOR_RPC_TIMEOUT,
+ "UTC");
+
+ stream =
+ schemaOperatorTranslator.translateRegular(
+ stream,
+ DEFAULT_PARALLELISM,
+ isBatchMode,
+ paimonSink
+ .getMetadataApplier()
+ .setAcceptedSchemaEvolutionTypes(
+ Arrays.stream(SchemaChangeEventTypeFamily.ALL)
+ .collect(Collectors.toSet())),
+ new ArrayList<>());
+ OperatorIDGenerator schemaOperatorIDGenerator =
+ new OperatorIDGenerator(schemaOperatorTranslator.getSchemaOperatorUid());
+
+ stream =
+ partitioningTranslator.translateRegular(
+ stream,
+ DEFAULT_PARALLELISM,
+ DEFAULT_PARALLELISM,
+ isBatchMode,
+ schemaOperatorIDGenerator.generate(),
+ paimonSink.getDataChangeEventHashFunctionProvider(DEFAULT_PARALLELISM),
+ new OperatorUidGenerator());
+
+ DataSinkTranslator sinkTranslator = new DataSinkTranslator();
+ sinkTranslator.translate(
+ new SinkDef("paimon", "Paimon Sink", conf),
+ stream,
+ paimonSink,
+ isBatchMode,
+ schemaOperatorIDGenerator.generate(),
+ new OperatorUidGenerator());
+ env.execute("runJobWithEvents").getJobExecutionResult();
+ }
+
private static MultiTableCommittable correctCheckpointId(MultiTableCommittable committable) {
// update the right checkpointId for MultiTableCommittable
return new MultiTableCommittable(