loserwang1024 commented on code in PR #4574:
URL: https://github.com/apache/flink-cdc/pull/4574#discussion_r4268721943
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java:
##########
@@ -218,13 +225,82 @@ private void applyAddColumnTable(AddColumnEvent event) {
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 (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)
Review Comment:
Could we simplify this to a single schema reconciliation followed by one ADD
attempt, propagating any subsequent failure?
The outer while also handles concurrent column additions and partially stale
schema reads. That provides additional resilience, but makes the implementation
more complex than basic replay idempotency requires. Is handling these
situations within the same call an intended requirement of this fix? If so,
please explain that scope and the termination condition in the code.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java:
##########
Review Comment:
Could we add a pipeline-level regression test that verifies actual writes
after ADD COLUMN replay?
FlussMetadataApplierTest validates the external schema, but does not
exercise the sink's schema refresh, schema ID selection, or column mapping. We
should start a pipeline with the old schema while Fluss already contains some
or all of the added columns, replay the ADD event, then write records and read
them back.
Please cover primary-key and log tables, full and partial replay, and
different upstream/target column orders. I locally verified these four
combinations: they pass with this PR and fail with the previous ADD
implementation due to duplicate-column errors. These cases would be useful
regression coverage in the PR.
A checkpoint recovery test that fails after the external DDL succeeds but
before the next checkpoint completes would provide further confidence.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
