This is an automated email from the ASF dual-hosted git repository.
loserwang1024 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 c1998ea96 [FLINK-40982][pipeline-connector/fluss] Make ADD COLUMN
replay idempotent (#4574)
c1998ea96 is described below
commit c1998ea9633a4c10b752706d2f09328f8b933a00
Author: Xiaobing Fang <[email protected]>
AuthorDate: Wed Oct 21 15:13:07 2026 +0800
[FLINK-40982][pipeline-connector/fluss] Make ADD COLUMN replay idempotent
(#4574)
---
.../fluss/sink/FlussMetaDataApplier.java | 98 ++++-
.../fluss/sink/FlussMetadataApplierTest.java | 452 ++++++++++++++++++++-
.../fluss/sink/FlussSinkPipelineITCase.java | 54 +++
3 files changed, 598 insertions(+), 6 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
index 3622d4992..3ec152430 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java
@@ -17,6 +17,7 @@
package org.apache.flink.cdc.connectors.fluss.sink;
+import org.apache.flink.cdc.common.annotation.VisibleForTesting;
import org.apache.flink.cdc.common.event.AddColumnEvent;
import org.apache.flink.cdc.common.event.CreateTableEvent;
import org.apache.flink.cdc.common.event.DropTableEvent;
@@ -35,11 +36,15 @@ import org.apache.fluss.client.Connection;
import org.apache.fluss.client.ConnectionFactory;
import org.apache.fluss.client.admin.Admin;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.InvalidAlterTableException;
import org.apache.fluss.metadata.DatabaseDescriptor;
import org.apache.fluss.metadata.TableChange;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.types.RowType;
+import org.apache.fluss.utils.ExceptionUtils;
+import org.apache.fluss.utils.ExponentialBackoff;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -62,6 +67,11 @@ import static
org.apache.flink.cdc.connectors.fluss.utils.FlussConversions.toFlu
/** {@link MetadataApplier} for fluss. */
public class FlussMetaDataApplier implements MetadataApplier,
ExistingTableSchemaExpansionSupport {
private static final Logger LOG =
LoggerFactory.getLogger(FlussMetaDataApplier.class);
+ private static final int MAX_SCHEMA_READ_RETRIES = 3;
+
+ // Zero jitter keeps the backoff sequence deterministic for tests.
+ private static final ExponentialBackoff SCHEMA_READ_BACKOFF =
+ new ExponentialBackoff(100L, 2, 400L, 0.0);
private final Configuration flussClientConfig;
private final Map<String, String> tableProperties;
private final Map<String, List<String>> bucketKeysMap;
@@ -218,13 +228,99 @@ public class FlussMetaDataApplier implements
MetadataApplier, ExistingTableSchem
Admin admin = connection.getAdmin()) {
TableId tableId = event.tableId();
TablePath tablePath = new TablePath(tableId.getSchemaName(),
tableId.getTableName());
- admin.alterTable(tablePath, tableChanges, true).get();
+ applyAddColumnChanges(admin, tablePath, tableChanges);
+ } catch (InterruptedException e) {
+ // Wrapping loses the exception type, so keep the cancellation
signal alive.
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
} catch (Exception e) {
LOG.error("Failed to apply schema change {}", event, e);
throw new RuntimeException(e);
}
}
+ @VisibleForTesting
+ void applyAddColumnChanges(Admin admin, TablePath tablePath,
List<TableChange> tableChanges)
+ throws Exception {
+ if (tableChanges.isEmpty()) {
+ return;
+ }
+ try {
+ admin.alterTable(tablePath, tableChanges, false).get();
+ } catch (Exception e) {
+ List<TableChange> remainingChanges =
+ reconcileWithTargetSchema(admin, tablePath, tableChanges,
e);
+ if (!remainingChanges.isEmpty()) {
+ // Reconcile once; concurrent changes or further stale reads
must fail normally.
+ admin.alterTable(tablePath, remainingChanges, false).get();
+ }
+ }
+ }
+
+ /**
+ * Returns the changes the target schema still needs, or rethrows {@code
cause} when the failed
+ * ADD is not a replay of an already applied one.
+ */
+ private List<TableChange> reconcileWithTargetSchema(
+ Admin admin, TablePath tablePath, List<TableChange> tableChanges,
Exception cause)
+ throws Exception {
+ if (!(ExceptionUtils.stripExecutionException(cause)
+ instanceof InvalidAlterTableException)) {
+ throw cause;
+ }
+ Set<String> requestedNames = new HashSet<>();
+ for (TableChange change : tableChanges) {
+ if (!requestedNames.add(((TableChange.AddColumn)
change).getName())) {
+ throw cause;
+ }
+ }
+ List<TableChange> remainingChanges =
+ getRemainingChanges(admin.getTableInfo(tablePath).get(),
tableChanges);
+ // alterTable is served by the CoordinatorServer while getTableInfo is
served by a random
+ // TabletServer reading ZooKeeper without sync(), so the read can lag
behind the write that
+ // made this ADD fail. Retry before concluding the failure is
unrelated to a replay.
+ for (int attempt = 0;
+ remainingChanges.size() == tableChanges.size() && attempt <
MAX_SCHEMA_READ_RETRIES;
+ attempt++) {
+ Thread.sleep(SCHEMA_READ_BACKOFF.backoff(attempt));
+ remainingChanges =
+ getRemainingChanges(admin.getTableInfo(tablePath).get(),
tableChanges);
+ }
+ if (remainingChanges.size() == tableChanges.size()) {
+ // Nothing was already applied, so the failure cannot be a replay.
+ throw cause;
+ }
+ return remainingChanges;
+ }
+
+ /**
+ * Filters out the changes already reflected in {@code tableInfo}. Field
ids are deliberately
+ * ignored: {@code toFlussType} derives them as -1 while the server
assigns real ones.
+ */
+ private List<TableChange> getRemainingChanges(
+ TableInfo tableInfo, List<TableChange> tableChanges) {
+ RowType rowType = tableInfo.getRowType();
+ List<TableChange> remainingChanges = new ArrayList<>();
+ for (TableChange change : tableChanges) {
+ TableChange.AddColumn addition = (TableChange.AddColumn) change;
+ int index = rowType.getFieldIndex(addition.getName());
+ if (index < 0) {
+ remainingChanges.add(change);
+ } else if
(!rowType.getTypeAt(index).equals(addition.getDataType())) {
+ throw new IllegalArgumentException(
+ "Column "
+ + addition.getName()
+ + " already exists in table "
+ + tableInfo.getTablePath()
+ + " with type "
+ + rowType.getTypeAt(index)
+ + ", but ADD COLUMN requests "
+ + addition.getDataType());
+ }
+ }
+ return remainingChanges;
+ }
+
private void sanityCheck(
TableDescriptor inferredFlussTable,
TableInfo currentTableInfo,
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
index dface1798..1faf881f8 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.cdc.common.event.TableId;
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.types.DataType;
import org.apache.flink.cdc.common.types.DataTypes;
import org.apache.flink.cdc.common.types.IntType;
import
org.apache.flink.cdc.runtime.operators.schema.common.ExistingTableSchemaExpander;
@@ -33,8 +34,12 @@ import org.apache.flink.table.api.ValidationException;
import org.apache.fluss.client.Connection;
import org.apache.fluss.client.ConnectionFactory;
import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.exception.InvalidAlterTableException;
import org.apache.fluss.exception.InvalidConfigException;
+import org.apache.fluss.exception.TableNotExistException;
+import org.apache.fluss.exception.UnknownServerException;
import org.apache.fluss.metadata.DatabaseDescriptor;
+import org.apache.fluss.metadata.TableChange;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
@@ -44,18 +49,28 @@ import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
+import java.lang.reflect.Proxy;
+import java.util.ArrayList;
import java.util.Arrays;
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.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
import static
org.apache.flink.cdc.connectors.fluss.utils.FlussConversions.toFlussSchema;
+import static
org.apache.flink.cdc.connectors.fluss.utils.FlussConversions.toFlussType;
import static org.apache.fluss.config.ConfigOptions.TABLE_REPLICATION_FACTOR;
import static org.apache.fluss.types.DataTypeChecks.equalsWithFieldId;
import static org.assertj.core.api.Assertions.assertThat;
@@ -120,8 +135,8 @@ public class FlussMetadataApplierTest {
"row_col"
};
- org.apache.flink.cdc.common.types.DataType[] cdcDataTypes =
- new org.apache.flink.cdc.common.types.DataType[] {
+ DataType[] cdcDataTypes =
+ new DataType[] {
DataTypes.BINARY(10),
DataTypes.VARBINARY(10),
DataTypes.BYTES(),
@@ -296,8 +311,7 @@ public class FlussMetadataApplierTest {
void testUnsupportedType() throws Exception {
String[] fieldNames = new String[] {"timestamp_tz_col"};
- org.apache.flink.cdc.common.types.DataType[] cdcDataTypes =
- new org.apache.flink.cdc.common.types.DataType[]
{DataTypes.TIMESTAMP_TZ()};
+ DataType[] cdcDataTypes = new DataType[] {DataTypes.TIMESTAMP_TZ()};
try (FlussMetaDataApplier applier =
new FlussMetaDataApplier(
@@ -600,7 +614,8 @@ public class FlussMetadataApplierTest {
Collections.singletonList(
AddColumnEvent.last(oldColumn)))))
.rootCause()
- .hasMessageContaining("Column name already exists.");
+ .hasMessageContaining("Column name already exists in
table")
+ .hasMessageContaining("with type INT, but ADD COLUMN
requests STRING");
applier.applySchemaChange(
new AddColumnEvent(
tableId,
Collections.singletonList(AddColumnEvent.last(newColumn))));
@@ -615,6 +630,433 @@ public class FlussMetadataApplierTest {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testReplayAddColumns(boolean partiallyApplied) throws Exception {
+ TableId tableId = TableId.tableId(DATABASE_NAME, "replay_columns");
+ TablePath tablePath = new TablePath(DATABASE_NAME, "replay_columns");
+ Column name = Column.physicalColumn("name", DataTypes.VARCHAR(100));
+ Column details =
+ Column.physicalColumn(
+ "details",
+ DataTypes.ROW(DataTypes.FIELD("amount",
DataTypes.DECIMAL(10, 2))));
+ AddColumnEvent event =
+ new AddColumnEvent(
+ tableId,
+ Arrays.asList(AddColumnEvent.last(name),
AddColumnEvent.last(details)));
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ applier.applySchemaChange(
+ new CreateTableEvent(
+ tableId,
+ Schema.newBuilder().physicalColumn("id",
DataTypes.INT()).build()));
+ applier.applySchemaChange(
+ partiallyApplied
+ ? new AddColumnEvent(
+ tableId,
Collections.singletonList(AddColumnEvent.last(name)))
+ : event);
+ }
+
+ // A restored job has a new applier, but the external schema has not
rolled back.
+ try (FlussMetaDataApplier restoredApplier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ restoredApplier.applySchemaChange(event);
+ TableInfo afterReplay = admin.getTableInfo(tablePath).get();
+ assertThat(afterReplay.getSchema())
+ .isEqualTo(
+ toFlussSchema(
+ Schema.newBuilder()
+ .physicalColumn("id",
DataTypes.INT())
+ .physicalColumn("name",
DataTypes.STRING())
+ .column(details)
+ .build()));
+ restoredApplier.applySchemaChange(event);
+ assertThat(admin.getTableInfo(tablePath).get().getSchemaId())
+ .isEqualTo(afterReplay.getSchemaId());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testReplayAddColumnsWithStaleSchema(boolean partiallyApplied) throws
Exception {
+ TableId tableId = TableId.tableId(DATABASE_NAME, "stale_columns");
+ TablePath tablePath = new TablePath(DATABASE_NAME, "stale_columns");
+ Column first = Column.physicalColumn("a", DataTypes.INT());
+ Column second = Column.physicalColumn("b", DataTypes.INT());
+ AddColumnEvent event =
+ new AddColumnEvent(
+ tableId,
+ Arrays.asList(AddColumnEvent.last(first),
AddColumnEvent.last(second)));
+ List<TableChange> changes = addIntColumns("a", "b");
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ applier.applySchemaChange(
+ new CreateTableEvent(
+ tableId,
+ Schema.newBuilder().physicalColumn("id",
DataTypes.INT()).build()));
+ TableInfo staleTableInfo = admin.getTableInfo(tablePath).get();
+ applier.applySchemaChange(
+ partiallyApplied
+ ? new AddColumnEvent(
+ tableId,
Collections.singletonList(AddColumnEvent.last(first)))
+ : event);
+ AtomicBoolean returnStaleSchema = new AtomicBoolean(true);
+ Admin staleAdmin =
+ (Admin)
+ Proxy.newProxyInstance(
+ Admin.class.getClassLoader(),
+ new Class<?>[] {Admin.class},
+ (proxy, method, args) -> {
+ if
(method.getName().equals("getTableInfo")
+ &&
returnStaleSchema.getAndSet(false)) {
+ return
CompletableFuture.completedFuture(
+ staleTableInfo);
+ }
+ return method.invoke(admin, args);
+ });
+
+ applier.applyAddColumnChanges(staleAdmin, tablePath, changes);
+
+ TableInfo afterReplay = admin.getTableInfo(tablePath).get();
+ assertThat(afterReplay.getSchema())
+ .isEqualTo(
+ toFlussSchema(
+ Schema.newBuilder()
+ .physicalColumn("id",
DataTypes.INT())
+ .column(first)
+ .column(second)
+ .build()));
+ applier.applySchemaChange(event);
+ assertThat(admin.getTableInfo(tablePath).get().getSchemaId())
+ .isEqualTo(afterReplay.getSchemaId());
+ }
+ }
+
+ @Test
+ @Timeout(30)
+ void testReplayAddColumnFailsWhenSchemaRemainsStale() throws Exception {
+ TableId tableId = TableId.tableId(DATABASE_NAME,
"persistently_stale_schema");
+ TablePath tablePath = new TablePath(DATABASE_NAME,
"persistently_stale_schema");
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ applier.applySchemaChange(
+ new CreateTableEvent(
+ tableId,
+ Schema.newBuilder().physicalColumn("id",
DataTypes.INT()).build()));
+ TableInfo staleTableInfo = admin.getTableInfo(tablePath).get();
+ List<TableChange> changes = addIntColumns("a");
+ admin.alterTable(tablePath, changes, false).get();
+ TableInfo beforeReplay = admin.getTableInfo(tablePath).get();
+ AtomicInteger schemaReads = new AtomicInteger();
+ AtomicInteger alterCalls = new AtomicInteger();
+ Admin staleAdmin =
+ (Admin)
+ Proxy.newProxyInstance(
+ Admin.class.getClassLoader(),
+ new Class<?>[] {Admin.class},
+ (proxy, method, args) -> {
+ if
(method.getName().equals("getTableInfo")) {
+ schemaReads.incrementAndGet();
+ return
CompletableFuture.completedFuture(
+ staleTableInfo);
+ }
+ if
(method.getName().equals("alterTable")) {
+ alterCalls.incrementAndGet();
+ }
+ return method.invoke(admin, args);
+ });
+
+ assertThatThrownBy(() -> applier.applyAddColumnChanges(staleAdmin,
tablePath, changes))
+ .rootCause()
+ .isInstanceOf(InvalidAlterTableException.class)
+ .hasMessageContaining("Column a already exists");
+ assertThat(schemaReads.get()).isBetween(2, 4);
+ assertThat(alterCalls.get()).isEqualTo(1);
+ assertThat(admin.getTableInfo(tablePath).get().getSchemaId())
+ .isEqualTo(beforeReplay.getSchemaId());
+ }
+ }
+
+ @Test
+ void testReplayAddColumnsPropagatesConcurrentAdditionFailure() throws
Exception {
+ TableId tableId = TableId.tableId(DATABASE_NAME,
"concurrent_addition");
+ TablePath tablePath = new TablePath(DATABASE_NAME,
"concurrent_addition");
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ applier.applySchemaChange(
+ new CreateTableEvent(
+ tableId,
+ Schema.newBuilder().physicalColumn("id",
DataTypes.INT()).build()));
+ admin.alterTable(tablePath, addIntColumns("a"), false).get();
+ AtomicInteger schemaReads = new AtomicInteger();
+ AtomicInteger alterCalls = new AtomicInteger();
+ Admin racingAdmin =
+ (Admin)
+ Proxy.newProxyInstance(
+ Admin.class.getClassLoader(),
+ new Class<?>[] {Admin.class},
+ (proxy, method, args) -> {
+ if
(method.getName().equals("alterTable")
+ &&
alterCalls.incrementAndGet() == 2) {
+ // Another writer adds b after we
have observed a.
+ admin.alterTable(tablePath,
addIntColumns("b"), false)
+ .get();
+ }
+ if
(method.getName().equals("getTableInfo")) {
+ schemaReads.incrementAndGet();
+ }
+ return method.invoke(admin, args);
+ });
+
+ assertThatThrownBy(
+ () ->
+ applier.applyAddColumnChanges(
+ racingAdmin, tablePath,
addIntColumns("a", "b")))
+ .rootCause()
+ .isInstanceOf(InvalidAlterTableException.class)
+ .hasMessageContaining("Column b already exists");
+
+ TableInfo afterReplay = admin.getTableInfo(tablePath).get();
+ assertThat(afterReplay.getSchema())
+ .isEqualTo(
+ toFlussSchema(
+ Schema.newBuilder()
+ .physicalColumn("id",
DataTypes.INT())
+ .physicalColumn("a",
DataTypes.INT())
+ .physicalColumn("b",
DataTypes.INT())
+ .build()));
+ assertThat(alterCalls.get()).isEqualTo(2);
+ assertThat(schemaReads.get()).isEqualTo(1);
+ applier.applyAddColumnChanges(admin, tablePath, addIntColumns("a",
"b"));
+ assertThat(admin.getTableInfo(tablePath).get().getSchemaId())
+ .isEqualTo(afterReplay.getSchemaId());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testSchemaReadRetryPropagatesInterruption(boolean pendingRead) throws
Exception {
+ TableId tableId = TableId.tableId(DATABASE_NAME,
"interrupted_schema_read");
+ TablePath tablePath = new TablePath(DATABASE_NAME,
"interrupted_schema_read");
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ applier.applySchemaChange(
+ new CreateTableEvent(
+ tableId,
+ Schema.newBuilder().physicalColumn("id",
DataTypes.INT()).build()));
+ TableInfo staleTableInfo = admin.getTableInfo(tablePath).get();
+ admin.alterTable(tablePath, addIntColumns("a"), false).get();
+ AtomicInteger schemaReads = new AtomicInteger();
+ Admin interruptedAdmin =
+ (Admin)
+ Proxy.newProxyInstance(
+ Admin.class.getClassLoader(),
+ new Class<?>[] {Admin.class},
+ (proxy, method, args) -> {
+ if
(method.getName().equals("getTableInfo")) {
+ schemaReads.incrementAndGet();
+ Thread.currentThread().interrupt();
+ return pendingRead
+ ? new CompletableFuture<>()
+ :
CompletableFuture.completedFuture(
+ staleTableInfo);
+ }
+ return method.invoke(admin, args);
+ });
+ try {
+ assertThatThrownBy(
+ () ->
+ applier.applyAddColumnChanges(
+ interruptedAdmin, tablePath,
addIntColumns("a")))
+ .isInstanceOf(InterruptedException.class);
+ assertThat(schemaReads.get()).isEqualTo(1);
+ } finally {
+ Thread.interrupted();
+ }
+ }
+ }
+
+ private static List<TableChange> addIntColumns(String... names) {
+ List<TableChange> changes = new ArrayList<>();
+ for (String name : names) {
+ changes.add(
+ TableChange.addColumn(
+ name,
+ toFlussType(DataTypes.INT()),
+ null,
+ TableChange.ColumnPosition.last()));
+ }
+ return changes;
+ }
+
+ @ParameterizedTest
+ @MethodSource("conflictingColumnTypes")
+ void testReplayAddColumnWithConflictingType(DataType requestedType) throws
Exception {
+ TableId tableId = TableId.tableId(DATABASE_NAME,
"conflicting_columns");
+ TablePath tablePath = new TablePath(DATABASE_NAME,
"conflicting_columns");
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ applier.applySchemaChange(
+ new CreateTableEvent(
+ tableId,
+ Schema.newBuilder()
+ .physicalColumn("amount",
DataTypes.DECIMAL(10, 2))
+ .build()));
+ TableInfo beforeReplay = admin.getTableInfo(tablePath).get();
+ AddColumnEvent event =
+ new AddColumnEvent(
+ tableId,
+ Arrays.asList(
+ AddColumnEvent.last(
+ Column.physicalColumn("missing",
DataTypes.INT())),
+ AddColumnEvent.last(
+ Column.physicalColumn("amount",
requestedType))));
+ assertThatThrownBy(() -> applier.applySchemaChange(event))
+ .rootCause()
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Column amount already exists in
table")
+ .hasMessageContaining("but ADD COLUMN requests");
+ TableInfo afterReplay = admin.getTableInfo(tablePath).get();
+
assertThat(afterReplay.getSchema()).isEqualTo(beforeReplay.getSchema());
+
assertThat(afterReplay.getSchemaId()).isEqualTo(beforeReplay.getSchemaId());
+ }
+ }
+
+ private static Stream<DataType> conflictingColumnTypes() {
+ return Stream.of(
+ DataTypes.STRING(),
+ DataTypes.DECIMAL(12, 2),
+ DataTypes.DECIMAL(10, 3),
+ DataTypes.DECIMAL(10, 2).notNull());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ @Timeout(30)
+ void testInvalidAddColumnIsNotIgnored(boolean partiallyApplied) throws
Exception {
+ TableId tableId = TableId.tableId(DATABASE_NAME, "invalid_columns");
+ TablePath tablePath = new TablePath(DATABASE_NAME, "invalid_columns");
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ applier.applySchemaChange(
+ new CreateTableEvent(
+ tableId,
+ Schema.newBuilder().physicalColumn("id",
DataTypes.INT()).build()));
+ TableInfo beforeAddition = admin.getTableInfo(tablePath).get();
+ List<AddColumnEvent.ColumnWithPosition> columns = new
ArrayList<>();
+ if (partiallyApplied) {
+ columns.add(AddColumnEvent.last(Column.physicalColumn("id",
DataTypes.INT())));
+ }
+ columns.add(
+ AddColumnEvent.last(
+ Column.physicalColumn("required",
DataTypes.INT().notNull())));
+
+ assertThatThrownBy(
+ () -> applier.applySchemaChange(new
AddColumnEvent(tableId, columns)))
+ .rootCause()
+ .isInstanceOf(UnknownServerException.class)
+ .hasMessageContaining("Column required must be nullable");
+ TableInfo afterAddition = admin.getTableInfo(tablePath).get();
+
assertThat(afterAddition.getSchema()).isEqualTo(beforeAddition.getSchema());
+
assertThat(afterAddition.getSchemaId()).isEqualTo(beforeAddition.getSchemaId());
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"false, false", "true, false", "true, true"})
+ @Timeout(30)
+ void testDuplicateAddColumnsAreRejected(boolean columnExists, boolean
addMissingColumn)
+ throws Exception {
+ TableId tableId = TableId.tableId(DATABASE_NAME,
"duplicate_additions");
+ TablePath tablePath = new TablePath(DATABASE_NAME,
"duplicate_additions");
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ Schema.Builder schema = Schema.newBuilder().physicalColumn("id",
DataTypes.INT());
+ if (columnExists) {
+ schema.physicalColumn("name", DataTypes.STRING());
+ }
+ applier.applySchemaChange(new CreateTableEvent(tableId,
schema.build()));
+ TableInfo beforeAddition = admin.getTableInfo(tablePath).get();
+ List<AddColumnEvent.ColumnWithPosition> additions =
+ new ArrayList<>(
+ Collections.nCopies(
+ 2,
+ AddColumnEvent.last(
+ Column.physicalColumn("name",
DataTypes.STRING()))));
+ if (addMissingColumn) {
+ additions.add(AddColumnEvent.last(Column.physicalColumn("age",
DataTypes.INT())));
+ }
+ AddColumnEvent event = new AddColumnEvent(tableId, additions);
+ assertThatThrownBy(() -> applier.applySchemaChange(event))
+ .rootCause()
+ .isInstanceOf(InvalidAlterTableException.class)
+ .hasMessageContaining("Column name already exists");
+ TableInfo afterAddition = admin.getTableInfo(tablePath).get();
+
assertThat(afterAddition.getSchema()).isEqualTo(beforeAddition.getSchema());
+
assertThat(afterAddition.getSchemaId()).isEqualTo(beforeAddition.getSchemaId());
+ }
+ }
+
+ @Test
+ void testAddColumnToMissingTable() throws Exception {
+ TableId tableId = TableId.tableId(DATABASE_NAME, "missing_table");
+ try (FlussMetaDataApplier applier =
+ new FlussMetaDataApplier(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap())) {
+ assertThatThrownBy(
+ () ->
+ applier.applySchemaChange(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+
AddColumnEvent.last(
+
Column.physicalColumn(
+
"id",
+
DataTypes.INT()))))))
+ .rootCause()
+ .isInstanceOf(TableNotExistException.class);
+ }
+ }
+
@Test
void testRecreateTableWithDifferentSchema() throws Exception {
TableId tableId = TableId.tableId("default_namespace", DATABASE_NAME,
"table1");
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussSinkPipelineITCase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussSinkPipelineITCase.java
index 455f1e8cf..c0532faf5 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussSinkPipelineITCase.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussSinkPipelineITCase.java
@@ -57,6 +57,8 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import java.time.Duration;
import java.util.ArrayList;
@@ -709,6 +711,58 @@ public class FlussSinkPipelineITCase {
"+I[1, a, null, null]", "+I[2, b, null, null]", "+I[3,
c, null, cc]"));
}
+ @ParameterizedTest
+ @CsvSource({"true,true", "true,false", "false,true", "false,false"})
+ void testWriteAfterAddColumnReplay(boolean primaryKeyTable, boolean
partiallyApplied)
+ throws Exception {
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "CREATE TABLE %s.%s (payload STRING, id STRING
NOT NULL, b STRING%s%s)",
+ TABLE_1.getSchemaName(),
+ TABLE_1.getTableName(),
+ partiallyApplied ? "" : ", a STRING",
+ primaryKeyTable ? ", PRIMARY KEY (id) NOT
ENFORCED" : ""))
+ .await();
+
+ Schema.Builder schemaBuilder =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.STRING().notNull())
+ .physicalColumn("payload", DataTypes.STRING());
+ if (primaryKeyTable) {
+ schemaBuilder.primaryKey("id");
+ }
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(
+ RowType.of(
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()));
+ List<Event> events =
+ Arrays.asList(
+ new CreateTableEvent(TABLE_1, schemaBuilder.build()),
+ new AddColumnEvent(
+ TABLE_1,
+ Arrays.asList(
+ AddColumnEvent.last(
+ Column.physicalColumn("a",
DataTypes.STRING())),
+ AddColumnEvent.last(
+ Column.physicalColumn("b",
DataTypes.STRING())))),
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+
BinaryStringData.fromString("value-1"),
+
BinaryStringData.fromString("a-101"),
+
BinaryStringData.fromString("b-202")
+ })));
+
+ composeAndExecuteInEvolveMode(Collections.singletonList(events));
+ checkResult(TABLE_1, Collections.singletonList("+I[value-1, 1, b-202,
a-101]"));
+ }
+
private void composeAndExecuteInEvolveMode(List<List<Event>>
customSourceEvents)
throws Exception {
composeAndExecute(
