[
https://issues.apache.org/jira/browse/FLINK-41107?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18130805#comment-18130805
]
Xin Chen commented on FLINK-41107:
----------------------------------
*Proposed Solution*
Implement accepted schema evolution type handling in both HudiMetadataApplier
and MaxComputeMetadataApplier, following the existing pattern used by
DorisMetadataApplier.
1. Add an enabled-event-type field
{code:java}
private Set<SchemaChangeEventType> enabledSchemaEvolutionTypes;{code}
Initialize it in the constructor:
{code:java}
this.enabledSchemaEvolutionTypes = getSupportedSchemaEvolutionTypes();{code}
2. Implement the filtering contract
{code:java}
@Override
public MetadataApplier setAcceptedSchemaEvolutionTypes(
Set<SchemaChangeEventType> schemaEvolutionTypes) {
this.enabledSchemaEvolutionTypes = schemaEvolutionTypes;
return this;
}
@Override
public boolean acceptsSchemaEvolutionType(
SchemaChangeEventType schemaChangeEventType) {
return enabledSchemaEvolutionTypes.contains(schemaChangeEventType);
}
{code}
3. Declare sink-supported event types
For Hudi, return only event types that are actually implemented:
{code:java}
@Override
public Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
return EnumSet.of(
SchemaChangeEventType.CREATE_TABLE,
SchemaChangeEventType.ADD_COLUMN,
SchemaChangeEventType.ALTER_COLUMN_TYPE,
SchemaChangeEventType.DROP_COLUMN,
SchemaChangeEventType.RENAME_COLUMN);
}{code}
{{{}TRUNCATE_TABLE{}}}, {{{}DROP_TABLE{}}}, and {{ALTER_TABLE_COMMENT}} should
not be declared because Hudi does not support them.
For MaxCompute, return the event types currently implemented by
{{{}MaxComputeMetadataApplier{}}}, including {{TRUNCATE_TABLE}} and
{{{}DROP_TABLE{}}}.
This does not mean these destructive operations are always executed. It only
declares sink capability. Whether an event is forwarded remains controlled by:
{code:java}
sink:
include.schema.changes: ...
exclude.schema.changes: ...{code}
Expected Behavior After the Fix：
{code:java}
sink:
exclude.schema.changes:
- truncate.table
- drop.table{code}
Expected result:
* {{TruncateTableEvent}} and {{DropTableEvent}} are filtered by
{{{}SchemaDerivator{}}}.
* MaxCompute does not truncate or drop downstream tables.
* Hudi does not receive unsupported truncate/drop events.
* The pipeline continues running normally.
Tests
Add unit and integration tests to verify that:
- setAcceptedSchemaEvolutionTypes(...) changes the accepted event set.
- Excluded TRUNCATE_TABLE and DROP_TABLE return false from
acceptsSchemaEvolutionType(...).
- MaxCompute does not call truncate/drop logic when those event types are
excluded.
- Hudi does not fail when excluded truncate/drop events are received from the
source.
> schema.change.behavior: lenient or exclude.schema.changes is ignored by Hudi
> and MaxCompute sinks because their MetadataApplier implementations do not
> honor accepted schema evolution types
> ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41107
> URL: https://issues.apache.org/jira/browse/FLINK-41107
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Affects Versions: cdc-3.6.0, cdc-3.7.0
> Reporter: Xin Chen
> Priority: Major
>
> HudiMetadataApplier and MaxComputeMetadataApplier do not honor the schema
> evolution event types configured through include.schema.changes and
> exclude.schema.changes.
> In the Pipeline runtime, the resolved allowed schema change event types are
> passed to each sink through:
>
> {code:java}
> MetadataApplier#setAcceptedSchemaEvolutionTypes(...){code}
>
>
> Before forwarding an event to the sink, SchemaDerivator checks:
>
> {code:java}
> metadataApplier.acceptsSchemaEvolutionType(schemaChangeEvent.getType()){code}
>
> However, both Hudi and MaxCompute metadata appliers rely on the default
> methods in MetadataApplier:
>
> {code:java}
> default MetadataApplier setAcceptedSchemaEvolutionTypes(
> Set<SchemaChangeEventType> schemaEvolutionTypes) {
> return this;
> }
> default boolean acceptsSchemaEvolutionType(
> SchemaChangeEventType schemaChangeEventType) {
> return true;
> }
> {code}
>
> As a result, the configured accepted event types are discarded and every
> schema change event is accepted.
> This makes configurations such as the following ineffective:
>
> {code:java}
> sink:
> exclude.schema.changes:
> - truncate.table
> - drop.table{code}
> or
> {code:java}
> schema.change.behavior: lenient{code}
>
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
