Skip to content

[FLINK-37677][cdc-common][cdc-runtime] Handle Exclusion of create.table Events in Flink CDC Schema Evolution #4015

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[FLINK-37677][cdc-common][cdc-runtime] Handle Exclusion of create.table Events in Flink CDC Schema Evolution #4015

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
76a5a25
[FLINK-37677][cdc-common][cdc-runtime] Sink supports skipping create …
beryllw May 13, 2025
26e672e
fix spotless-check
beryllw May 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/MetadataApplier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ default MetadataApplier setAcceptedSchemaEvolutionTypes(
return this;
}

/** Checks if this metadata applier should this event type. */
/** Checks if this metadata applier should accept this event type. */
default boolean acceptsSchemaEvolutionType(SchemaChangeEventType schemaChangeEventType) {
return true;
}
Expand Down
5 changes: 4 additions & 1 deletion ...e/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaDerivator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.flink.cdc.common.event.DropColumnEvent;
import org.apache.flink.cdc.common.event.RenameColumnEvent;
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
import org.apache.flink.cdc.common.event.SchemaChangeEventType;
import org.apache.flink.cdc.common.event.SchemaChangeEventWithPreSchema;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
Expand Down Expand Up @@ -171,7 +172,9 @@ public static List<SchemaChangeEvent> normalizeSchemaChangeEvents(

List<SchemaChangeEvent> finalSchemaChangeEvents = new ArrayList<>();
for (SchemaChangeEvent schemaChangeEvent : rewrittenSchemaChangeEvents) {
if (metadataApplier.acceptsSchemaEvolutionType(schemaChangeEvent.getType())) {
// always accept create.table event
if (metadataApplier.acceptsSchemaEvolutionType(schemaChangeEvent.getType())
|| schemaChangeEvent.getType().equals(SchemaChangeEventType.CREATE_TABLE)) {
finalSchemaChangeEvents.add(schemaChangeEvent);
} else {
LOG.info("Ignored schema change {}.", schemaChangeEvent);
Expand Down
14 changes: 10 additions & 4 deletions ...ain/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -462,11 +462,17 @@ private Tuple2<Set<TableId>, List<SchemaChangeEvent>> deduceEvolvedSchemaChanges

private boolean applyAndUpdateEvolvedSchemaChange(SchemaChangeEvent schemaChangeEvent) {
try {
metadataApplier.applySchemaChange(schemaChangeEvent);
// filter create.table schema change event
if (metadataApplier.acceptsSchemaEvolutionType(schemaChangeEvent.getType())) {
metadataApplier.applySchemaChange(schemaChangeEvent);
LOG.info(
"Successfully applied schema change event {} to external system.",
schemaChangeEvent);
} else {
LOG.info(
"Skip apply schema change event {} to external system.", schemaChangeEvent);
}
schemaManager.applyEvolvedSchemaChange(schemaChangeEvent);
LOG.info(
"Successfully applied schema change event {} to external system.",
schemaChangeEvent);
return true;
} catch (Throwable t) {
handleUnrecoverableError(
Expand Down
7 changes: 6 additions & 1 deletion ...rc/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -434,7 +434,12 @@ private void applySchemaChange(int sourceSubTaskId) {

private boolean applyAndUpdateEvolvedSchemaChange(SchemaChangeEvent schemaChangeEvent) {
try {
metadataApplier.applySchemaChange(schemaChangeEvent);
// filter create.table schema change event
if (metadataApplier.acceptsSchemaEvolutionType(schemaChangeEvent.getType())) {
metadataApplier.applySchemaChange(schemaChangeEvent);
} else {
LOG.info("Skip apply schema change {}.", schemaChangeEvent);
}
schemaManager.applyEvolvedSchemaChange(schemaChangeEvent);
LOG.info(
"Successfully applied schema change event {} to external system.",
Comment on lines +437 to 445
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Move this log in the true branch since no changes will be applied if skipped

Expand Down
60 changes: 60 additions & 0 deletions ...src/test/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaEvolveTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2561,6 +2561,66 @@ tableId, buildRecord(INT, 12, INT, 0, SMALLINT, (short) 11)),
}
}

/** Tests lenient schema change behavior exclude create.table event. */
@Test
void testLenientSchemaEvolvesExcludeCreate() throws Exception {
TableId tableId = CUSTOMERS_TABLE_ID;
Schema schemaV1 =
Schema.newBuilder()
.physicalColumn("id", INT)
.physicalColumn("name", STRING.notNull())
.physicalColumn("age", SMALLINT)
.primaryKey("id")
.build();

SchemaChangeBehavior behavior = SchemaChangeBehavior.LENIENT;

SchemaOperator schemaOperator =
new SchemaOperator(new ArrayList<>(), Duration.ofSeconds(30), behavior);
RegularEventOperatorTestHarness<SchemaOperator, Event> harness =
RegularEventOperatorTestHarness.withDurationAndExcludeCreateTableBehavior(
schemaOperator, 5, Duration.ofSeconds(3), behavior);
Comment on lines +2581 to +2582
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: can we reuse RegularEventOperatorTestHarness#withDurationAndFineGrainedBehaviorWithError?

harness.open();

// Test CreateTableEvent
{
List<Event> createAndInsertDataEvents =
Arrays.asList(
new CreateTableEvent(tableId, schemaV1),
DataChangeEvent.insertEvent(
tableId,
buildRecord(INT, 1, STRING, "Alice", SMALLINT, (short) 17)),
DataChangeEvent.insertEvent(
tableId,
buildRecord(INT, 2, STRING, "Bob", SMALLINT, (short) 18)),
DataChangeEvent.insertEvent(
tableId,
buildRecord(INT, 3, STRING, "Carol", SMALLINT, (short) 19)));

processEvent(schemaOperator, createAndInsertDataEvents);

FlushEvent result;
result =
new FlushEvent(
0,
Collections.singletonList(tableId),
SchemaChangeEventType.CREATE_TABLE);

Assertions.assertThat(
harness.getOutputRecords().stream()
.map(StreamRecord::getValue)
.collect(Collectors.toList()))
.isEqualTo(
ListUtils.union(
Collections.singletonList(result), createAndInsertDataEvents));

Assertions.assertThat(harness.getLatestOriginalSchema(tableId)).isEqualTo(schemaV1);
Assertions.assertThat(harness.getLatestEvolvedSchema(tableId)).isEqualTo(schemaV1);

harness.clearOutputRecords();
}
}

private RecordData buildRecord(final Object... args) {
List<DataType> dataTypes = new ArrayList<>();
List<Object> objects = new ArrayList<>();
Expand Down
16 changes: 16 additions & 0 deletions ...ava/org/apache/flink/cdc/runtime/testutils/operators/RegularEventOperatorTestHarness.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,22 @@ RegularEventOperatorTestHarness<OP, E> withDurationAndBehavior(
Collections.emptySet());
}

public static <OP extends AbstractStreamOperator<E>, E extends Event>
RegularEventOperatorTestHarness<OP, E> withDurationAndExcludeCreateTableBehavior(
OP operator,
int numOutputs,
Duration evolveDuration,
SchemaChangeBehavior behavior) {
return new RegularEventOperatorTestHarness<>(
operator,
numOutputs,
evolveDuration,
DEFAULT_RPC_TIMEOUT,
behavior,
Arrays.stream(SchemaChangeEventTypeFamily.COLUMN).collect(Collectors.toSet()),
Arrays.stream(SchemaChangeEventTypeFamily.TABLE).collect(Collectors.toSet()));
}

public static <OP extends AbstractStreamOperator<E>, E extends Event>
RegularEventOperatorTestHarness<OP, E> withDurationAndFineGrainedBehavior(
OP operator,
Expand Down
Loading