yuxiqian commented on code in PR #3339:
URL: https://github.com/apache/flink-cdc/pull/3339#discussion_r1612889656
##########
flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/SchemaOperatorTranslator.java:
##########
@@ -58,12 +59,18 @@ public DataStream<Event> translate(
MetadataApplier metadataApplier,
List<RouteDef> routes) {
switch (schemaChangeBehavior) {
- case EVOLVE:
- return addSchemaOperator(input, parallelism, metadataApplier,
routes);
case IGNORE:
- return dropSchemaChangeEvent(input, parallelism);
+ return dropSchemaChangeEvent(
+ addSchemaOperator(input, parallelism, metadataApplier,
routes, true),
+ parallelism);
+ case TRY_EVOLVE:
+ return addSchemaOperator(input, parallelism, metadataApplier,
routes, true);
+ case EVOLVE:
+ return addSchemaOperator(input, parallelism, metadataApplier,
routes, false);
case EXCEPTION:
- return exceptionOnSchemaChange(input, parallelism);
+ return exceptionOnSchemaChange(
Review Comment:
Thanks for pointing it out! I think it would be better fix it in this PR
since SchemaOperator & Registry will be greatly modified to implement
FLINK-35242.
(Seems #3352 / FLINK-35432 is about an irrelevant problem about MySQL ddl
parsing, and should be fine to be reviewed and merged.)
##########
flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/SchemaOperatorTranslator.java:
##########
@@ -58,12 +59,18 @@ public DataStream<Event> translate(
MetadataApplier metadataApplier,
List<RouteDef> routes) {
switch (schemaChangeBehavior) {
- case EVOLVE:
- return addSchemaOperator(input, parallelism, metadataApplier,
routes);
case IGNORE:
- return dropSchemaChangeEvent(input, parallelism);
+ return dropSchemaChangeEvent(
+ addSchemaOperator(input, parallelism, metadataApplier,
routes, true),
+ parallelism);
+ case TRY_EVOLVE:
+ return addSchemaOperator(input, parallelism, metadataApplier,
routes, true);
+ case EVOLVE:
+ return addSchemaOperator(input, parallelism, metadataApplier,
routes, false);
case EXCEPTION:
- return exceptionOnSchemaChange(input, parallelism);
+ return exceptionOnSchemaChange(
Review Comment:
Thanks for pointing it out! I think it would be better fix it in this PR
since SchemaOperator & Registry will be greatly modified to implement
FLINK-35242.
(Seems #3352 / FLINK-35432 is about an irrelevant problem about MySQL ddl
parsing, and should be fine to be reviewed and merged independently.)
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
