This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 92c22603b2 [flink][cdc] Fix bucket-aware state recovery for
multi-table CDC sink (#9417)
92c22603b2 is described below
commit 92c22603b217923652a5f6d126bcd19548d78173
Author: Wenchao Wu <[email protected]>
AuthorDate: Fri Aug 28 16:22:48 2026 +0800
[flink][cdc] Fix bucket-aware state recovery for multi-table CDC sink
(#9417)
---
.../cdc/CdcMultiplexRecordChannelComputer.java | 108 ++++++++---
.../sink/cdc/CdcRecordStoreMultiWriteOperator.java | 85 +++++++-
.../flink/sink/cdc/FlinkCdcMultiTableSink.java | 89 ++++++++-
.../sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java | 10 +-
.../cdc/CdcMultiplexRecordChannelComputerTest.java | 85 +++++++-
.../cdc/CdcRecordStoreMultiWriteOperatorTest.java | 215 +++++++++++++++++++--
.../flink/sink/cdc/FlinkCdcMultiTableSinkTest.java | 66 ++++++-
7 files changed, 579 insertions(+), 79 deletions(-)
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java
index 2858b2d4eb..053108f1cf 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputer.java
@@ -21,13 +21,11 @@ package org.apache.paimon.flink.sink.cdc;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.ChannelComputer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
@@ -35,10 +33,11 @@ import java.util.Objects;
/** {@link ChannelComputer} for {@link CdcMultiplexRecord}. */
public class CdcMultiplexRecordChannelComputer implements
ChannelComputer<CdcMultiplexRecord> {
- private static final Logger LOG =
- LoggerFactory.getLogger(CdcMultiplexRecordChannelComputer.class);
-
private static final long serialVersionUID = 1L;
+
+ private static final int TABLE_LOOKUP_MAX_RETRIES = 10;
+ private static final long TABLE_LOOKUP_RETRY_INTERVAL_MILLIS = 500L;
+
private final CatalogLoader catalogLoader;
private transient int numChannels;
@@ -58,42 +57,95 @@ public class CdcMultiplexRecordChannelComputer implements
ChannelComputer<CdcMul
@Override
public int channel(CdcMultiplexRecord multiplexRecord) {
ChannelComputer<CdcRecord> channelComputer =
computeChannelComputer(multiplexRecord);
- int recordChannel =
- channelComputer != null ?
channelComputer.channel(multiplexRecord.record()) : 0;
- return Math.floorMod(
- Objects.hash(multiplexRecord.databaseName(),
multiplexRecord.tableName())
- + recordChannel,
+ int recordChannel = channelComputer.channel(multiplexRecord.record());
+ return mixTableIntoChannel(
+ multiplexRecord.databaseName(),
+ multiplexRecord.tableName(),
+ recordChannel,
+ numChannels);
+ }
+
+ /**
+ * Computes the channel a given bucket is routed to, without needing a
record. This mirrors
+ * {@link #channel}, so that {@link CdcRecordStoreMultiWriteOperator} can
decide which subtask
+ * owns the state of a bucket.
+ */
+ static int computeChannel(
+ String databaseName,
+ String tableName,
+ BinaryRow partition,
+ int bucket,
+ int numChannels) {
+ return mixTableIntoChannel(
+ databaseName,
+ tableName,
+ ChannelComputer.select(partition, bucket, numChannels),
numChannels);
}
+ /** Offsets the per-table channel by the table identity, so that tables
are spread out. */
+ private static int mixTableIntoChannel(
+ String databaseName, String tableName, int recordChannel, int
numChannels) {
+ return Math.floorMod(Objects.hash(databaseName, tableName) +
recordChannel, numChannels);
+ }
+
private ChannelComputer<CdcRecord>
computeChannelComputer(CdcMultiplexRecord record) {
return channelComputers.computeIfAbsent(
Identifier.create(record.databaseName(), record.tableName()),
id -> {
- FileStoreTable table;
try (Catalog catalog = catalogLoader.load()) {
- table = (FileStoreTable) catalog.getTable(id);
- } catch (Catalog.TableNotExistException e) {
- LOG.error("Failed to get table {}", id.getFullName(),
e);
- return null;
+ FileStoreTable table = getTable(catalog, id);
+ if (table.bucketMode() != BucketMode.HASH_FIXED) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Combine mode Sink only supports
FIXED bucket mode, but %s is %s",
+ table.name(), table.bucketMode()));
+ }
+
+ CdcRecordChannelComputer channelComputer =
+ new CdcRecordChannelComputer(table.schema());
+ channelComputer.setup(numChannels);
+ return channelComputer;
+ } catch (RuntimeException e) {
+ throw e;
} catch (Exception e) {
throw new RuntimeException(e);
}
-
- if (table.bucketMode() != BucketMode.HASH_FIXED) {
- throw new UnsupportedOperationException(
- String.format(
- "Combine mode Sink only supports FIXED
bucket mode, but %s is %s",
- table.name(), table.bucketMode()));
- }
-
- CdcRecordChannelComputer channelComputer =
- new CdcRecordChannelComputer(table.schema());
- channelComputer.setup(numChannels);
- return channelComputer;
});
}
+ private FileStoreTable getTable(Catalog catalog, Identifier tableId) {
+ Catalog.TableNotExistException lastException = null;
+ for (int retry = 0; retry <= TABLE_LOOKUP_MAX_RETRIES; retry++) {
+ try {
+ return (FileStoreTable) catalog.getTable(tableId);
+ } catch (Catalog.TableNotExistException e) {
+ lastException = e;
+ // Records of a newly added table can arrive before the table
is visible here. Do
+ // not use a temporary channel: the writer-state restore
filter must calculate the
+ // exact same owner from the real partition and bucket.
+ if (retry == TABLE_LOOKUP_MAX_RETRIES) {
+ break;
+ }
+ try {
+ Thread.sleep(TABLE_LOOKUP_RETRY_INTERVAL_MILLIS);
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(
+ "Interrupted while waiting for table " +
tableId.getFullName(),
+ interrupted);
+ }
+ }
+ }
+ throw new RuntimeException(
+ String.format(
+ "Table %s is still unavailable after %s retries (%s ms
total wait).",
+ tableId.getFullName(),
+ TABLE_LOOKUP_MAX_RETRIES,
+ TABLE_LOOKUP_MAX_RETRIES *
TABLE_LOOKUP_RETRY_INTERVAL_MILLIS),
+ lastException);
+ }
+
@Override
public String toString() {
return "shuffle by bucket";
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java
index da612bd337..20dd9847c4 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperator.java
@@ -35,6 +35,7 @@ import org.apache.paimon.options.Options;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.ExecutorThreadFactory;
+import org.apache.paimon.utils.Preconditions;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
@@ -43,6 +44,8 @@ import
org.apache.flink.streaming.api.operators.StreamOperatorFactory;
import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedList;
@@ -62,6 +65,17 @@ import static
org.apache.paimon.flink.sink.cdc.CdcRecordUtils.toGenericRow;
/**
* A {@link PrepareCommitOperator} to write {@link CdcRecord}. Record schema
may change. If current
* known schema does not fit record schema, this operator will wait for schema
changes.
+ *
+ * <p>When {@code stateDatabaseName} is given, this operator assumes its input
is partitioned by
+ * {@link CdcMultiplexRecordChannelComputer} and that every incoming record
belongs to that
+ * database. {@link FlinkCdcMultiTableSink} guarantees both by applying the
partitioner itself. The
+ * {@link StoreSinkWriteState} filter below then distributes state values
among subtasks with
+ * exactly the same formula as the channel computer, so a record and the state
of the bucket it
+ * belongs to always end up in the same subtask.
+ *
+ * <p>The compatibility constructor does not have a database name and
therefore keeps the legacy
+ * behavior of restoring all union state values into every subtask. It does
not guarantee unique
+ * bucket-state ownership after a restore.
*/
public class CdcRecordStoreMultiWriteOperator
extends PrepareCommitOperator<CdcMultiplexRecord,
MultiTableCommittable> {
@@ -71,6 +85,7 @@ public class CdcRecordStoreMultiWriteOperator
private final StoreSinkWrite.Provider storeSinkWriteProvider;
private final String initialCommitUser;
private final CatalogLoader catalogLoader;
+ @Nullable private final String stateDatabaseName;
private Catalog catalog;
private Map<Identifier, FileStoreTable> tables;
@@ -84,11 +99,13 @@ public class CdcRecordStoreMultiWriteOperator
CatalogLoader catalogLoader,
StoreSinkWrite.Provider storeSinkWriteProvider,
String initialCommitUser,
+ @Nullable String stateDatabaseName,
Options options) {
super(parameters, options);
this.catalogLoader = catalogLoader;
this.storeSinkWriteProvider = storeSinkWriteProvider;
this.initialCommitUser = initialCommitUser;
+ this.stateDatabaseName = stateDatabaseName;
}
@Override
@@ -104,12 +121,28 @@ public class CdcRecordStoreMultiWriteOperator
StateUtils.getSingleValueFromState(
context, "commit_user_state", String.class,
initialCommitUser);
- // TODO: should use CdcRecordMultiChannelComputer to filter
- state =
- new StoreSinkWriteStateImpl(
-
RuntimeContextUtils.getIndexOfThisSubtask(getRuntimeContext()),
- context,
- (tableName, partition, bucket) -> true);
+ int numTasks =
RuntimeContextUtils.getNumberOfParallelSubtasks(getRuntimeContext());
+ int subtaskId =
RuntimeContextUtils.getIndexOfThisSubtask(getRuntimeContext());
+ StoreSinkWriteState.StateValueFilter stateFilter;
+ if (stateDatabaseName == null) {
+ // Preserve the behavior of the old constructor for compatibility.
Without a database
+ // name, every subtask restores all union state and bucket
ownership is not guaranteed.
+ stateFilter = (tableName, partition, bucket) -> true;
+ } else {
+ // Keep this filter in sync with
CdcMultiplexRecordChannelComputer, which partitions the
+ // input of this operator. Otherwise state values would be
restored into a subtask which
+ // never writes the corresponding bucket.
+ stateFilter =
+ (tableName, partition, bucket) ->
+ subtaskId
+ ==
CdcMultiplexRecordChannelComputer.computeChannel(
+ stateDatabaseName,
+ tableName,
+ partition,
+ bucket,
+ numTasks);
+ }
+ state = new StoreSinkWriteStateImpl(subtaskId, context, stateFilter);
tables = new HashMap<>();
writes = new HashMap<>();
compactExecutor =
@@ -123,6 +156,11 @@ public class CdcRecordStoreMultiWriteOperator
CdcMultiplexRecord record = element.getValue();
String databaseName = record.databaseName();
+ Preconditions.checkArgument(
+ stateDatabaseName == null ||
stateDatabaseName.equals(databaseName),
+ "This writer only accepts records from database %s, but
received a record from %s.",
+ stateDatabaseName,
+ databaseName);
String tableName = record.tableName();
Identifier tableId = Identifier.create(databaseName, tableName);
@@ -226,8 +264,12 @@ public class CdcRecordStoreMultiWriteOperator
@Override
public void close() throws Exception {
super.close();
- for (StoreSinkWrite write : writes.values()) {
- write.close();
+ // initializeState may have failed before these were assigned, and
Flink still closes the
+ // operator. Do not mask the original failure with a
NullPointerException.
+ if (writes != null) {
+ for (StoreSinkWrite write : writes.values()) {
+ write.close();
+ }
}
if (compactExecutor != null) {
compactExecutor.shutdownNow();
@@ -275,22 +317,48 @@ public class CdcRecordStoreMultiWriteOperator
return commitUser;
}
+ @VisibleForTesting
+ public StoreSinkWriteState state() {
+ return state;
+ }
+
/** {@link StreamOperatorFactory} of {@link
CdcRecordStoreMultiWriteOperator}. */
public static class Factory
extends PrepareCommitOperator.Factory<CdcMultiplexRecord,
MultiTableCommittable> {
private final StoreSinkWrite.Provider storeSinkWriteProvider;
private final String initialCommitUser;
private final CatalogLoader catalogLoader;
+ @Nullable private final String stateDatabaseName;
+
+ /**
+ * @deprecated Use {@link #Factory(CatalogLoader,
StoreSinkWrite.Provider, String, String,
+ * Options)} instead. Without a database name, every subtask
restores all union writer
+ * state and unique bucket-state ownership is not guaranteed.
+ */
+ @Deprecated
+ public Factory(
+ CatalogLoader catalogLoader,
+ StoreSinkWrite.Provider storeSinkWriteProvider,
+ String initialCommitUser,
+ Options options) {
+ super(options);
+ this.catalogLoader = catalogLoader;
+ this.storeSinkWriteProvider = storeSinkWriteProvider;
+ this.initialCommitUser = initialCommitUser;
+ this.stateDatabaseName = null;
+ }
public Factory(
CatalogLoader catalogLoader,
StoreSinkWrite.Provider storeSinkWriteProvider,
String initialCommitUser,
+ String stateDatabaseName,
Options options) {
super(options);
this.catalogLoader = catalogLoader;
this.storeSinkWriteProvider = storeSinkWriteProvider;
this.initialCommitUser = initialCommitUser;
+ this.stateDatabaseName =
Preconditions.checkNotNull(stateDatabaseName);
}
@Override
@@ -303,6 +371,7 @@ public class CdcRecordStoreMultiWriteOperator
catalogLoader,
storeSinkWriteProvider,
initialCommitUser,
+ stateDatabaseName,
options);
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
index cdf619725d..b560920a91 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
@@ -36,6 +36,7 @@ import
org.apache.paimon.flink.sink.WrappedManifestCommittableSerializer;
import org.apache.paimon.manifest.WrappedManifestCommittable;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
+import org.apache.paimon.utils.Preconditions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
@@ -55,6 +56,11 @@ import static
org.apache.paimon.flink.utils.ParallelismUtils.forwardParallelism;
/**
* A {@link FlinkSink} which accepts {@link CdcRecord} and waits for a schema
change if necessary.
+ *
+ * <p>When created with a database name, this sink partitions its input with
{@link
+ * CdcMultiplexRecordChannelComputer} itself, so that records and the writer
states of their buckets
+ * are routed to the same subtask. The compatibility constructor preserves the
legacy behavior and
+ * expects its input to have already been partitioned.
*/
public class FlinkCdcMultiTableSink implements Serializable {
@@ -69,9 +75,17 @@ public class FlinkCdcMultiTableSink implements Serializable {
private final double commitCpuCores;
@Nullable private final MemorySize commitHeapMemory;
private final String commitUser;
+ @Nullable private final String databaseName;
private boolean eagerInit = false;
private TableFilter tableFilter;
+ /**
+ * @deprecated Use {@link #FlinkCdcMultiTableSink(CatalogLoader, String,
double, MemorySize,
+ * double, MemorySize, String, boolean, TableFilter)} instead. Without
a database name,
+ * every subtask restores all union writer state, unique bucket-state
ownership is not
+ * guaranteed after a restore, and the input must be partitioned by
the caller.
+ */
+ @Deprecated
public FlinkCdcMultiTableSink(
CatalogLoader catalogLoader,
double writeCpuCores,
@@ -82,6 +96,28 @@ public class FlinkCdcMultiTableSink implements Serializable {
boolean eagerInit,
TableFilter tableFilter) {
this.catalogLoader = catalogLoader;
+ this.databaseName = null;
+ this.writeCpuCores = writeCpuCores;
+ this.writeHeapMemory = writeHeapMemory;
+ this.commitCpuCores = commitCpuCores;
+ this.commitHeapMemory = commitHeapMemory;
+ this.commitUser = commitUser;
+ this.eagerInit = eagerInit;
+ this.tableFilter = tableFilter;
+ }
+
+ public FlinkCdcMultiTableSink(
+ CatalogLoader catalogLoader,
+ String databaseName,
+ double writeCpuCores,
+ @Nullable MemorySize writeHeapMemory,
+ double commitCpuCores,
+ @Nullable MemorySize commitHeapMemory,
+ String commitUser,
+ boolean eagerInit,
+ TableFilter tableFilter) {
+ this.catalogLoader = catalogLoader;
+ this.databaseName = Preconditions.checkNotNull(databaseName);
this.writeCpuCores = writeCpuCores;
this.writeHeapMemory = writeHeapMemory;
this.commitCpuCores = commitCpuCores;
@@ -107,25 +143,58 @@ public class FlinkCdcMultiTableSink implements
Serializable {
}
public DataStreamSink<?> sinkFrom(DataStream<CdcMultiplexRecord> input) {
+ return sinkFrom(input, null);
+ }
+
+ /**
+ * @param parallelism parallelism of the writer and committer operators,
or null to forward the
+ * parallelism of {@code input}.
+ */
+ public DataStreamSink<?> sinkFrom(
+ DataStream<CdcMultiplexRecord> input, @Nullable Integer
parallelism) {
// This commitUser is valid only for new jobs.
// After the job starts, this commitUser will be recorded into the
states of write and
// commit operators.
// When the job restarts, commitUser will be recovered from states and
this value is
// ignored.
- return sinkFrom(input, commitUser, createWriteProvider());
+ return sinkFrom(input, parallelism, commitUser, createWriteProvider());
+ }
+
+ public DataStreamSink<?> sinkFrom(
+ DataStream<CdcMultiplexRecord> input,
+ String commitUser,
+ StoreSinkWrite.Provider sinkProvider) {
+ return sinkFrom(input, null, commitUser, sinkProvider);
}
public DataStreamSink<?> sinkFrom(
DataStream<CdcMultiplexRecord> input,
+ @Nullable Integer parallelism,
String commitUser,
StoreSinkWrite.Provider sinkProvider) {
StreamExecutionEnvironment env = input.getExecutionEnvironment();
assertStreamingConfiguration(env);
+ Preconditions.checkArgument(
+ databaseName != null || parallelism == null,
+ "Explicit parallelism is only supported by the constructor
which takes a "
+ + "database name.");
+
+ // Keep the old constructor's topology unchanged for compatibility.
The database-aware
+ // constructor can shuffle by bucket itself and use the same formula
to redistribute writer
+ // state on restore.
+ DataStream<CdcMultiplexRecord> shuffled =
+ databaseName == null
+ ? input
+ : FlinkStreamPartitioner.partition(
+ input,
+ new
CdcMultiplexRecordChannelComputer(catalogLoader),
+ parallelism);
+
MultiTableCommittableTypeInfo typeInfo = new
MultiTableCommittableTypeInfo();
SingleOutputStreamOperator<MultiTableCommittable> written =
- input.transform(
+ shuffled.transform(
WRITER_NAME, typeInfo,
createWriteOperator(sinkProvider, commitUser));
- forwardParallelism(written, input);
+ forwardParallelism(written, shuffled);
configureSlotSharingGroup(written, writeCpuCores, writeHeapMemory);
// shuffle committables by table
@@ -133,7 +202,7 @@ public class FlinkCdcMultiTableSink implements Serializable
{
FlinkStreamPartitioner.partition(
written,
new MultiTableCommittableChannelComputer(),
- input.getParallelism());
+ shuffled.getParallelism());
SingleOutputStreamOperator<?> committed =
partitioned.transform(
@@ -145,13 +214,23 @@ public class FlinkCdcMultiTableSink implements
Serializable {
commitUser,
createCommitterFactory(tableFilter),
createCommittableStateManager()));
- forwardParallelism(committed, input);
+ forwardParallelism(committed, shuffled);
configureSlotSharingGroup(committed, commitCpuCores, commitHeapMemory);
return committed.sinkTo(new
DiscardingSink<>()).name("end").setParallelism(1);
}
protected OneInputStreamOperatorFactory<CdcMultiplexRecord,
MultiTableCommittable>
createWriteOperator(StoreSinkWrite.Provider writeProvider, String
commitUser) {
+ return databaseName == null
+ ? createCompatibilityWriteOperator(writeProvider, commitUser)
+ : new CdcRecordStoreMultiWriteOperator.Factory(
+ catalogLoader, writeProvider, commitUser,
databaseName, new Options());
+ }
+
+ @SuppressWarnings("deprecation")
+ private OneInputStreamOperatorFactory<CdcMultiplexRecord,
MultiTableCommittable>
+ createCompatibilityWriteOperator(
+ StoreSinkWrite.Provider writeProvider, String commitUser) {
return new CdcRecordStoreMultiWriteOperator.Factory(
catalogLoader, writeProvider, commitUser, new Options());
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
index f5cd33f019..592d7aee14 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
@@ -185,15 +185,10 @@ public class FlinkCdcSyncDatabaseSinkBuilder<T> {
DataStream<CdcMultiplexRecord> converted =
CaseSensitiveUtils.cdcMultiplexRecordConvert(catalogLoader,
newlyAddedTableStream);
- DataStream<CdcMultiplexRecord> partitioned =
- partition(
- converted,
- new CdcMultiplexRecordChannelComputer(catalogLoader),
- parallelism);
-
FlinkCdcMultiTableSink sink =
new FlinkCdcMultiTableSink(
catalogLoader,
+ database,
writerCpu,
writerMemory,
committerCpu,
@@ -201,7 +196,8 @@ public class FlinkCdcSyncDatabaseSinkBuilder<T> {
commitUser,
eagerInit,
tableFilter);
- sink.sinkFrom(partitioned);
+ // the sink shuffles by bucket itself
+ sink.sinkFrom(converted, parallelism);
}
private void buildForFixedBucket(FileStoreTable table,
DataStream<CdcRecord> parsed) {
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputerTest.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputerTest.java
index 43b7d2ba63..5ad0ef9357 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputerTest.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcMultiplexRecordChannelComputerTest.java
@@ -28,6 +28,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
@@ -47,7 +48,9 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
@@ -59,6 +62,7 @@ public class CdcMultiplexRecordChannelComputerTest {
private Path warehouse;
private String databaseName;
private Identifier tableWithPartition;
+ private Schema tableWithPartitionSchema;
private Catalog catalog;
private Identifier tableWithoutPartition;
@@ -92,16 +96,16 @@ public class CdcMultiplexRecordChannelComputerTest {
},
new String[] {"k", "v"});
+ tableWithPartitionSchema =
+ new Schema(
+ rowTypeWithPartition.getFields(),
+ Collections.singletonList("pt"),
+ Arrays.asList("pt", "k"),
+ conf.toMap(),
+ "");
List<Tuple2<Identifier, Schema>> tables =
Arrays.asList(
- Tuple2.of(
- tableWithPartition,
- new Schema(
- rowTypeWithPartition.getFields(),
- Collections.singletonList("pt"),
- Arrays.asList("pt", "k"),
- conf.toMap(),
- "")),
+ Tuple2.of(tableWithPartition,
tableWithPartitionSchema),
Tuple2.of(
tableWithoutPartition,
new Schema(
@@ -140,7 +144,7 @@ public class CdcMultiplexRecordChannelComputerTest {
}
@Test
- public void testSchemaNoPartition() {
+ public void testSchemaNoPartition() throws Exception {
ThreadLocalRandom random = ThreadLocalRandom.current();
int numInputs = random.nextInt(1000) + 1;
List<Map<String, String>> input = new ArrayList<>();
@@ -154,13 +158,59 @@ public class CdcMultiplexRecordChannelComputerTest {
testImpl(tableWithoutPartition, input);
}
- private void testImpl(Identifier tableId, List<Map<String, String>> input)
{
+ @Test
+ public void testWaitForTableBeforeComputingChannel() throws Exception {
+ int numChannels = 3;
+ Map<String, String> data = new HashMap<>();
+ data.put("pt", "1");
+ data.put("k", "2");
+ data.put("v", "3");
+ CdcRecord record = new CdcRecord(RowKind.INSERT, data);
+
+ FileStoreTable table = (FileStoreTable)
catalog.getTable(tableWithPartition);
+ CdcRecordKeyAndBucketExtractor extractor =
+ new CdcRecordKeyAndBucketExtractor(table.schema());
+ extractor.setRecord(record);
+ int expectedChannel =
+ CdcMultiplexRecordChannelComputer.computeChannel(
+ databaseName,
+ tableWithPartition.getObjectName(),
+ extractor.partition(),
+ extractor.bucket(),
+ numChannels);
+
+ catalog.dropTable(tableWithPartition, false);
+ CdcMultiplexRecordChannelComputer channelComputer =
+ new CdcMultiplexRecordChannelComputer(catalogLoader);
+ channelComputer.setup(numChannels);
+ CompletableFuture<Integer> channelFuture =
+ CompletableFuture.supplyAsync(
+ () ->
+ channelComputer.channel(
+ CdcMultiplexRecord.fromCdcRecord(
+ databaseName,
+
tableWithPartition.getObjectName(),
+ record)));
+
+ try {
+ Thread.sleep(50);
+ assertThat(channelFuture.isDone()).isFalse();
+ } finally {
+ catalog.createTable(tableWithPartition, tableWithPartitionSchema,
false);
+ }
+ assertThat(channelFuture.get(5,
TimeUnit.SECONDS)).isEqualTo(expectedChannel);
+ }
+
+ private void testImpl(Identifier tableId, List<Map<String, String>> input)
throws Exception {
ThreadLocalRandom random = ThreadLocalRandom.current();
int numChannels = random.nextInt(10) + 1;
CdcMultiplexRecordChannelComputer channelComputer =
new CdcMultiplexRecordChannelComputer(catalogLoader);
channelComputer.setup(numChannels);
+ FileStoreTable table = (FileStoreTable) catalog.getTable(tableId);
+ CdcRecordKeyAndBucketExtractor extractor =
+ new CdcRecordKeyAndBucketExtractor(table.schema());
// assert that insert and delete records are routed into same channel
@@ -168,6 +218,21 @@ public class CdcMultiplexRecordChannelComputerTest {
CdcRecord insertRecord = new CdcRecord(RowKind.INSERT, data);
CdcRecord deleteRecord = new CdcRecord(RowKind.DELETE, data);
+ extractor.setRecord(insertRecord);
+ assertThat(
+ channelComputer.channel(
+ CdcMultiplexRecord.fromCdcRecord(
+ tableId.getDatabaseName(),
+ tableId.getObjectName(),
+ insertRecord)))
+ .isEqualTo(
+ CdcMultiplexRecordChannelComputer.computeChannel(
+ tableId.getDatabaseName(),
+ tableId.getObjectName(),
+ extractor.partition(),
+ extractor.bucket(),
+ numChannels));
+
assertThat(
channelComputer.channel(
CdcMultiplexRecord.fromCdcRecord(
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperatorTest.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperatorTest.java
index ee64162ad9..6d0a0d7c8a 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperatorTest.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/CdcRecordStoreMultiWriteOperatorTest.java
@@ -24,10 +24,12 @@ import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.flink.sink.MultiTableCommittable;
import org.apache.paimon.flink.sink.MultiTableCommittableTypeInfo;
import org.apache.paimon.flink.sink.StoreSinkWrite;
import org.apache.paimon.flink.sink.StoreSinkWriteImpl;
+import org.apache.paimon.flink.sink.StoreSinkWriteState;
import org.apache.paimon.fs.Path;
import org.apache.paimon.operation.AbstractFileStoreWrite;
import org.apache.paimon.options.CatalogOptions;
@@ -48,6 +50,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.state.JavaSerializer;
+import org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -71,10 +74,13 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link CdcRecordStoreMultiWriteOperator}. */
public class CdcRecordStoreMultiWriteOperatorTest {
+ private static final String STATE_NAME = "paimon_test_state";
+
@TempDir java.nio.file.Path tempDir;
private String commitUser;
@@ -689,33 +695,202 @@ public class CdcRecordStoreMultiWriteOperatorTest {
harness.close();
}
+ @Test
+ @Timeout(30)
+ public void testWriterStateIsPartitionedOnRestore() throws Exception {
+ // Write a state value for every bucket of both tables, from a single
subtask.
+ int numBuckets = 4;
+ List<StoreSinkWriteState.StateValue> stateValues = new ArrayList<>();
+ for (int bucket = 0; bucket < numBuckets; bucket++) {
+ stateValues.add(
+ new StoreSinkWriteState.StateValue(
+ BinaryRow.EMPTY_ROW, bucket, new byte[] {(byte)
bucket}));
+ }
+
+ OneInputStreamOperatorTestHarness<CdcMultiplexRecord,
MultiTableCommittable> harness =
+ createTestHarness(catalogLoader, 1, 0);
+ harness.open();
+ CdcRecordStoreMultiWriteOperator operator =
+ (CdcRecordStoreMultiWriteOperator) harness.getOperator();
+ operator.state().put(firstTable.getObjectName(), STATE_NAME,
stateValues);
+ operator.state().put(secondTable.getObjectName(), STATE_NAME,
stateValues);
+ OperatorSubtaskState snapshot = harness.snapshot(0, 1);
+ harness.close();
+
+ // Restore with a scaled up parallelism. Every state value must be
restored into exactly
+ // one subtask, and that subtask must be the one the channel computer
routes the
+ // corresponding bucket to -- otherwise the subtask owning the bucket
would silently lose
+ // its state.
+ int numTasks = 3;
+ for (String tableName :
+ Arrays.asList(firstTable.getObjectName(),
secondTable.getObjectName())) {
+ for (int bucket = 0; bucket < numBuckets; bucket++) {
+ int currentBucket = bucket;
+ List<Integer> owners = new ArrayList<>();
+ for (int subtaskId = 0; subtaskId < numTasks; subtaskId++) {
+ OneInputStreamOperatorTestHarness<CdcMultiplexRecord,
MultiTableCommittable>
+ restored = createTestHarness(catalogLoader,
numTasks, subtaskId);
+ restored.initializeState(snapshot);
+ restored.open();
+ List<StoreSinkWriteState.StateValue> restoredValues =
+ ((CdcRecordStoreMultiWriteOperator)
restored.getOperator())
+ .state()
+ .get(tableName, STATE_NAME);
+ if (restoredValues != null
+ && restoredValues.stream().anyMatch(v ->
v.bucket() == currentBucket)) {
+ owners.add(subtaskId);
+ }
+ restored.close();
+ }
+
+ int expected =
+ CdcMultiplexRecordChannelComputer.computeChannel(
+ databaseName, tableName, BinaryRow.EMPTY_ROW,
bucket, numTasks);
+ assertThat(owners)
+ .as(
+ "state of %s bucket %s must be owned by
exactly one subtask",
+ tableName, bucket)
+ .containsExactly(expected);
+ }
+ }
+ }
+
+ @Test
+ public void testRejectRecordFromDifferentDatabase() throws Exception {
+ OneInputStreamOperatorTestHarness<CdcMultiplexRecord,
MultiTableCommittable> harness =
+ createTestHarness(catalogLoader);
+ harness.open();
+
+ Map<String, String> data = new HashMap<>();
+ data.put("pt", "0");
+ data.put("k", "1");
+ data.put("v", "10");
+ CdcMultiplexRecord record =
+ CdcMultiplexRecord.fromCdcRecord(
+ "another_database",
+ firstTable.getObjectName(),
+ new CdcRecord(RowKind.INSERT, data));
+
+ assertThatThrownBy(() -> harness.processElement(record, 1))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("only accepts records from database " +
databaseName)
+ .hasMessageContaining("another_database");
+ harness.close();
+ }
+
+ @Test
+ @Timeout(30)
+ public void testRestoreWithoutDatabaseName() throws Exception {
+ // Preserve the exact legacy union-state behavior: every restored
subtask receives every
+ // state value. This is compatible with old savepoints, but does not
provide unique bucket
+ // ownership for stateful writers.
+ int numTasks = 2;
+ List<OperatorSubtaskState> snapshots = new ArrayList<>();
+ for (int subtaskId = 0; subtaskId < numTasks; subtaskId++) {
+ OneInputStreamOperatorTestHarness<CdcMultiplexRecord,
MultiTableCommittable> harness =
+ createTestHarness(
+ createOperatorFactoryWithoutDatabaseName(),
numTasks, subtaskId);
+ harness.open();
+ CdcRecordStoreMultiWriteOperator operator =
+ (CdcRecordStoreMultiWriteOperator) harness.getOperator();
+ StoreSinkWriteState.StateValue stateValue =
+ new StoreSinkWriteState.StateValue(
+ BinaryRow.EMPTY_ROW, subtaskId, new byte[] {(byte)
subtaskId});
+ operator.state()
+ .put(
+ firstTable.getObjectName(),
+ STATE_NAME,
+ Collections.singletonList(stateValue));
+ snapshots.add(harness.snapshot(0, 1));
+ harness.close();
+ }
+
+ OperatorSubtaskState unionState =
+ AbstractStreamOperatorTestHarness.repackageState(
+ snapshots.toArray(new OperatorSubtaskState[0]));
+ for (int subtaskId = 0; subtaskId < numTasks; subtaskId++) {
+ OneInputStreamOperatorTestHarness<CdcMultiplexRecord,
MultiTableCommittable> restored =
+ createTestHarness(
+ createOperatorFactoryWithoutDatabaseName(),
numTasks, subtaskId);
+ restored.initializeState(unionState);
+ restored.open();
+ List<StoreSinkWriteState.StateValue> restoredValues =
+ ((CdcRecordStoreMultiWriteOperator) restored.getOperator())
+ .state()
+ .get(firstTable.getObjectName(), STATE_NAME);
+ assertThat(restoredValues)
+ .extracting(StoreSinkWriteState.StateValue::bucket)
+ .containsExactlyInAnyOrder(0, 1);
+ restored.close();
+ }
+ }
+
private OneInputStreamOperatorTestHarness<CdcMultiplexRecord,
MultiTableCommittable>
createTestHarness(CatalogLoader catalogLoader) throws Exception {
- CdcRecordStoreMultiWriteOperator.Factory operatorFactory =
- new CdcRecordStoreMultiWriteOperator.Factory(
- catalogLoader,
- (t, commitUser, state, ioManager, memoryPoolFactory,
metricGroup) ->
- new StoreSinkWriteImpl(
- t,
- commitUser,
- state,
- ioManager,
- false,
- false,
- true,
- memoryPoolFactory,
- metricGroup),
- commitUser,
- Options.fromMap(new HashMap<>()));
TypeSerializer<CdcMultiplexRecord> inputSerializer = new
JavaSerializer<>();
- TypeSerializer<MultiTableCommittable> outputSerializer =
- new MultiTableCommittableTypeInfo().createSerializer(new
ExecutionConfig());
OneInputStreamOperatorTestHarness<CdcMultiplexRecord,
MultiTableCommittable> harness =
- new OneInputStreamOperatorTestHarness<>(operatorFactory,
inputSerializer);
- harness.setup(outputSerializer);
+ new OneInputStreamOperatorTestHarness<>(
+ createOperatorFactory(catalogLoader), inputSerializer);
+ harness.setup(outputSerializer());
return harness;
}
+ private OneInputStreamOperatorTestHarness<CdcMultiplexRecord,
MultiTableCommittable>
+ createTestHarness(CatalogLoader catalogLoader, int numTasks, int
subtaskId)
+ throws Exception {
+ return createTestHarness(createOperatorFactory(catalogLoader),
numTasks, subtaskId);
+ }
+
+ private OneInputStreamOperatorTestHarness<CdcMultiplexRecord,
MultiTableCommittable>
+ createTestHarness(
+ CdcRecordStoreMultiWriteOperator.Factory operatorFactory,
+ int numTasks,
+ int subtaskId)
+ throws Exception {
+ OneInputStreamOperatorTestHarness<CdcMultiplexRecord,
MultiTableCommittable> harness =
+ new OneInputStreamOperatorTestHarness<>(
+ operatorFactory, numTasks, numTasks, subtaskId);
+ harness.setup(outputSerializer());
+ return harness;
+ }
+
+ private TypeSerializer<MultiTableCommittable> outputSerializer() {
+ return new MultiTableCommittableTypeInfo().createSerializer(new
ExecutionConfig());
+ }
+
+ private CdcRecordStoreMultiWriteOperator.Factory createOperatorFactory(
+ CatalogLoader catalogLoader) {
+ return new CdcRecordStoreMultiWriteOperator.Factory(
+ catalogLoader,
+ storeSinkWriteProvider(),
+ commitUser,
+ databaseName,
+ Options.fromMap(new HashMap<>()));
+ }
+
+ @SuppressWarnings("deprecation")
+ private CdcRecordStoreMultiWriteOperator.Factory
createOperatorFactoryWithoutDatabaseName() {
+ return new CdcRecordStoreMultiWriteOperator.Factory(
+ catalogLoader,
+ storeSinkWriteProvider(),
+ commitUser,
+ Options.fromMap(new HashMap<>()));
+ }
+
+ private StoreSinkWrite.Provider storeSinkWriteProvider() {
+ return (t, commitUser, state, ioManager, memoryPoolFactory,
metricGroup) ->
+ new StoreSinkWriteImpl(
+ t,
+ commitUser,
+ state,
+ ioManager,
+ false,
+ false,
+ true,
+ memoryPoolFactory,
+ metricGroup);
+ }
+
private static class Runner implements Runnable {
private final OneInputStreamOperatorTestHarness<CdcMultiplexRecord,
MultiTableCommittable>
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSinkTest.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSinkTest.java
index 8ed54fac67..741a0fb9d1 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSinkTest.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSinkTest.java
@@ -20,6 +20,7 @@ package org.apache.paimon.flink.sink.cdc;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.FlinkConnectorOptions;
+import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
import org.apache.paimon.options.Options;
import org.apache.flink.api.dag.Transformation;
@@ -34,12 +35,13 @@ import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link FlinkCdcMultiTableSink}. */
public class FlinkCdcMultiTableSinkTest {
@Test
- public void testTransformationParallelism() {
+ public void testTransformationParallelismAndShuffle() {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(8);
int inputParallelism = ThreadLocalRandom.current().nextInt(8) + 1;
@@ -50,6 +52,7 @@ public class FlinkCdcMultiTableSinkTest {
FlinkCdcMultiTableSink sink =
new FlinkCdcMultiTableSink(
() -> FlinkCatalogFactory.createPaimonCatalog(new
Options()),
+ "test_db",
FlinkConnectorOptions.SINK_WRITER_CPU.defaultValue(),
null,
FlinkConnectorOptions.SINK_COMMITTER_CPU.defaultValue(),
@@ -76,5 +79,66 @@ public class FlinkCdcMultiTableSinkTest {
(OneInputTransformation<?, ?>) partitioner.getInputs().get(0);
assertThat(writer.getName()).isEqualTo("CDC MultiplexWriter");
assertThat(writer.getParallelism()).isEqualTo(inputParallelism);
+
+ // The sink must shuffle its input by bucket itself, otherwise the
writer states restored by
+ // CdcRecordStoreMultiWriteOperator would land in subtasks which never
write the
+ // corresponding buckets. Do not drop this shuffle.
+ PartitionTransformation<?> writerInput =
+ (PartitionTransformation<?>) writer.getInputs().get(0);
+
assertThat(writerInput.getPartitioner()).isInstanceOf(FlinkStreamPartitioner.class);
+ assertThat(writerInput.getPartitioner()).hasToString("shuffle by
bucket");
+ assertThat(writerInput.getParallelism()).isEqualTo(inputParallelism);
+
assertThat(writerInput.getInputs().get(0)).isSameAs(input.getTransformation());
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ public void testCompatibilityConstructorPreservesInputTopology() {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStreamSource<CdcMultiplexRecord> input =
+ env.fromData(CdcMultiplexRecord.class, new
CdcMultiplexRecord("", "", null));
+
+ FlinkCdcMultiTableSink sink =
+ new FlinkCdcMultiTableSink(
+ () -> FlinkCatalogFactory.createPaimonCatalog(new
Options()),
+ FlinkConnectorOptions.SINK_WRITER_CPU.defaultValue(),
+ null,
+
FlinkConnectorOptions.SINK_COMMITTER_CPU.defaultValue(),
+ null,
+ UUID.randomUUID().toString(),
+ false,
+ null);
+ Transformation<?> end = sink.sinkFrom(input).getTransformation();
+ OneInputTransformation<?, ?> committer =
+ (OneInputTransformation<?, ?>) end.getInputs().get(0);
+ PartitionTransformation<?> committablePartitioner =
+ (PartitionTransformation<?>) committer.getInputs().get(0);
+ OneInputTransformation<?, ?> writer =
+ (OneInputTransformation<?, ?>)
committablePartitioner.getInputs().get(0);
+
+
assertThat(writer.getInputs().get(0)).isSameAs(input.getTransformation());
+ assertThatThrownBy(() -> sink.sinkFrom(input, 4))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Explicit parallelism")
+ .hasMessageContaining("database name");
+ }
+
+ @Test
+ public void testDatabaseAwareConstructorRejectsNullDatabase() {
+ assertThatThrownBy(
+ () ->
+ new FlinkCdcMultiTableSink(
+ () ->
+
FlinkCatalogFactory.createPaimonCatalog(
+ new Options()),
+ null,
+
FlinkConnectorOptions.SINK_WRITER_CPU.defaultValue(),
+ null,
+
FlinkConnectorOptions.SINK_COMMITTER_CPU.defaultValue(),
+ null,
+ UUID.randomUUID().toString(),
+ false,
+ null))
+ .isInstanceOf(NullPointerException.class);
}
}
