This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 101143217 [FLINK-41107][pipeline-connector] Honor accepted schema
evolution types in Hudi and MaxCompute sinks (#4599)
101143217 is described below
commit 101143217237f9e4021de43ecb15863568f973d4
Author: YasuoStudyJava <[email protected]>
AuthorDate: Tue Nov 3 17:20:14 2026 +0800
[FLINK-41107][pipeline-connector] Honor accepted schema evolution types in
Hudi and MaxCompute sinks (#4599)
HudiMetadataApplier and MaxComputeMetadataApplier relied on the default
implementations of MetadataApplier#setAcceptedSchemaEvolutionTypes and
MetadataApplier#acceptsSchemaEvolutionType, which discard the configured event
types and accept every schema change event. As a result,
sink.exclude.schema.changes / sink.include.schema.changes (and therefore
schema.change.behavior: lenient) had no effect for these two sinks: excluded
truncate.table and drop.table events were still forwarded to [...]
This closes the gap:
- HudiMetadataApplier keeps the enabled schema evolution types and declares
only the event types it can actually apply: create.table, add.column,
alter.column.type, drop.column and rename.column.
- MaxComputeMetadataApplier keeps the enabled schema evolution types so
that events excluded by the pipeline configuration are no longer applied.
- Add unit tests covering the default supported types and the filtering
performed after setAcceptedSchemaEvolutionTypes is called.
Co-authored-by: chenxin49 <[email protected]>
---
.../connectors/hudi/sink/HudiMetadataApplier.java | 27 ++++++++++++
.../hudi/sink/HudiMetadataApplierTest.java | 23 ++++++++++
.../maxcompute/MaxComputeMetadataApplier.java | 23 ++++++++++
.../maxcompute/MaxComputeMetadataApplierTest.java | 49 ++++++++++++++++++++++
4 files changed, 122 insertions(+)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/main/java/org/apache/flink/cdc/connectors/hudi/sink/HudiMetadataApplier.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/main/java/org/apache/flink/cdc/connectors/hudi/sink/HudiMetadataApplier.java
index c47bc3153..a57cde5f9 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/main/java/org/apache/flink/cdc/connectors/hudi/sink/HudiMetadataApplier.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/main/java/org/apache/flink/cdc/connectors/hudi/sink/HudiMetadataApplier.java
@@ -24,6 +24,7 @@ import org.apache.flink.cdc.common.event.CreateTableEvent;
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.TableId;
import org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor;
import org.apache.flink.cdc.common.schema.Column;
@@ -53,9 +54,11 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
/**
@@ -74,10 +77,34 @@ public class HudiMetadataApplier implements MetadataApplier
{
private transient HoodieCatalog catalog;
private final org.apache.flink.configuration.Configuration catalogConfig;
+ private Set<SchemaChangeEventType> enabledSchemaEvolutionTypes;
public HudiMetadataApplier(Configuration config) {
this.config = config;
this.catalogConfig = convertToCatalogConfig(config);
+ this.enabledSchemaEvolutionTypes = getSupportedSchemaEvolutionTypes();
+ }
+
+ @Override
+ public MetadataApplier setAcceptedSchemaEvolutionTypes(
+ Set<SchemaChangeEventType> schemaEvolutionTypes) {
+ this.enabledSchemaEvolutionTypes = schemaEvolutionTypes;
+ return this;
+ }
+
+ @Override
+ public boolean acceptsSchemaEvolutionType(SchemaChangeEventType
schemaChangeEventType) {
+ return enabledSchemaEvolutionTypes.contains(schemaChangeEventType);
+ }
+
+ @Override
+ public Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
+ return EnumSet.of(
+ SchemaChangeEventType.CREATE_TABLE,
+ SchemaChangeEventType.ADD_COLUMN,
+ SchemaChangeEventType.ALTER_COLUMN_TYPE,
+ SchemaChangeEventType.DROP_COLUMN,
+ SchemaChangeEventType.RENAME_COLUMN);
}
@Override
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/test/java/org/apache/flink/cdc/connectors/hudi/sink/HudiMetadataApplierTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/test/java/org/apache/flink/cdc/connectors/hudi/sink/HudiMetadataApplierTest.java
index b12a47f2e..3ba265783 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/test/java/org/apache/flink/cdc/connectors/hudi/sink/HudiMetadataApplierTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/test/java/org/apache/flink/cdc/connectors/hudi/sink/HudiMetadataApplierTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
import org.apache.flink.cdc.common.event.CreateTableEvent;
import org.apache.flink.cdc.common.event.DropColumnEvent;
import org.apache.flink.cdc.common.event.RenameColumnEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEventType;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.sink.MetadataApplier;
@@ -36,11 +37,13 @@ import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatNoException;
/** Tests for {@link HudiMetadataApplier}. */
@@ -48,6 +51,26 @@ class HudiMetadataApplierTest {
@TempDir public static java.nio.file.Path temporaryFolder;
+ @Test
+ void testAcceptedSchemaEvolutionTypes() {
+ Configuration config = new Configuration();
+ config.set(HudiConfig.PATH, temporaryFolder.toString());
+ HudiMetadataApplier metadataApplier = new HudiMetadataApplier(config);
+
+
assertThat(metadataApplier.acceptsSchemaEvolutionType(SchemaChangeEventType.TRUNCATE_TABLE))
+ .isFalse();
+
assertThat(metadataApplier.acceptsSchemaEvolutionType(SchemaChangeEventType.DROP_TABLE))
+ .isFalse();
+
+ metadataApplier.setAcceptedSchemaEvolutionTypes(
+ EnumSet.of(SchemaChangeEventType.CREATE_TABLE));
+
+
assertThat(metadataApplier.acceptsSchemaEvolutionType(SchemaChangeEventType.CREATE_TABLE))
+ .isTrue();
+
assertThat(metadataApplier.acceptsSchemaEvolutionType(SchemaChangeEventType.ADD_COLUMN))
+ .isFalse();
+ }
+
@Test
void testApplySchemaChange() throws Exception {
String basePath =
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/MaxComputeMetadataApplier.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/MaxComputeMetadataApplier.java
index 06d202c1b..cb25e773e 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/MaxComputeMetadataApplier.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/MaxComputeMetadataApplier.java
@@ -26,6 +26,7 @@ import org.apache.flink.cdc.common.event.DropColumnEvent;
import org.apache.flink.cdc.common.event.DropTableEvent;
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.TruncateTableEvent;
import org.apache.flink.cdc.common.exceptions.SchemaEvolveException;
import org.apache.flink.cdc.common.sink.MetadataApplier;
@@ -41,15 +42,37 @@ import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.EnumSet;
+import java.util.Set;
+
/** A {@link MetadataApplier} for "MaxCompute" connector. */
public class MaxComputeMetadataApplier implements MetadataApplier {
private static final long serialVersionUID = 1L;
private static final Logger LOG =
LoggerFactory.getLogger(MaxComputeMetadataApplier.class);
private final MaxComputeOptions maxComputeOptions;
+ private Set<SchemaChangeEventType> enabledSchemaEvolutionTypes;
public MaxComputeMetadataApplier(MaxComputeOptions maxComputeOptions) {
this.maxComputeOptions = maxComputeOptions;
+ this.enabledSchemaEvolutionTypes = getSupportedSchemaEvolutionTypes();
+ }
+
+ @Override
+ public MetadataApplier setAcceptedSchemaEvolutionTypes(
+ Set<SchemaChangeEventType> schemaEvolutionTypes) {
+ this.enabledSchemaEvolutionTypes = schemaEvolutionTypes;
+ return this;
+ }
+
+ @Override
+ public boolean acceptsSchemaEvolutionType(SchemaChangeEventType
schemaChangeEventType) {
+ return enabledSchemaEvolutionTypes.contains(schemaChangeEventType);
+ }
+
+ @Override
+ public Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
+ return EnumSet.allOf(SchemaChangeEventType.class);
}
@Override
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/MaxComputeMetadataApplierTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/MaxComputeMetadataApplierTest.java
new file mode 100644
index 000000000..ab84b4e67
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/test/java/org/apache/flink/cdc/connectors/maxcompute/MaxComputeMetadataApplierTest.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.maxcompute;
+
+import org.apache.flink.cdc.common.event.SchemaChangeEventType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.EnumSet;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class MaxComputeMetadataApplierTest {
+
+ @Test
+ void testAcceptedSchemaEvolutionTypes() {
+ MaxComputeMetadataApplier applier = new
MaxComputeMetadataApplier(null);
+
+
assertThat(applier.acceptsSchemaEvolutionType(SchemaChangeEventType.TRUNCATE_TABLE))
+ .isTrue();
+
assertThat(applier.acceptsSchemaEvolutionType(SchemaChangeEventType.DROP_TABLE)).isTrue();
+
+ applier.setAcceptedSchemaEvolutionTypes(
+ EnumSet.complementOf(
+ EnumSet.of(
+ SchemaChangeEventType.TRUNCATE_TABLE,
+ SchemaChangeEventType.DROP_TABLE)));
+
+
assertThat(applier.acceptsSchemaEvolutionType(SchemaChangeEventType.TRUNCATE_TABLE))
+ .isFalse();
+
assertThat(applier.acceptsSchemaEvolutionType(SchemaChangeEventType.DROP_TABLE)).isFalse();
+
assertThat(applier.acceptsSchemaEvolutionType(SchemaChangeEventType.ADD_COLUMN)).isTrue();
+ }
+}
