This is an automated email from the ASF dual-hosted git repository. lvyanquan pushed a commit to branch codex/pr-4540 in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
commit 469dda0cb3a7e1f59c4b36464fc117dcfea799b1 Author: lvyanquan <[email protected]> AuthorDate: Tue Sep 22 22:34:36 2026 +0800 [FLINK-40647][runtime] Scope schema expansion to streaming mode and harden its failure handling Restrict the new existing-table schema expansion option to streaming pipelines and fix several correctness and test-harness issues found while validating it end to end. * Batch mode: revert the BatchSchemaOperator wiring added by this feature, so batch pipelines keep the pre-existing sink behavior. The translator now warns when the option is set in batch mode, and the docs state that expansion is streaming only. * Rename the default mode OFF to DISABLED. A bare `OFF` YAML scalar is parsed as the boolean `false`, so the option now rejects non-textual values instead of coercing them, which also drops the parser workaround. * Fail fast when a sink lacks ExistingTableSchemaExpansionSupport instead of degrading to the sink's own handling under TRY_EXPAND, which previously hid a connector misconfiguration. * Split TRY_EXPAND into a probing phase (delegating to the sink when the table cannot be expanded safely) and an applying phase (propagating failures). Before this change a failure while applying or verifying derived DDL fell back to applying the original CreateTableEvent to an already existing table, silently dropping the missing columns. * Centralize the lazy Paimon catalog creation in getCatalog() and reset the cached instance on close so the applier stays reusable. * E2E: the SQL client creates Paimon artifacts as the JobManager user while the sink runs in the TaskManager container, so widening the shared volume permissions is required for applyCreateTable/applyAddColumn to work. The Fluss source table insert additionally needs the Fluss jar shipped through pipeline.jars, because copyJarToFlinkLib only reaches the JobManager and sql-client.sh has no --jar option; the read-back loop now surfaces a failed insert job instead of timing out silently. * Drop the batch expansion E2E case, whose logic is shared with the streaming cases and already covered by unit tests. Co-Authored-By: Codex <[email protected]> AI-Model: gpt-5.6-sol AI-Contributed/Feature: 314/330 AI-Contributed/UT: 367/374 --- .../docs/core-concept/schema-evolution.md | 23 ++- docs/content/docs/core-concept/schema-evolution.md | 23 ++- .../cli/parser/YamlPipelineDefinitionParser.java | 17 +-- .../parser/YamlPipelineDefinitionParserTest.java | 37 +++-- .../pipeline/ExistingTableSchemaExpansionMode.java | 6 +- .../flink/cdc/composer/definition/SinkDef.java | 4 +- .../flink/translator/SchemaOperatorTranslator.java | 24 +++- .../fluss/sink/FlussMetadataApplierTest.java | 7 +- .../paimon/sink/PaimonMetadataApplier.java | 28 ++-- .../paimon/sink/PaimonMetadataApplierTest.java | 39 +++++- .../ExistingTableSchemaExpansionE2eITCase.java | 103 +++++++------- .../schema/common/ExistingTableSchemaExpander.java | 99 +++++++++---- .../operators/schema/common/SchemaRegistry.java | 17 +-- .../schema/distributed/SchemaCoordinator.java | 2 +- .../distributed/SchemaCoordinatorProvider.java | 2 +- .../schema/distributed/SchemaOperatorFactory.java | 2 +- .../schema/regular/BatchSchemaOperator.java | 77 +--------- .../schema/regular/SchemaCoordinator.java | 2 +- .../schema/regular/SchemaCoordinatorProvider.java | 2 +- .../schema/regular/SchemaOperatorFactory.java | 2 +- .../common/ExistingTableSchemaExpanderTest.java | 155 +++++++++++++++------ .../schema/regular/SchemaCoordinatorTest.java | 33 +++++ 22 files changed, 413 insertions(+), 291 deletions(-) diff --git a/docs/content.zh/docs/core-concept/schema-evolution.md b/docs/content.zh/docs/core-concept/schema-evolution.md index 1789b9a96..c08d1323d 100644 --- a/docs/content.zh/docs/core-concept/schema-evolution.md +++ b/docs/content.zh/docs/core-concept/schema-evolution.md @@ -77,18 +77,29 @@ pipeline: ## 已有目标表的安全 Schema 扩展 -设置 Sink 选项 `existing-table.schema-expansion.mode` 来控制框架在初始 `CreateTableEvent` 遇到已有目标表时的处理方式，默认值为 `OFF`。对于实现了该能力的 Sink，框架可能将缺失的普通非键物理列按 nullable 补充，并安全拓宽普通非键列类型；派生的 DDL 事件会记录在日志中。 +设置 Sink 选项 `existing-table.schema-expansion.mode` 来控制框架在初始 `CreateTableEvent` 遇到已有目标表时的处理方式，默认值为 `DISABLED`。对于实现了该能力的 Sink，框架可能将缺失的普通非键物理列按 nullable 补充，并安全拓宽普通非键列类型；派生的 DDL 事件会记录在日志中。 + +这是由 Flink CDC 框架直接消费的 `sink` 级选项。连接器创建前，框架会从 Sink 配置中移除该选项，因此它不会作为配置属性传递到连接器的 `MetadataApplier`。与 `catalog.properties.*` 不同，该选项不能嵌套在连接器专属属性下；将其写入 `catalog.properties.existing-table.schema-expansion.mode` 不会生效。 + +| 选项 | 作用范围 | 是否传递给连接器 | +|---|---|---| +| `existing-table.schema-expansion.mode` | Flink CDC 框架 | 否 | +| `catalog.properties.*` | 连接器专属 Catalog 选项 | 是 | + +只有其 `MetadataApplier` 实现了 `ExistingTableSchemaExpansionSupport` 的 Sink 才支持扩展。当前实现该能力的 pipeline connector 是 Paimon 和 Fluss。显式配置任何非 `DISABLED` 模式时，如果当前 connector 缺少扩展能力，作业都会立即失败。 + +该能力仅对流式作业生效。当 `execution.runtime-mode` 为 `BATCH` 时，该选项会被忽略并打印告警日志，此时沿用 Sink 原生的 Schema 处理。 | 模式 | 已有目标表 | 目标表不存在 | 失败处理 | |---|---|---|---| -| `OFF` | 不检查、不扩展，保持 Sink 原行为 | Sink 原生建表 | 不适用 | +| `DISABLED` | 不检查、不扩展，保持 Sink 原行为 | Sink 原生建表 | 不适用 | | `CHECK` | 只校验每个上游列能否被目标表容纳，不执行任何 DDL | 作业失败，目标表需由外部创建 | 任何不兼容、读取失败或能力缺失都会以聚合错误使作业失败 | -| `TRY_EXPAND` | 检查并尽力执行安全 DDL，随后读回目标 schema 验证 | Sink 原生建表 | 本机制的失败仅记录日志并交给 Sink 原行为 | +| `TRY_EXPAND` | 检查并尽力执行安全 DDL，随后读回目标 schema 验证 | Sink 原生建表 | 缺少扩展能力时作业失败；本机制的其他失败仅记录日志并交给 Sink 原行为 | | `EXPAND` | 检查并执行安全 DDL，随后读回目标 schema 验证 | Sink 原生建表 | 任何不兼容、DDL 不支持、执行或验证失败都会使作业失败 | -`CHECK` 不执行任何 DDL，因此不受 `include.schema.changes` 和 Sink DDL 能力的影响。它守护初始表状态，**不受 `schema.change.behavior` 控制**（包括 `IGNORE` 和 `EXCEPTION` 也会执行检查）；`TRY_EXPAND` 和 `EXPAND` 在 `schema.change.behavior` 为 `IGNORE` 或 `EXCEPTION` 时跳过框架侧初始处理。注意：`CHECK` 只约束已有目标表的初始处理，后续源端 Schema 变更仍由 `schema.change.behavior` 控制，因此它不是全作业级别的“永不执行 DDL”开关。检查失败时，聚合错误会列出每处差异（表、列、上游类型与目标类型），并附建议的 `ALTER TABLE` 修复语句，可按目标系统方言调整后人工执行。 +`CHECK` 不执行任何 DDL，因此不受 `include.schema.changes` 和 Sink DDL 能力的影响。它守护初始表状态，**不受 `schema.change.behavior` 控制**（包括 `IGNORE` 和 `EXCEPTION` 也会执行检查）；`TRY_EXPAND` 和 `EXPAND` 在 `schema.change.behavior` 为 `IGNORE` 或 `EXCEPTION` 时跳过框架侧初始处理。注意：`CHECK` 只约束已有目标表的初始处理，后续源端 Schema 变更仍由 `schema.change.behavior` 控制，因此它不是全作业级别的“永不执行 DDL”开关。检查失败时，聚合错误会列出每处差异（表、列、上游类型与目标类型），并附方言无关的 `ALTER TABLE` 修复 SQL 模板，需按目标连接器方言调整并人工确认后执行。 -`TRY_EXPAND` 只吞掉本机制自身的失败：它既不屏蔽 Sink 自身 Schema 处理抛出的错误，也不保证扩展失败后所有上游列都能落入目标表。 +`TRY_EXPAND` 仅在确认 connector 支持该能力后吞掉本机制自身的失败：它既不屏蔽 Sink 自身 Schema 处理抛出的错误，也不保证扩展失败后所有上游列都能落入目标表。 ```yaml sink: @@ -96,7 +107,7 @@ sink: existing-table.schema-expansion.mode: "EXPAND" ``` -建议给模式值加引号（例如 `"OFF"`），避免裸写 `OFF` 被 YAML 解析为布尔值。 +模式值必须使用字符串（例如 `"DISABLED"` 或 `"EXPAND"`）；布尔值将不被接受。 ## 按类型配置行为 diff --git a/docs/content/docs/core-concept/schema-evolution.md b/docs/content/docs/core-concept/schema-evolution.md index 6ab021c8f..41e0608e6 100644 --- a/docs/content/docs/core-concept/schema-evolution.md +++ b/docs/content/docs/core-concept/schema-evolution.md @@ -75,18 +75,29 @@ This is useful when your downstream sink is unready for any schema changes, but ## Existing Table Schema Expansion -Set the sink option `existing-table.schema-expansion.mode` to control how the framework handles the initial `CreateTableEvent` when the target table already exists. The default is `OFF`. For sinks that implement this capability, the framework may add missing non-key physical columns as nullable columns and safely widen non-key column types. Derived DDL events are logged. +Set the sink option `existing-table.schema-expansion.mode` to control how the framework handles the initial `CreateTableEvent` when the target table already exists. The default is `DISABLED`. For sinks that implement this capability, the framework may add missing non-key physical columns as nullable columns and safely widen non-key column types. Derived DDL events are logged. + +This is a framework-level option consumed directly by Flink CDC's `sink` block. It is removed from the sink configuration before the connector is created, so it never reaches the connector's `MetadataApplier` as a configuration property. Unlike `catalog.properties.*`, it cannot be nested under connector-specific properties; placing it under `catalog.properties.existing-table.schema-expansion.mode` has no effect. + +| Option | Scope | Passed to connector | +|---|---|---| +| `existing-table.schema-expansion.mode` | Flink CDC framework | No | +| `catalog.properties.*` | Connector-specific catalog options | Yes | + +Expansion is only available for sinks whose `MetadataApplier` implements `ExistingTableSchemaExpansionSupport`. The pipeline connectors currently implementing this capability are Paimon and Fluss. Any other explicitly configured non-`DISABLED` mode fails fast because the required expansion capability is unavailable. + +Expansion applies to streaming pipelines only. When `execution.runtime-mode` is `BATCH`, the option is ignored, a warning is logged, and the sink's original schema handling applies. | Mode | Behavior on an existing target table | Behavior when the target table is missing | Failure handling | |---|---|---|---| -| `OFF` | No check or expansion; the sink's original behavior applies | Sink creates the table | N/A | +| `DISABLED` | No check or expansion; the sink's original behavior applies | Sink creates the table | N/A | | `CHECK` | Validate that every upstream column can be contained by the target table, without issuing any DDL | Fails the job; the table must be created externally | Any incompatibility, read failure, or missing capability fails the job with an aggregated error | -| `TRY_EXPAND` | Check and best-effort apply safe DDL, then verify the result by reading the target schema back | Sink creates the table | Failures of this mechanism are logged and delegated to the sink's original behavior | +| `TRY_EXPAND` | Check and best-effort apply safe DDL, then verify the result by reading the target schema back | Sink creates the table | A missing expansion capability fails the job; other failures of this mechanism are logged and delegated to the sink's original behavior | | `EXPAND` | Check and apply safe DDL, then verify the result by reading the target schema back | Sink creates the table | Any incompatibility, unsupported DDL, execution or verification failure fails the job | -`CHECK` never issues DDL, so it is independent of `include.schema.changes` and of the sink's DDL capabilities. It guards the initial table state and runs **regardless of `schema.change.behavior`** (including `IGNORE` and `EXCEPTION`); `TRY_EXPAND` and `EXPAND` skip the framework-side initial handling when `schema.change.behavior` is `IGNORE` or `EXCEPTION`. Note that `CHECK` only constrains the initial table handling: subsequent source schema changes are still controlled by `schema.chang [...] +`CHECK` never issues DDL, so it is independent of `include.schema.changes` and of the sink's DDL capabilities. It guards the initial table state and runs **regardless of `schema.change.behavior`** (including `IGNORE` and `EXCEPTION`); `TRY_EXPAND` and `EXPAND` skip the framework-side initial handling when `schema.change.behavior` is `IGNORE` or `EXCEPTION`. Note that `CHECK` only constrains the initial table handling: subsequent source schema changes are still controlled by `schema.chang [...] -`TRY_EXPAND` swallows failures of this mechanism only; it neither hides errors from the sink's own schema handling nor guarantees that all upstream columns end up in the target table after a failed expansion. +`TRY_EXPAND` swallows expansion-mechanism failures only after the connector is confirmed to support the capability; it neither hides errors from the sink's own schema handling nor guarantees that all upstream columns end up in the target table after a failed expansion. ```yaml sink: @@ -94,7 +105,7 @@ sink: existing-table.schema-expansion.mode: "EXPAND" ``` -Quote the mode value (for example `"OFF"`) to avoid the bare `OFF` scalar being parsed as a YAML boolean. +The mode value must be quoted as a string (for example `"DISABLED"` or `"EXPAND"`); boolean values are not accepted. ## Per-Event Type Control diff --git a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java index 9692cba80..70d3a4615 100644 --- a/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java +++ b/flink-cdc-cli/src/main/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParser.java @@ -297,23 +297,12 @@ public class YamlPipelineDefinitionParser implements PipelineDefinitionParser { private static ExistingTableSchemaExpansionMode parseExpansionMode(JsonNode optionValue) { if (optionValue == null) { - return ExistingTableSchemaExpansionMode.OFF; - } - // A bare `OFF` YAML scalar is parsed by Jackson YAML into a boolean `false`; accept it as - // OFF for ergonomics. Any other boolean is ambiguous and rejected. - if (optionValue.isBoolean()) { - if (!optionValue.booleanValue()) { - return ExistingTableSchemaExpansionMode.OFF; - } - throw new IllegalArgumentException( - String.format( - "Option \"%s\" must be one of OFF, CHECK, TRY_EXPAND, EXPAND, but was \"%s\".", - EXISTING_TABLE_SCHEMA_EXPANSION_MODE, optionValue)); + return ExistingTableSchemaExpansionMode.DISABLED; } if (!optionValue.isTextual()) { throw new IllegalArgumentException( String.format( - "Option \"%s\" must be one of OFF, CHECK, TRY_EXPAND, EXPAND, but was \"%s\".", + "Option \"%s\" must be one of DISABLED, CHECK, TRY_EXPAND, EXPAND, but was \"%s\".", EXISTING_TABLE_SCHEMA_EXPANSION_MODE, optionValue)); } String value = optionValue.asText(); @@ -324,7 +313,7 @@ public class YamlPipelineDefinitionParser implements PipelineDefinitionParser { } throw new IllegalArgumentException( String.format( - "Option \"%s\" must be one of OFF, CHECK, TRY_EXPAND, EXPAND, but was \"%s\".", + "Option \"%s\" must be one of DISABLED, CHECK, TRY_EXPAND, EXPAND, but was \"%s\".", EXISTING_TABLE_SCHEMA_EXPANSION_MODE, value)); } diff --git a/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java b/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java index 621646e13..ee5d10b74 100644 --- a/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java +++ b/flink-cdc-cli/src/test/java/org/apache/flink/cdc/cli/parser/YamlPipelineDefinitionParserTest.java @@ -534,28 +534,12 @@ class YamlPipelineDefinitionParserTest { } @Test - void testExistingTableSchemaExpansionModeDefaultsToOff() throws Exception { + void testExistingTableSchemaExpansionModeDefaultsToDisabled() throws Exception { PipelineDef defaultPipeline = new YamlPipelineDefinitionParser() .parse("source:\n type: foo\nsink:\n type: bar\n", new Configuration()); assertThat(defaultPipeline.getSink().getExistingTableSchemaExpansionMode()) - .isEqualTo(ExistingTableSchemaExpansionMode.OFF); - } - - @Test - void testBareOffParsesAsExpansionModeOff() throws Exception { - // A bare OFF scalar is parsed into a boolean false by Jackson YAML and must be accepted. - PipelineDef parsed = - new YamlPipelineDefinitionParser() - .parse( - "source:\n" - + " type: foo\n" - + "sink:\n" - + " type: bar\n" - + " existing-table.schema-expansion.mode: OFF\n", - new Configuration()); - assertThat(parsed.getSink().getExistingTableSchemaExpansionMode()) - .isEqualTo(ExistingTableSchemaExpansionMode.OFF); + .isEqualTo(ExistingTableSchemaExpansionMode.DISABLED); } @Test @@ -571,7 +555,7 @@ class YamlPipelineDefinitionParserTest { + " existing-table.schema-expansion.mode: INVALID\n", new Configuration())) .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("OFF, CHECK, TRY_EXPAND, EXPAND"); + .hasMessageContaining("DISABLED, CHECK, TRY_EXPAND, EXPAND"); assertThatThrownBy( () -> @@ -584,7 +568,20 @@ class YamlPipelineDefinitionParserTest { + " existing-table.schema-expansion.mode: true\n", new Configuration())) .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("OFF, CHECK, TRY_EXPAND, EXPAND"); + .hasMessageContaining("DISABLED, CHECK, TRY_EXPAND, EXPAND"); + + assertThatThrownBy( + () -> + new YamlPipelineDefinitionParser() + .parse( + "source:\n" + + " type: foo\n" + + "sink:\n" + + " type: bar\n" + + " existing-table.schema-expansion.mode: false\n", + new Configuration())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("DISABLED, CHECK, TRY_EXPAND, EXPAND"); } @Test diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/ExistingTableSchemaExpansionMode.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/ExistingTableSchemaExpansionMode.java index 68d16dfa9..b24a497be 100644 --- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/ExistingTableSchemaExpansionMode.java +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/ExistingTableSchemaExpansionMode.java @@ -30,7 +30,7 @@ import org.apache.flink.cdc.common.annotation.Experimental; @Experimental public enum ExistingTableSchemaExpansionMode { /** Do not check or expand the existing target table; keep the sink's original behavior. */ - OFF, + DISABLED, /** * Validate that the existing target table can contain every upstream column without executing @@ -43,6 +43,10 @@ public enum ExistingTableSchemaExpansionMode { /** * Check and best-effort expand the existing target table with safe DDL. Failures of this * mechanism are logged and delegated to the sink's original behavior. + * + * <p>A connector that does not implement {@link + * org.apache.flink.cdc.common.sink.ExistingTableSchemaExpansionSupport} is a configuration + * error and fails the job rather than silently degrading. */ TRY_EXPAND, diff --git a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/SinkDef.java b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/SinkDef.java index a39c3e560..d160085c2 100644 --- a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/SinkDef.java +++ b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/definition/SinkDef.java @@ -55,7 +55,7 @@ public class SinkDef { this.config = config; this.includedSchemaEvolutionTypes = Arrays.stream(SchemaChangeEventTypeFamily.ALL).collect(Collectors.toSet()); - this.existingTableSchemaExpansionMode = ExistingTableSchemaExpansionMode.OFF; + this.existingTableSchemaExpansionMode = ExistingTableSchemaExpansionMode.DISABLED; } public SinkDef( @@ -68,7 +68,7 @@ public class SinkDef { name, config, includedSchemaEvolutionTypes, - ExistingTableSchemaExpansionMode.OFF); + ExistingTableSchemaExpansionMode.DISABLED); } public SinkDef( diff --git a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/SchemaOperatorTranslator.java b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/SchemaOperatorTranslator.java index 106c057a7..4a98fc033 100644 --- a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/SchemaOperatorTranslator.java +++ b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/SchemaOperatorTranslator.java @@ -35,6 +35,9 @@ import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.time.Duration; import java.util.ArrayList; import java.util.List; @@ -42,6 +45,9 @@ import java.util.List; /** Translator used to build {@link SchemaOperator} for schema event process. */ @Internal public class SchemaOperatorTranslator { + + private static final Logger LOG = LoggerFactory.getLogger(SchemaOperatorTranslator.class); + private final SchemaChangeBehavior schemaChangeBehavior; private final ExistingTableSchemaExpansionMode existingTableSchemaExpansionMode; private final String schemaOperatorUid; @@ -55,7 +61,7 @@ public class SchemaOperatorTranslator { String timezone) { this( schemaChangeBehavior, - ExistingTableSchemaExpansionMode.OFF, + ExistingTableSchemaExpansionMode.DISABLED, schemaOperatorUid, rpcTimeOut, timezone); @@ -176,6 +182,15 @@ public class SchemaOperatorTranslator { List<RouteDef> routes, RouteMode routeMode, String timezone) { + // Existing target table schema expansion is only wired into the streaming schema + // operators, so a batch pipeline keeps the sink's original schema handling. + if (existingTableSchemaExpansionMode != ExistingTableSchemaExpansionMode.DISABLED) { + LOG.warn( + "Existing target table schema expansion is not supported in batch mode, but " + + "mode {} was configured. The option is ignored and the sink's " + + "original schema handling applies.", + existingTableSchemaExpansionMode); + } List<RouteRule> routingRules = new ArrayList<>(); for (RouteDef route : routes) { routingRules.add( @@ -189,12 +204,7 @@ public class SchemaOperatorTranslator { "SchemaBatchOperator", new EventTypeInfo(), new BatchSchemaOperator( - routingRules, - routeMode, - metadataApplier, - schemaChangeBehavior, - existingTableSchemaExpansionMode, - timezone)); + routingRules, routeMode, metadataApplier, timezone)); stream.uid(schemaOperatorUid).setParallelism(parallelism); return stream; } 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 55e08c2ac..5a2f82dad 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 @@ -264,9 +264,9 @@ public class FlussMetadataApplierTest { ExistingTableSchemaExpander expander = new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT); - expander.expand(new CreateTableEvent(tableId, desiredSchema)); + expander.handleExistingTableCreation(new CreateTableEvent(tableId, desiredSchema)); assertThat(applier.getExistingTableSchema(tableId)).contains(expectedSchema); - expander.expand(new CreateTableEvent(tableId, desiredSchema)); + expander.handleExistingTableCreation(new CreateTableEvent(tableId, desiredSchema)); assertThat(applier.getExistingTableSchema(tableId)).contains(expectedSchema); assertThat( @@ -282,7 +282,8 @@ public class FlussMetadataApplierTest { .primaryKey("id") .build(); - expander.expand(new CreateTableEvent(tableId, schemaWithDifferentExistingColumnType)); + expander.handleExistingTableCreation( + new CreateTableEvent(tableId, schemaWithDifferentExistingColumnType)); assertThat(applier.getExistingTableSchema(tableId)).contains(expectedSchema); } } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplier.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplier.java index 6711a8890..82114ff7a 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplier.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplier.java @@ -126,9 +126,7 @@ public class PaimonMetadataApplier implements MetadataApplier, ExistingTableSche @Override public Optional<Schema> getExistingTableSchema(TableId tableId) { - if (catalog == null) { - catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions); - } + Catalog catalog = getCatalog(); try { Table table = catalog.getTable( @@ -164,18 +162,14 @@ public class PaimonMetadataApplier implements MetadataApplier, ExistingTableSche @Override public boolean isColumnNameCaseSensitive() { - if (catalog == null) { - catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions); - } - return catalog.caseSensitive(); + return getCatalog().caseSensitive(); } @Override public void applySchemaChange(SchemaChangeEvent schemaChangeEvent) throws SchemaEvolveException { - if (catalog == null) { - catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions); - } + // Eagerly initialize the catalog: the visitor methods below use the cached field directly. + getCatalog(); SchemaChangeEventVisitor.voidVisit( schemaChangeEvent, this::applyAddColumn, @@ -192,7 +186,21 @@ public class PaimonMetadataApplier implements MetadataApplier, ExistingTableSche public void close() throws Exception { if (catalog != null) { catalog.close(); + catalog = null; + } + } + + /** + * Lazily creates and caches the Paimon {@link Catalog}. All catalog access must go through this + * method so the lifecycle stays centralized: the catalog is created once on first use and + * released by {@link #close()}, which also resets the cached instance so the applier stays + * reusable after close. + */ + private Catalog getCatalog() { + if (catalog == null) { + catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions); } + return catalog; } private void applyCreateTable(CreateTableEvent event) throws SchemaEvolveException { diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java index ec62a846a..5c4665f1e 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/PaimonMetadataApplierTest.java @@ -303,9 +303,9 @@ class PaimonMetadataApplierTest { new ExistingTableSchemaExpander( metadataApplier, metadataApplier, SchemaChangeBehavior.LENIENT); - expander.expand(createTableEvent); + expander.handleExistingTableCreation(createTableEvent); metadataApplier.applySchemaChange(createTableEvent); - expander.expand(createTableEvent); + expander.handleExistingTableCreation(createTableEvent); Assertions.assertThat(metadataApplier.getExistingTableSchema(tableId)) .get() .satisfies( @@ -788,4 +788,39 @@ class PaimonMetadataApplierTest { Assertions.assertThat(table).isNotNull(); } + + @Test + void testCloseIsIdempotentAndRecreatesCatalogOnReuse() + throws Catalog.DatabaseNotEmptyException, Catalog.DatabaseNotExistException, Exception { + initialize("filesystem"); + PaimonMetadataApplier metadataApplier = new PaimonMetadataApplier(catalogOptions); + + // Trigger lazy catalog creation. + metadataApplier.applySchemaChange( + new CreateTableEvent( + TableId.parse("test.close_idempotent"), + Schema.newBuilder() + .physicalColumn( + "id", org.apache.flink.cdc.common.types.DataTypes.INT()) + .primaryKey("id") + .build())); + + // close() must be idempotent and must not fail when called repeatedly. + metadataApplier.close(); + metadataApplier.close(); + + // After close the cached catalog is reset; the applier stays reusable and recreates it. + metadataApplier.applySchemaChange( + new CreateTableEvent( + TableId.parse("test.close_reuse"), + Schema.newBuilder() + .physicalColumn( + "id", org.apache.flink.cdc.common.types.DataTypes.INT()) + .primaryKey("id") + .build())); + + Assertions.assertThat(catalog.getTable(Identifier.fromString("test.close_reuse"))) + .isNotNull(); + metadataApplier.close(); + } } diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/ExistingTableSchemaExpansionE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/ExistingTableSchemaExpansionE2eITCase.java index aba76229f..a49ddb39f 100644 --- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/ExistingTableSchemaExpansionE2eITCase.java +++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/stage2/ExistingTableSchemaExpansionE2eITCase.java @@ -21,6 +21,7 @@ import org.apache.flink.api.common.JobStatus; import org.apache.flink.cdc.common.test.utils.TestUtils; import org.apache.flink.cdc.connectors.mysql.testutils.UniqueDatabase; import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment; +import org.apache.flink.runtime.client.JobStatusMessage; import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap; @@ -48,14 +49,15 @@ import java.util.Collections; import java.util.List; import java.util.Objects; import java.util.UUID; +import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import java.util.stream.Stream; /** * End-to-end tests verifying that the {@code existing-table.schema-expansion.mode} sink option * reaches the schema expander through the whole wiring path (YAML option, Composer, schema operator - * factory, schema registry or batch schema operator, metadata applier) in all three execution - * topologies: regular streaming, batch, and distributed streaming. + * factory, schema registry, metadata applier) in the streaming and distributed execution + * topologies. * * <p>Every case pre-creates the target table with a strict subset of the source columns, so a flag * lost on any wiring hop leaves the missing columns unwritten and fails the assertions. @@ -218,52 +220,6 @@ class ExistingTableSchemaExpansionE2eITCase extends PipelineTestEnvironment { validatePaimonSinkResult(warehouse, database, "customers", expectedCustomersRows()); } - /** - * Batch runtime mode makes the Composer wire a batch schema operator instead of the - * coordinator-based one. MySQL accepts the {@code snapshot} startup mode only in batch - * pipelines, which also bounds the source so that the job reaches a terminal state. - */ - @Test - void testBatchPath() throws Exception { - String warehouse = sharedVolume.toString() + "/" + "paimon_" + UUID.randomUUID(); - String database = inventoryDatabase.getDatabaseName(); - preCreatePaimonProductsTable(warehouse, database); - - String pipelineJob = - String.format( - "source:\n" - + " type: mysql\n" - + " hostname: mysql\n" - + " port: 3306\n" - + " username: %s\n" - + " password: %s\n" - + " tables: %s.\\.*\n" - + " server-id: 5400-5404\n" - + " server-time-zone: UTC\n" - + " scan.startup.mode: snapshot\n" - + "\n" - + "sink:\n" - + " type: paimon\n" - + " catalog.properties.warehouse: %s\n" - + " catalog.properties.metastore: filesystem\n" - + " catalog.properties.cache-enabled: false\n" - + " existing-table.schema-expansion.mode: EXPAND\n" - + "\n" - + "pipeline:\n" - + " schema.change.behavior: evolve\n" - + " parallelism: %s\n" - + " execution.runtime-mode: BATCH", - MYSQL_TEST_USER, MYSQL_TEST_PASSWORD, database, warehouse, parallelism); - Path paimonCdcConnector = TestUtils.getResource("paimon-cdc-pipeline-connector.jar"); - Path hadoopJar = TestUtils.getResource("flink-shade-hadoop.jar"); - submitPipelineJob(pipelineJob, paimonCdcConnector, hadoopJar); - waitUntilJobFinished(EXPANSION_TESTCASE_TIMEOUT); - LOG.info("Batch pipeline job has finished"); - - validatePaimonSinkResult(warehouse, database, "products", expectedProductsRows()); - validatePaimonSinkResult(warehouse, database, "customers", expectedCustomersRows()); - } - /** * Fluss reports parallel metadata, so the Composer wires the distributed schema operator backed * by the distributed schema coordinator. That topology only accepts {@code LENIENT}, {@code @@ -637,9 +593,21 @@ class ExistingTableSchemaExpansionE2eITCase extends PipelineTestEnvironment { * table holding a strict subset of the source columns. */ private void prepareFlussTables(String sourceDatabase, String sinkDatabase) throws Exception { + // copyJarToFlinkLib only reaches the JobManager container and the SQL client has no --jar + // option, so the TaskManager would lack the Fluss sink classes (such as + // FlinkStreamPartitioner) that the insert job needs. Publish the jar on the shared volume + // and hand it to the job through pipeline.jars, which ships it to every node. (Pointing + // pipeline.jars straight at /opt/flink/lib does not work: that directory is not part of the + // TaskManager image, and the jar never reaches the writers.) + String flussJar = String.format("fluss-flink-%s.jar", flinkVersion); + String containerJar = sharedVolume.toString() + "/" + flussJar; + jobManager.copyFileToContainer( + MountableFile.forHostPath(TestUtils.getResource(flussJar)), containerJar); + String sql = String.format( - "SET 'execution.runtime-mode' = 'batch';\n" + "SET 'pipeline.jars' = 'file://%s';\n" + + "SET 'execution.runtime-mode' = 'batch';\n" + "CREATE CATALOG fluss_catalog WITH (\n" + " 'type' = 'fluss',\n" + " 'bootstrap.servers' = 'coordinator-server:9123',\n" @@ -665,15 +633,22 @@ class ExistingTableSchemaExpansionE2eITCase extends PipelineTestEnvironment { + " (101, 'One', 'Alice'),\n" + " (102, 'Two', 'Bob'),\n" + " (103, 'Three', 'Cecily');", - sourceDatabase, sinkDatabase, sourceDatabase, sinkDatabase, sourceDatabase); + containerJar, + sourceDatabase, + sinkDatabase, + sourceDatabase, + sinkDatabase, + sourceDatabase); executeFlussSql(sql, "prepare_fluss"); - // The SQL client may return before the inserted rows are visible to new Fluss readers. - // Verify that the records can actually be read before starting the CDC pipeline. + // The SQL client only submits the batch insert and returns, so the rows may not be visible + // to new Fluss readers yet. Verify that they can actually be read before starting the + // pipeline, and fail fast if the detached insert job died. List<String> expectedRows = Arrays.asList("101, One, Alice", "102, Two, Bob", "103, Three, Cecily"); long deadline = System.currentTimeMillis() + Duration.ofMinutes(2).toMillis(); while (System.currentTimeMillis() < deadline) { + assertFlussSourceInsertNotFailed(); try { List<String> actualRows = fetchFlussTableRows(sourceDatabase, "products", 3); if (actualRows.containsAll(expectedRows) @@ -689,6 +664,23 @@ class ExistingTableSchemaExpansionE2eITCase extends PipelineTestEnvironment { "Source Fluss table was not populated with the expected rows within 2 minutes."); } + /** + * The SQL client submits batch DML and returns right away, so a job that dies afterwards stays + * invisible to {@code execInContainer}'s exit code. Surface it here instead of letting the + * read-back loop below time out with a misleading message. + */ + private void assertFlussSourceInsertNotFailed() throws Exception { + for (JobStatusMessage job : getRestClusterClient().listJobs().get(10, TimeUnit.SECONDS)) { + if (job.getJobState() == JobStatus.FAILED) { + throw new IllegalStateException( + String.format( + "Populating the Fluss source table failed: job %s (%s) ended with " + + "status FAILED.", + job.getJobName(), job.getJobId())); + } + } + } + /** Runs a script against the Paimon catalog, whose connector must be passed explicitly. */ private void executePaimonSql(String sql, String scriptName) throws Exception { String containerSqlPath = sharedVolume.toString() + "/" + scriptName + ".sql"; @@ -703,6 +695,13 @@ class ExistingTableSchemaExpansionE2eITCase extends PipelineTestEnvironment { "-f", containerSqlPath); checkSqlResult(result, scriptName); + // The SQL client creates catalog artifacts (databases, tables, schema files) as the + // JobManager container's user, while the pipeline's sink and schema operator run in the + // TaskManager container and must create or alter files under the same shared volume. + // Widen permissions so the TaskManager can write those artifacts atomically; otherwise + // applyCreateTable/applyAddColumn fails with "Mkdirs failed" or "Permission denied" and + // the job dies without writing any record. + runInContainerAsRoot(jobManager, "chmod", "-R", "a+rwX", sharedVolume.toString()); } /** Runs a script against the Fluss catalog, whose connector already sits in Flink's lib. */ diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java index 1e8ee4b37..a91044997 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpander.java @@ -98,11 +98,17 @@ public class ExistingTableSchemaExpander { /** * Handles the initial {@link CreateTableEvent} for an existing target table. * - * @return whether the caller should proceed to apply the original {@link CreateTableEvent} to - * the sink. {@code CHECK} mode returns {@code false} after a successful check so that no - * external DDL is issued by the pipeline. + * <p>This is a two-phase contract: the expander first tries to check/expand the existing target + * table, then signals whether the sink should still apply the original {@link + * CreateTableEvent}. + * + * @return {@code true} if the caller should proceed to apply the original {@link + * CreateTableEvent} to the sink; {@code false} means the framework has fully handled this + * event and the original {@link CreateTableEvent} must not be applied. {@code CHECK} mode + * returns {@code false} after a successful check so that no external DDL is issued by the + * pipeline. */ - public boolean expand(CreateTableEvent createTableEvent) { + public boolean handleExistingTableCreation(CreateTableEvent createTableEvent) { if (mode == ExistingTableSchemaExpansionMode.CHECK) { // CHECK guards the initial table state and runs regardless of schema.change.behavior. checkCompatibility(createTableEvent); @@ -120,7 +126,7 @@ public class ExistingTableSchemaExpander { case EXPAND: expandStrictly(createTableEvent); return true; - case OFF: + case DISABLED: default: return true; } @@ -150,35 +156,70 @@ public class ExistingTableSchemaExpander { } private void tryExpand(CreateTableEvent createTableEvent) { + // Phase 1 - probe the target table. Failures here are "cannot expand safely" cases: the + // connector lacks DDL capability, or the target table does not exist yet. Delegating to the + // sink's original behavior is safe because no derived DDL has been issued. + if (!supportsAnyExpansionDdl()) { + LOG.info( + "Neither ADD_COLUMN nor ALTER_COLUMN_TYPE is enabled or supported for target table {}. Delegating schema handling to the sink.", + createTableEvent.tableId()); + return; + } + Optional<Schema> targetSchema; try { - if (!supportsAnyExpansionDdl()) { - LOG.info( - "Neither ADD_COLUMN nor ALTER_COLUMN_TYPE is enabled or supported for target table {}. Delegating schema handling to the sink.", - createTableEvent.tableId()); - return; - } - Optional<Schema> targetSchema = queryTargetSchema(createTableEvent.tableId()); - if (!targetSchema.isPresent()) { - LOG.info( - "Target table {} does not exist. Delegating table creation to the sink.", - createTableEvent.tableId()); - return; - } - ExpansionPlan plan = analyze(createTableEvent, targetSchema.get()); - for (String incompatibility : plan.incompatibilities) { - LOG.warn( - "Target table {} has an unsupported difference: {}. Delegating it to the sink.", - createTableEvent.tableId(), - incompatibility); - } - applyPlan(createTableEvent, plan); - verifyExpansion(createTableEvent, plan, false); + targetSchema = queryTargetSchema(createTableEvent.tableId()); } catch (Exception e) { LOG.warn( - "Best-effort schema expansion failed for existing target table {}. Delegating schema handling to the sink.", + "Failed to read the existing target table {} before expansion. Delegating schema handling to the sink.", createTableEvent.tableId(), e); + return; + } + if (!targetSchema.isPresent()) { + LOG.info( + "Target table {} does not exist. Delegating table creation to the sink.", + createTableEvent.tableId()); + return; } + + ExpansionPlan plan = analyze(createTableEvent, targetSchema.get()); + for (String incompatibility : plan.incompatibilities) { + LOG.warn( + "Target table {} has an unsupported difference: {}. Delegating it to the sink.", + createTableEvent.tableId(), + incompatibility); + } + + // The plan needs repair DDL, but the sink cannot perform the specific required event type. + // This is a capability gap, not a transient failure, so delegate to the sink instead of + // failing the job (mirrors applyPlan's capability checks, which throw for EXPAND mode). + if (!plan.columnsToAdd.isEmpty() + && !supportsSchemaEvolutionType(SchemaChangeEventType.ADD_COLUMN)) { + LOG.warn( + "Target table {} is missing columns {}, but ADD_COLUMN is not enabled or supported by the sink. Delegating schema handling to the sink.", + createTableEvent.tableId(), + getColumnNames(plan.columnsToAdd)); + return; + } + if (!plan.columnsToWiden.isEmpty() + && !supportsSchemaEvolutionType(SchemaChangeEventType.ALTER_COLUMN_TYPE)) { + LOG.warn( + "Target table {} has narrow columns {}, but ALTER_COLUMN_TYPE is not enabled or supported by the sink. Delegating schema handling to the sink.", + createTableEvent.tableId(), + plan.columnsToWiden.keySet()); + return; + } + + // Phase 2 - apply and verify. At this point the expander has identified supportable + // differences and is about to issue derived DDL. A failure here (transient network/catalog + // error, DDL execution failure, or read-back verification failure) must propagate so the + // job fails over and the idempotent expander retries; otherwise the original + // CreateTableEvent + // would be applied to an existing table (typically a no-op) and the missing columns would + // be + // silently dropped forever. + applyPlan(createTableEvent, plan); + verifyExpansion(createTableEvent, plan, false); } private void expandStrictly(CreateTableEvent createTableEvent) { @@ -231,7 +272,7 @@ public class ExistingTableSchemaExpander { if (!repairSuggestions.isEmpty()) { message += String.format( - "\nSuggested repair statements (adjust to the target system's DDL dialect):%s", + "\nSuggested repair SQL templates (review and adjust to the target connector's DDL dialect before execution):%s", repairSuggestions); } return new SchemaEvolveException(createTableEvent, message); diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaRegistry.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaRegistry.java index d8ea2f345..35fceefd6 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaRegistry.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/common/SchemaRegistry.java @@ -156,7 +156,7 @@ public abstract class SchemaRegistry implements OperatorCoordinator, Coordinatio this.router = new TableIdRouter(routingRules, routeMode); // TRY_EXPAND/EXPAND never run under IGNORE/EXCEPTION, so skip their initialization there. // CHECK guards the initial table state and initializes regardless of the behavior. - if (existingTableSchemaExpansionMode != ExistingTableSchemaExpansionMode.OFF + if (existingTableSchemaExpansionMode != ExistingTableSchemaExpansionMode.DISABLED && (existingTableSchemaExpansionMode == ExistingTableSchemaExpansionMode.CHECK || (behavior != SchemaChangeBehavior.IGNORE && behavior != SchemaChangeBehavior.EXCEPTION))) { @@ -184,11 +184,7 @@ public abstract class SchemaRegistry implements OperatorCoordinator, Coordinatio String.format( "Existing target table schema expansion is enabled with mode %s, but MetadataApplier %s does not support it.", existingTableSchemaExpansionMode, metadataApplier.getClass().getName()); - if (existingTableSchemaExpansionMode == ExistingTableSchemaExpansionMode.TRY_EXPAND) { - LOG.warn("{}. The sink's original schema handling will be used.", message); - } else { - throw new FlinkRuntimeException(message); - } + throw new FlinkRuntimeException(message); } private void handleExpansionInitFailure(Exception cause) { @@ -196,11 +192,7 @@ public abstract class SchemaRegistry implements OperatorCoordinator, Coordinatio String.format( "Failed to initialize existing target table schema expansion with mode %s.", existingTableSchemaExpansionMode); - if (existingTableSchemaExpansionMode == ExistingTableSchemaExpansionMode.TRY_EXPAND) { - LOG.warn("{}. The sink's original schema handling will be used.", message, cause); - } else { - throw new FlinkRuntimeException(message, cause); - } + throw new FlinkRuntimeException(message, cause); } /** @@ -211,7 +203,8 @@ public abstract class SchemaRegistry implements OperatorCoordinator, Coordinatio */ protected boolean expandExistingTableSchemaIfNeeded(SchemaChangeEvent schemaChangeEvent) { if (existingTableSchemaExpander != null && schemaChangeEvent instanceof CreateTableEvent) { - return existingTableSchemaExpander.expand((CreateTableEvent) schemaChangeEvent); + return existingTableSchemaExpander.handleExistingTableCreation( + (CreateTableEvent) schemaChangeEvent); } return true; } diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java index 0f3822355..732f3ac39 100755 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java @@ -123,7 +123,7 @@ public class SchemaCoordinator extends SchemaRegistry { routingRules, routeMode, schemaChangeBehavior, - ExistingTableSchemaExpansionMode.OFF, + ExistingTableSchemaExpansionMode.DISABLED, rpcTimeout); } diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinatorProvider.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinatorProvider.java index 2834c0fd2..3dd95930f 100755 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinatorProvider.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinatorProvider.java @@ -61,7 +61,7 @@ public class SchemaCoordinatorProvider implements OperatorCoordinator.Provider { routingRules, routeMode, schemaChangeBehavior, - ExistingTableSchemaExpansionMode.OFF, + ExistingTableSchemaExpansionMode.DISABLED, rpcTimeout); } diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaOperatorFactory.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaOperatorFactory.java index a17f3da72..1169bb469 100755 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaOperatorFactory.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaOperatorFactory.java @@ -61,7 +61,7 @@ public class SchemaOperatorFactory extends SimpleOperatorFactory<Event> routeMode, rpcTimeout, schemaChangeBehavior, - ExistingTableSchemaExpansionMode.OFF, + ExistingTableSchemaExpansionMode.DISABLED, timezone); } diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/BatchSchemaOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/BatchSchemaOperator.java index 1545d1950..e830b1f2a 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/BatchSchemaOperator.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/BatchSchemaOperator.java @@ -23,16 +23,13 @@ import org.apache.flink.cdc.common.event.DataChangeEvent; import org.apache.flink.cdc.common.event.Event; import org.apache.flink.cdc.common.event.SchemaChangeEvent; import org.apache.flink.cdc.common.event.TableId; -import org.apache.flink.cdc.common.pipeline.ExistingTableSchemaExpansionMode; import org.apache.flink.cdc.common.pipeline.RouteMode; import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior; import org.apache.flink.cdc.common.route.RouteRule; import org.apache.flink.cdc.common.route.TableIdRouter; import org.apache.flink.cdc.common.schema.Schema; -import org.apache.flink.cdc.common.sink.ExistingTableSchemaExpansionSupport; import org.apache.flink.cdc.common.sink.MetadataApplier; import org.apache.flink.cdc.runtime.operators.AbstractStreamOperatorAdapter; -import org.apache.flink.cdc.runtime.operators.schema.common.ExistingTableSchemaExpander; import org.apache.flink.cdc.runtime.operators.schema.common.SchemaDerivator; import org.apache.flink.cdc.runtime.operators.schema.common.SchemaManager; import org.apache.flink.streaming.api.graph.StreamConfig; @@ -41,7 +38,6 @@ import org.apache.flink.streaming.api.operators.OneInputStreamOperator; import org.apache.flink.streaming.api.operators.Output; import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; import org.apache.flink.streaming.runtime.tasks.StreamTask; -import org.apache.flink.util.FlinkRuntimeException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -50,7 +46,6 @@ import java.io.Serializable; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Optional; import java.util.stream.Collectors; /** The operator will apply create table event and router mapper in batch mode. */ @@ -65,15 +60,12 @@ public class BatchSchemaOperator extends AbstractStreamOperatorAdapter<Event> private final String timezone; private final List<RouteRule> routingRules; private final RouteMode routeMode; - private final SchemaChangeBehavior schemaChangeBehavior; - private final ExistingTableSchemaExpansionMode existingTableSchemaExpansionMode; // Transient fields that are set during open() private transient volatile Map<TableId, Schema> originalSchemaMap; private transient volatile Map<TableId, Schema> evolvedSchemaMap; private transient TableIdRouter router; private transient SchemaDerivator derivator; - private transient ExistingTableSchemaExpander existingTableSchemaExpander; protected transient SchemaManager schemaManager; protected MetadataApplier metadataApplier; private boolean alreadyMergedCreateTableTables = false; @@ -83,29 +75,11 @@ public class BatchSchemaOperator extends AbstractStreamOperatorAdapter<Event> RouteMode routeMode, MetadataApplier metadataApplier, String timezone) { - this( - routingRules, - routeMode, - metadataApplier, - SchemaChangeBehavior.IGNORE, - ExistingTableSchemaExpansionMode.OFF, - timezone); - } - - public BatchSchemaOperator( - List<RouteRule> routingRules, - RouteMode routeMode, - MetadataApplier metadataApplier, - SchemaChangeBehavior schemaChangeBehavior, - ExistingTableSchemaExpansionMode existingTableSchemaExpansionMode, - String timezone) { this.chainingStrategy = ChainingStrategy.ALWAYS; this.timezone = timezone; this.routingRules = routingRules; this.routeMode = routeMode; this.metadataApplier = metadataApplier; - this.schemaChangeBehavior = schemaChangeBehavior; - this.existingTableSchemaExpansionMode = existingTableSchemaExpansionMode; } @Override @@ -124,48 +98,6 @@ public class BatchSchemaOperator extends AbstractStreamOperatorAdapter<Event> this.router = new TableIdRouter(routingRules, routeMode); this.derivator = new SchemaDerivator(); this.schemaManager = new SchemaManager(SchemaChangeBehavior.IGNORE); - // TRY_EXPAND/EXPAND never run under IGNORE/EXCEPTION, so skip their initialization there. - // CHECK guards the initial table state and initializes regardless of the behavior. - if (existingTableSchemaExpansionMode != ExistingTableSchemaExpansionMode.OFF - && (existingTableSchemaExpansionMode == ExistingTableSchemaExpansionMode.CHECK - || (schemaChangeBehavior != SchemaChangeBehavior.IGNORE - && schemaChangeBehavior != SchemaChangeBehavior.EXCEPTION))) { - try { - Optional<ExistingTableSchemaExpansionSupport> supportOptional = - metadataApplier.getExistingTableSchemaExpansionSupport(); - if (supportOptional.isPresent()) { - this.existingTableSchemaExpander = - new ExistingTableSchemaExpander( - metadataApplier, - supportOptional.get(), - schemaChangeBehavior, - existingTableSchemaExpansionMode); - } else { - String message = - String.format( - "Existing target table schema expansion is enabled with mode %s, but MetadataApplier %s does not support it.", - existingTableSchemaExpansionMode, - metadataApplier.getClass().getName()); - if (existingTableSchemaExpansionMode - == ExistingTableSchemaExpansionMode.TRY_EXPAND) { - LOG.warn("{}. The sink's original schema handling will be used.", message); - } else { - throw new FlinkRuntimeException(message); - } - } - } catch (Exception e) { - String message = - String.format( - "Failed to initialize existing target table schema expansion with mode %s.", - existingTableSchemaExpansionMode); - if (existingTableSchemaExpansionMode - == ExistingTableSchemaExpansionMode.TRY_EXPAND) { - LOG.warn("{}. The sink's original schema handling will be used.", message, e); - } else { - throw new FlinkRuntimeException(message, e); - } - } - } } /** @@ -237,15 +169,8 @@ public class BatchSchemaOperator extends AbstractStreamOperatorAdapter<Event> } private boolean applyAndUpdateEvolvedSchemaChange(SchemaChangeEvent schemaChangeEvent) { - boolean shouldApplyOriginalCreateTable = true; - if (existingTableSchemaExpander != null && schemaChangeEvent instanceof CreateTableEvent) { - shouldApplyOriginalCreateTable = - existingTableSchemaExpander.expand((CreateTableEvent) schemaChangeEvent); - } try { - if (shouldApplyOriginalCreateTable) { - metadataApplier.applySchemaChange(schemaChangeEvent); - } + metadataApplier.applySchemaChange(schemaChangeEvent); schemaManager.applyEvolvedSchemaChange(schemaChangeEvent); LOG.info( "Successfully applied schema change event {} to external system.", diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java index e82117748..c50fbc894 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinator.java @@ -114,7 +114,7 @@ public class SchemaCoordinator extends SchemaRegistry { routes, routeMode, schemaChangeBehavior, - ExistingTableSchemaExpansionMode.OFF, + ExistingTableSchemaExpansionMode.DISABLED, rpcTimeout); } diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinatorProvider.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinatorProvider.java index 5c3df0450..2b9abbc68 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinatorProvider.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinatorProvider.java @@ -61,7 +61,7 @@ public class SchemaCoordinatorProvider implements OperatorCoordinator.Provider { routingRules, routeMode, schemaChangeBehavior, - ExistingTableSchemaExpansionMode.OFF, + ExistingTableSchemaExpansionMode.DISABLED, rpcTimeout); } diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaOperatorFactory.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaOperatorFactory.java index c57d159e2..a023ddd92 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaOperatorFactory.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaOperatorFactory.java @@ -60,7 +60,7 @@ public class SchemaOperatorFactory extends SimpleOperatorFactory<Event> routeMode, rpcTimeout, schemaChangeBehavior, - ExistingTableSchemaExpansionMode.OFF, + ExistingTableSchemaExpansionMode.DISABLED, timezone); } diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpanderTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpanderTest.java index 7584d9307..e0f30eba4 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpanderTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/common/ExistingTableSchemaExpanderTest.java @@ -69,7 +69,7 @@ class ExistingTableSchemaExpanderTest { new TestingExistingTableSchemaExpansionSupport(null); new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand(eventWithMissingColumn()); + .handleExistingTableCreation(eventWithMissingColumn()); assertThat(applier.queryCalls).isOne(); assertThat(applier.appliedEvents).isEmpty(); } @@ -86,8 +86,8 @@ class ExistingTableSchemaExpanderTest { Column.physicalColumn("id", DataTypes.INT()), Column.physicalColumn("name", DataTypes.STRING().notNull())); - expander.expand(event); - expander.expand(event); + expander.handleExistingTableCreation(event); + expander.handleExistingTableCreation(event); assertThat(applier.appliedEvents).hasSize(1); AddColumnEvent addColumnEvent = (AddColumnEvent) applier.appliedEvents.get(0); @@ -111,7 +111,7 @@ class ExistingTableSchemaExpanderTest { .build(); new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand(new CreateTableEvent(TABLE_ID, pipelineSchema)); + .handleExistingTableCreation(new CreateTableEvent(TABLE_ID, pipelineSchema)); assertThat(applier.appliedEvents).isEmpty(); } @@ -122,7 +122,8 @@ class ExistingTableSchemaExpanderTest { schema(Column.physicalColumn("value", DataTypes.BIGINT()))); new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand(createTableEvent(Column.physicalColumn("value", DataTypes.INT()))); + .handleExistingTableCreation( + createTableEvent(Column.physicalColumn("value", DataTypes.INT()))); assertThat(applier.appliedEvents).isEmpty(); } @@ -136,8 +137,8 @@ class ExistingTableSchemaExpanderTest { CreateTableEvent event = createTableEvent(Column.physicalColumn("value", DataTypes.BIGINT())); - expander.expand(event); - expander.expand(event); + expander.handleExistingTableCreation(event); + expander.handleExistingTableCreation(event); assertThat(applier.appliedEvents).singleElement().isInstanceOf(AlterColumnTypeEvent.class); AlterColumnTypeEvent alterColumnTypeEvent = @@ -164,7 +165,7 @@ class ExistingTableSchemaExpanderTest { new TestingExistingTableSchemaExpansionSupport(targetSchema); new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand(new CreateTableEvent(TABLE_ID, pipelineSchema)); + .handleExistingTableCreation(new CreateTableEvent(TABLE_ID, pipelineSchema)); assertThat(applier.appliedEvents).isEmpty(); } @@ -175,7 +176,7 @@ class ExistingTableSchemaExpanderTest { schema(Column.physicalColumn("amount", DataTypes.DECIMAL(12, 2)))); new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand( + .handleExistingTableCreation( createTableEvent( Column.physicalColumn("amount", DataTypes.DECIMAL(12, 4)))); @@ -193,7 +194,8 @@ class ExistingTableSchemaExpanderTest { schema(Column.physicalColumn("value", targetType))); new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand(createTableEvent(Column.physicalColumn("value", pipelineType))); + .handleExistingTableCreation( + createTableEvent(Column.physicalColumn("value", pipelineType))); AlterColumnTypeEvent event = (AlterColumnTypeEvent) applier.appliedEvents.get(0); assertThat(event.getTypeMapping()).containsExactly(Map.entry("value", expectedWidenedType)); @@ -207,7 +209,8 @@ class ExistingTableSchemaExpanderTest { schema(Column.physicalColumn("value", targetType))); new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand(createTableEvent(Column.physicalColumn("value", pipelineType))); + .handleExistingTableCreation( + createTableEvent(Column.physicalColumn("value", pipelineType))); assertThat(applier.appliedEvents).isEmpty(); } @@ -218,7 +221,8 @@ class ExistingTableSchemaExpanderTest { schema(Column.physicalColumn("value", DataTypes.BOOLEAN()))); new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand(createTableEvent(Column.physicalColumn("value", DataTypes.TIMESTAMP()))); + .handleExistingTableCreation( + createTableEvent(Column.physicalColumn("value", DataTypes.TIMESTAMP()))); assertThat(applier.appliedEvents).isEmpty(); } @@ -238,7 +242,7 @@ class ExistingTableSchemaExpanderTest { }; new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand( + .handleExistingTableCreation( createTableEvent( Column.physicalColumn("event_time", DataTypes.TIMESTAMP(9)))); assertThat(applier.appliedEvents).isEmpty(); @@ -261,7 +265,8 @@ class ExistingTableSchemaExpanderTest { : type; new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand(createTableEvent(Column.physicalColumn("id", DataTypes.BIGINT()))); + .handleExistingTableCreation( + createTableEvent(Column.physicalColumn("id", DataTypes.BIGINT()))); assertThat(applier.appliedEvents).isEmpty(); assertThat(applier.queryCalls).isOne(); @@ -277,7 +282,7 @@ class ExistingTableSchemaExpanderTest { applier.normalizationFailureColumns.add("broken"); new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand( + .handleExistingTableCreation( createTableEvent( Column.physicalColumn("broken", DataTypes.BIGINT()), Column.physicalColumn("id", DataTypes.BIGINT()), @@ -300,7 +305,8 @@ class ExistingTableSchemaExpanderTest { applier.columnNameCaseSensitive = false; new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand(createTableEvent(Column.physicalColumn("value", DataTypes.BIGINT()))); + .handleExistingTableCreation( + createTableEvent(Column.physicalColumn("value", DataTypes.BIGINT()))); AlterColumnTypeEvent alterColumnTypeEvent = (AlterColumnTypeEvent) applier.appliedEvents.get(0); @@ -315,7 +321,8 @@ class ExistingTableSchemaExpanderTest { schema(Column.physicalColumn("VALUE", DataTypes.INT()))); new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand(createTableEvent(Column.physicalColumn("value", DataTypes.BIGINT()))); + .handleExistingTableCreation( + createTableEvent(Column.physicalColumn("value", DataTypes.BIGINT()))); assertThat(applier.appliedEvents).singleElement().isInstanceOf(AddColumnEvent.class); } @@ -329,7 +336,8 @@ class ExistingTableSchemaExpanderTest { applier.columnNameCaseSensitive = false; new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand(createTableEvent(Column.physicalColumn("value", DataTypes.BIGINT()))); + .handleExistingTableCreation( + createTableEvent(Column.physicalColumn("value", DataTypes.BIGINT()))); assertThat(applier.appliedEvents).isEmpty(); } @@ -340,7 +348,8 @@ class ExistingTableSchemaExpanderTest { schema(Column.physicalColumn("value", DataTypes.BIGINT().notNull()))); new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand(createTableEvent(Column.physicalColumn("value", DataTypes.BIGINT()))); + .handleExistingTableCreation( + createTableEvent(Column.physicalColumn("value", DataTypes.BIGINT()))); assertThat(applier.appliedEvents).isEmpty(); } @@ -354,7 +363,7 @@ class ExistingTableSchemaExpanderTest { new ExistingTableSchemaExpander( addUnavailableApplier, addUnavailableApplier, SchemaChangeBehavior.LENIENT) - .expand(eventWithMissingColumn()); + .handleExistingTableCreation(eventWithMissingColumn()); assertThat(addUnavailableApplier.appliedEvents).isEmpty(); assertThat(addUnavailableApplier.queryCalls).isZero(); @@ -368,7 +377,8 @@ class ExistingTableSchemaExpanderTest { alterUnavailableApplier, alterUnavailableApplier, SchemaChangeBehavior.LENIENT) - .expand(createTableEvent(Column.physicalColumn("id", DataTypes.BIGINT()))); + .handleExistingTableCreation( + createTableEvent(Column.physicalColumn("id", DataTypes.BIGINT()))); assertThat(alterUnavailableApplier.appliedEvents).isEmpty(); assertThat(alterUnavailableApplier.queryCalls).isOne(); } @@ -381,7 +391,7 @@ class ExistingTableSchemaExpanderTest { applier.supportedEventTypes = Collections.singleton(SchemaChangeEventType.CREATE_TABLE); new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand(eventWithMissingColumn()); + .handleExistingTableCreation(eventWithMissingColumn()); assertThat(applier.queryCalls).isZero(); assertThat(applier.appliedEvents).isEmpty(); @@ -401,7 +411,8 @@ class ExistingTableSchemaExpanderTest { Column.physicalColumn("id", DataTypes.BIGINT()), Column.physicalColumn("name", DataTypes.STRING())); - new ExistingTableSchemaExpander(applier, applier, schemaChangeBehavior).expand(event); + new ExistingTableSchemaExpander(applier, applier, schemaChangeBehavior) + .handleExistingTableCreation(event); assertThat(applier.appliedEvents).isEmpty(); assertThat(applier.queryCalls).isZero(); } @@ -421,7 +432,7 @@ class ExistingTableSchemaExpanderTest { normalizationFailureApplier, normalizationFailureApplier, SchemaChangeBehavior.LENIENT) - .expand( + .handleExistingTableCreation( createTableEvent( Column.physicalColumn( "id", DataTypes.BIGINT())))) @@ -441,7 +452,10 @@ class ExistingTableSchemaExpanderTest { queryFailureApplier, SchemaChangeBehavior.LENIENT, ExistingTableSchemaExpansionMode.EXPAND); - assertThatThrownBy(() -> queryFailureExpander.expand(eventWithMissingColumn())) + assertThatThrownBy( + () -> + queryFailureExpander.handleExistingTableCreation( + eventWithMissingColumn())) .isInstanceOf(FlinkRuntimeException.class); assertThat(queryFailureApplier.appliedEvents).isEmpty(); } @@ -459,28 +473,71 @@ class ExistingTableSchemaExpanderTest { ddlFailureApplier, SchemaChangeBehavior.LENIENT, ExistingTableSchemaExpansionMode.EXPAND) - .expand(eventWithMissingColumn())) + .handleExistingTableCreation(eventWithMissingColumn())) .isInstanceOf(FlinkRuntimeException.class); assertThat(ddlFailureApplier.queryCalls).isOne(); } @Test - void testDelegatesToSinkOnUnsupportedDdlFailure() { + void testTryExpandFailsFastOnUnsupportedDdlFailure() { TestingExistingTableSchemaExpansionSupport unsupportedApplier = new TestingExistingTableSchemaExpansionSupport( schema(Column.physicalColumn("id", DataTypes.INT()))); unsupportedApplier.unsupportedEventTypes.add(SchemaChangeEventType.ADD_COLUMN); - assertThatCode( + assertThatThrownBy( () -> new ExistingTableSchemaExpander( unsupportedApplier, unsupportedApplier, SchemaChangeBehavior.LENIENT) - .expand(eventWithMissingColumn())) - .doesNotThrowAnyException(); + .handleExistingTableCreation(eventWithMissingColumn())) + .isInstanceOf(UnsupportedSchemaChangeEventException.class); assertThat(unsupportedApplier.appliedEvents).isEmpty(); } + @Test + void testTryExpandFailsFastOnTransientDdlFailure() { + // After supportable differences are identified, a transient DDL failure must propagate so + // the job fails over and retries, instead of silently delegating to the sink and dropping + // the missing columns forever. + TestingExistingTableSchemaExpansionSupport ddlFailureApplier = + new TestingExistingTableSchemaExpansionSupport( + schema(Column.physicalColumn("id", DataTypes.INT()))); + ddlFailureApplier.failedEventTypes.add(SchemaChangeEventType.ADD_COLUMN); + + assertThatThrownBy( + () -> + new ExistingTableSchemaExpander( + ddlFailureApplier, + ddlFailureApplier, + SchemaChangeBehavior.LENIENT, + ExistingTableSchemaExpansionMode.TRY_EXPAND) + .handleExistingTableCreation(eventWithMissingColumn())) + .isInstanceOf(SchemaEvolveException.class); + assertThat(ddlFailureApplier.appliedEvents).isEmpty(); + } + + @Test + void testTryExpandFailsFastOnReadBackQueryFailure() { + // The initial read succeeds and derived DDL is issued, but the read-back verification query + // fails transiently. This must propagate for failover instead of being swallowed. + TestingExistingTableSchemaExpansionSupport readBackFailureApplier = + new TestingExistingTableSchemaExpansionSupport( + schema(Column.physicalColumn("id", DataTypes.INT()))); + readBackFailureApplier.queryFailureAfterCalls = 1; + + assertThatThrownBy( + () -> + new ExistingTableSchemaExpander( + readBackFailureApplier, + readBackFailureApplier, + SchemaChangeBehavior.LENIENT, + ExistingTableSchemaExpansionMode.TRY_EXPAND) + .handleExistingTableCreation(eventWithMissingColumn())) + .isInstanceOf(FlinkRuntimeException.class); + assertThat(readBackFailureApplier.appliedEvents).hasSize(1); + } + @Test void testVerifiesExpansionByReadingBackTargetSchema() { TestingExistingTableSchemaExpansionSupport unchangedTargetApplier = @@ -492,7 +549,7 @@ class ExistingTableSchemaExpanderTest { unchangedTargetApplier, unchangedTargetApplier, SchemaChangeBehavior.LENIENT) - .expand(eventWithMissingColumn()); + .handleExistingTableCreation(eventWithMissingColumn()); assertThat(unchangedTargetApplier.appliedEvents) .singleElement() .isInstanceOf(AddColumnEvent.class); @@ -508,7 +565,8 @@ class ExistingTableSchemaExpanderTest { unchangedTargetTypeApplier, unchangedTargetTypeApplier, SchemaChangeBehavior.LENIENT) - .expand(createTableEvent(Column.physicalColumn("id", DataTypes.BIGINT()))); + .handleExistingTableCreation( + createTableEvent(Column.physicalColumn("id", DataTypes.BIGINT()))); assertThat(unchangedTargetTypeApplier.appliedEvents) .singleElement() .isInstanceOf(AlterColumnTypeEvent.class); @@ -534,7 +592,9 @@ class ExistingTableSchemaExpanderTest { applier, SchemaChangeBehavior.LENIENT, ExistingTableSchemaExpansionMode.CHECK); - assertThat(expander.expand(createTableEvent(Column.physicalColumn("id", DataTypes.INT())))) + assertThat( + expander.handleExistingTableCreation( + createTableEvent(Column.physicalColumn("id", DataTypes.INT())))) .isFalse(); assertThat(applier.appliedEvents).isEmpty(); assertThat(applier.queryCalls).isOne(); @@ -552,7 +612,10 @@ class ExistingTableSchemaExpanderTest { missingColumnApplier, SchemaChangeBehavior.LENIENT, ExistingTableSchemaExpansionMode.CHECK); - assertThatThrownBy(() -> missingColumnExpander.expand(eventWithMissingColumn())) + assertThatThrownBy( + () -> + missingColumnExpander.handleExistingTableCreation( + eventWithMissingColumn())) .isInstanceOfSatisfying( SchemaEvolveException.class, e -> @@ -574,7 +637,7 @@ class ExistingTableSchemaExpanderTest { ExistingTableSchemaExpansionMode.CHECK); assertThatThrownBy( () -> - narrowColumnExpander.expand( + narrowColumnExpander.handleExistingTableCreation( createTableEvent( Column.physicalColumn("id", DataTypes.INT())))) .isInstanceOfSatisfying( @@ -599,7 +662,7 @@ class ExistingTableSchemaExpanderTest { applier, SchemaChangeBehavior.LENIENT, ExistingTableSchemaExpansionMode.EXPAND); - assertThatThrownBy(() -> expander.expand(eventWithMissingColumn())) + assertThatThrownBy(() -> expander.handleExistingTableCreation(eventWithMissingColumn())) .isInstanceOfSatisfying( SchemaEvolveException.class, e -> assertThat(e.getExceptionMessage()).contains("ADD_COLUMN")); @@ -618,7 +681,9 @@ class ExistingTableSchemaExpanderTest { applier, SchemaChangeBehavior.LENIENT, ExistingTableSchemaExpansionMode.EXPAND); - assertThat(expander.expand(createTableEvent(Column.physicalColumn("id", DataTypes.INT())))) + assertThat( + expander.handleExistingTableCreation( + createTableEvent(Column.physicalColumn("id", DataTypes.INT())))) .isTrue(); assertThat(applier.appliedEvents).isEmpty(); assertThat(applier.queryCalls).isOne(); @@ -637,7 +702,7 @@ class ExistingTableSchemaExpanderTest { applier, applier, behavior, ExistingTableSchemaExpansionMode.CHECK); assertThatThrownBy( () -> - expander.expand( + expander.handleExistingTableCreation( createTableEvent( Column.physicalColumn("id", DataTypes.INT())))) .isInstanceOf(SchemaEvolveException.class); @@ -654,7 +719,7 @@ class ExistingTableSchemaExpanderTest { applier, SchemaChangeBehavior.LENIENT, ExistingTableSchemaExpansionMode.CHECK); - assertThatThrownBy(() -> expander.expand(eventWithMissingColumn())) + assertThatThrownBy(() -> expander.handleExistingTableCreation(eventWithMissingColumn())) .isInstanceOfSatisfying( SchemaEvolveException.class, e -> @@ -679,7 +744,7 @@ class ExistingTableSchemaExpanderTest { ExistingTableSchemaExpansionMode.CHECK); assertThatThrownBy( () -> - expander.expand( + expander.handleExistingTableCreation( createTableEvent( Column.physicalColumn("id", DataTypes.INT()), Column.physicalColumn( @@ -705,7 +770,7 @@ class ExistingTableSchemaExpanderTest { applier, SchemaChangeBehavior.LENIENT, ExistingTableSchemaExpansionMode.EXPAND); - assertThatThrownBy(() -> expander.expand(eventWithMissingColumn())) + assertThatThrownBy(() -> expander.handleExistingTableCreation(eventWithMissingColumn())) .isInstanceOfSatisfying( SchemaEvolveException.class, e -> @@ -728,7 +793,7 @@ class ExistingTableSchemaExpanderTest { applier, SchemaChangeBehavior.LENIENT, ExistingTableSchemaExpansionMode.EXPAND); - assertThatThrownBy(() -> expander.expand(eventWithMissingColumn())) + assertThatThrownBy(() -> expander.handleExistingTableCreation(eventWithMissingColumn())) .isInstanceOfSatisfying( SchemaEvolveException.class, e -> assertThat(e.getExceptionMessage()).contains("after expansion")); @@ -745,8 +810,8 @@ class ExistingTableSchemaExpanderTest { applier, applier, SchemaChangeBehavior.LENIENT, - ExistingTableSchemaExpansionMode.OFF); - assertThat(expander.expand(eventWithMissingColumn())).isTrue(); + ExistingTableSchemaExpansionMode.DISABLED); + assertThat(expander.handleExistingTableCreation(eventWithMissingColumn())).isTrue(); assertThat(applier.queryCalls).isZero(); assertThat(applier.appliedEvents).isEmpty(); } @@ -758,7 +823,7 @@ class ExistingTableSchemaExpanderTest { schema(Column.physicalColumn("id", DataTypes.INT()))); new ExistingTableSchemaExpander(applier, applier, SchemaChangeBehavior.LENIENT) - .expand( + .handleExistingTableCreation( createTableEvent( Column.physicalColumn("id", DataTypes.BIGINT()), Column.physicalColumn( diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinatorTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinatorTest.java index e644b0800..c3d9e2be1 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinatorTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/regular/SchemaCoordinatorTest.java @@ -37,6 +37,7 @@ import org.apache.flink.cdc.runtime.operators.schema.regular.event.SchemaChangeR import org.apache.flink.cdc.runtime.testutils.operators.MockedOperatorCoordinatorContext; import org.apache.flink.runtime.jobgraph.OperatorID; import org.apache.flink.runtime.operators.coordination.CoordinationResponse; +import org.apache.flink.util.FlinkRuntimeException; import org.junit.jupiter.api.Test; @@ -164,6 +165,38 @@ class SchemaCoordinatorTest { } } + @Test + void tryExpandModeFailsWhenConnectorLacksExpansionSupport() { + UnsupportedExpansionMetadataApplier metadataApplier = + new UnsupportedExpansionMetadataApplier(); + + SchemaCoordinator coordinator = + new SchemaCoordinator( + "regular-schema-coordinator", + new MockedOperatorCoordinatorContext( + new OperatorID(), Thread.currentThread().getContextClassLoader()), + Executors.newSingleThreadExecutor(), + metadataApplier, + Collections.emptyList(), + RouteMode.ALL_MATCH, + SchemaChangeBehavior.LENIENT, + ExistingTableSchemaExpansionMode.TRY_EXPAND, + Duration.ofSeconds(10)); + + assertThatThrownBy(coordinator::start) + .isInstanceOf(FlinkRuntimeException.class) + .hasCauseInstanceOf(FlinkRuntimeException.class) + .hasRootCauseMessage( + "Existing target table schema expansion is enabled with mode TRY_EXPAND, but MetadataApplier " + + UnsupportedExpansionMetadataApplier.class.getName() + + " does not support it."); + } + + private static final class UnsupportedExpansionMetadataApplier implements MetadataApplier { + @Override + public void applySchemaChange(SchemaChangeEvent schemaChangeEvent) {} + } + private static final class ExpandingMetadataApplier implements MetadataApplier, ExistingTableSchemaExpansionSupport { private final Schema targetSchema;
