You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2024/08/27 16:53:49 UTC
(flink-cdc) 10/12: [FLINK-36128][cdc-runtime] Fix potential unrecoverable in-flight data exception by promoting LENIENT as the default schema change behavior
This is an automated email from the ASF dual-hosted git repository.
leonard pushed a commit to branch release-3.2
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
commit debd43cdd75eb53ceed8c2483e0a14edc1b265c6
Author: yuxiqian <34...@users.noreply.github.com>
AuthorDate: Mon Aug 26 17:46:37 2024 +0800
[FLINK-36128][cdc-runtime] Fix potential unrecoverable in-flight data exception by promoting LENIENT as the default schema change behavior
This closes #3574.
(cherry picked from commit 2e938a92f5335515f7eb46077254aee25acd9107)
---
.../cli/parser/YamlPipelineDefinitionParser.java | 38 ++++++++----
.../parser/YamlPipelineDefinitionParserTest.java | 36 +++++++++++-
.../cdc/common/event/SchemaChangeEventType.java | 20 +++++--
.../flink/cdc/common/pipeline/PipelineOptions.java | 2 +-
.../flink/cdc/common/utils/ChangeEventUtils.java | 8 +--
.../cdc/common/utils/ChangeEventUtilsTest.java | 67 ++++++++++++----------
.../flink/FlinkPipelineComposerITCase.java | 21 +++++++
.../flink/FlinkPipelineTransformITCase.java | 13 +++++
.../cdc/composer/flink/FlinkPipelineUdfITCase.java | 21 +++++++
.../flink/cdc/pipeline/tests/MysqlE2eITCase.java | 3 +-
.../tests/SchemaEvolvingTransformE2eITCase.java | 10 ++--
.../cdc/pipeline/tests/TransformE2eITCase.java | 18 ++++--
.../runtime/partitioning/PrePartitionOperator.java | 6 +-
13 files changed, 195 insertions(+), 68 deletions(-)
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 d5df8eda1..ffba60784 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
@@ -19,6 +19,8 @@ package org.apache.flink.cdc.cli.parser;
import org.apache.flink.cdc.common.configuration.Configuration;
import org.apache.flink.cdc.common.event.SchemaChangeEventType;
+import org.apache.flink.cdc.common.event.SchemaChangeEventTypeFamily;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
import org.apache.flink.cdc.common.utils.StringUtils;
import org.apache.flink.cdc.composer.definition.PipelineDef;
import org.apache.flink.cdc.composer.definition.RouteDef;
@@ -35,11 +37,13 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.yaml.YA
import java.nio.file.Path;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import static org.apache.flink.cdc.common.pipeline.PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR;
import static org.apache.flink.cdc.common.utils.ChangeEventUtils.resolveSchemaEvolutionOptions;
import static org.apache.flink.cdc.common.utils.Preconditions.checkNotNull;
@@ -99,6 +103,19 @@ public class YamlPipelineDefinitionParser implements PipelineDefinitionParser {
private PipelineDef parse(JsonNode pipelineDefJsonNode, Configuration globalPipelineConfig)
throws Exception {
+
+ // UDFs are optional. We parse UDF first and remove it from the pipelineDefJsonNode since
+ // it's not of plain data types and must be removed before calling toPipelineConfig.
+ List<UdfDef> udfDefs = new ArrayList<>();
+ Optional.ofNullable(((ObjectNode) pipelineDefJsonNode.get(PIPELINE_KEY)).remove(UDF_KEY))
+ .ifPresent(node -> node.forEach(udf -> udfDefs.add(toUdfDef(udf))));
+
+ // Pipeline configs are optional
+ Configuration userPipelineConfig = toPipelineConfig(pipelineDefJsonNode.get(PIPELINE_KEY));
+
+ SchemaChangeBehavior schemaChangeBehavior =
+ userPipelineConfig.get(PIPELINE_SCHEMA_CHANGE_BEHAVIOR);
+
// Source is required
SourceDef sourceDef =
toSourceDef(
@@ -113,7 +130,8 @@ public class YamlPipelineDefinitionParser implements PipelineDefinitionParser {
checkNotNull(
pipelineDefJsonNode.get(SINK_KEY),
"Missing required field \"%s\" in pipeline definition",
- SINK_KEY));
+ SINK_KEY),
+ schemaChangeBehavior);
// Transforms are optional
List<TransformDef> transformDefs = new ArrayList<>();
@@ -128,14 +146,6 @@ public class YamlPipelineDefinitionParser implements PipelineDefinitionParser {
Optional.ofNullable(pipelineDefJsonNode.get(ROUTE_KEY))
.ifPresent(node -> node.forEach(route -> routeDefs.add(toRouteDef(route))));
- // UDFs are optional
- List<UdfDef> udfDefs = new ArrayList<>();
- Optional.ofNullable(((ObjectNode) pipelineDefJsonNode.get(PIPELINE_KEY)).remove(UDF_KEY))
- .ifPresent(node -> node.forEach(udf -> udfDefs.add(toUdfDef(udf))));
-
- // Pipeline configs are optional
- Configuration userPipelineConfig = toPipelineConfig(pipelineDefJsonNode.get(PIPELINE_KEY));
-
// Merge user config into global config
Configuration pipelineConfig = new Configuration();
pipelineConfig.addAll(globalPipelineConfig);
@@ -162,7 +172,7 @@ public class YamlPipelineDefinitionParser implements PipelineDefinitionParser {
return new SourceDef(type, name, Configuration.fromMap(sourceMap));
}
- private SinkDef toSinkDef(JsonNode sinkNode) {
+ private SinkDef toSinkDef(JsonNode sinkNode, SchemaChangeBehavior schemaChangeBehavior) {
List<String> includedSETypes = new ArrayList<>();
List<String> excludedSETypes = new ArrayList<>();
@@ -172,6 +182,14 @@ public class YamlPipelineDefinitionParser implements PipelineDefinitionParser {
Optional.ofNullable(sinkNode.get(EXCLUDE_SCHEMA_EVOLUTION_TYPES))
.ifPresent(e -> e.forEach(tag -> excludedSETypes.add(tag.asText())));
+ if (includedSETypes.isEmpty()) {
+ // If no schema evolution types are specified, include all schema evolution types by
+ // default.
+ Arrays.stream(SchemaChangeEventTypeFamily.ALL)
+ .map(SchemaChangeEventType::getTag)
+ .forEach(includedSETypes::add);
+ }
+
Set<SchemaChangeEventType> declaredSETypes =
resolveSchemaEvolutionOptions(includedSETypes, excludedSETypes);
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 f57dd62c7..2ecf45870 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
@@ -27,6 +27,7 @@ import org.apache.flink.cdc.composer.definition.TransformDef;
import org.apache.flink.cdc.composer.definition.UdfDef;
import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableSet;
import org.apache.flink.shaded.guava31.com.google.common.io.Resources;
import org.junit.jupiter.api.Test;
@@ -37,6 +38,11 @@ import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
+import static org.apache.flink.cdc.common.event.SchemaChangeEventType.ADD_COLUMN;
+import static org.apache.flink.cdc.common.event.SchemaChangeEventType.ALTER_COLUMN_TYPE;
+import static org.apache.flink.cdc.common.event.SchemaChangeEventType.CREATE_TABLE;
+import static org.apache.flink.cdc.common.event.SchemaChangeEventType.DROP_COLUMN;
+import static org.apache.flink.cdc.common.event.SchemaChangeEventType.RENAME_COLUMN;
import static org.apache.flink.cdc.common.pipeline.PipelineOptions.PIPELINE_LOCAL_TIME_ZONE;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
@@ -384,7 +390,13 @@ class YamlPipelineDefinitionParserTest {
Configuration.fromMap(
ImmutableMap.<String, String>builder()
.put("bootstrap-servers", "localhost:9092")
- .build())),
+ .build()),
+ ImmutableSet.of(
+ DROP_COLUMN,
+ ALTER_COLUMN_TYPE,
+ ADD_COLUMN,
+ CREATE_TABLE,
+ RENAME_COLUMN)),
Collections.singletonList(
new RouteDef(
"mydb.default.app_order_.*",
@@ -401,7 +413,16 @@ class YamlPipelineDefinitionParserTest {
private final PipelineDef minimizedDef =
new PipelineDef(
new SourceDef("mysql", null, new Configuration()),
- new SinkDef("kafka", null, new Configuration()),
+ new SinkDef(
+ "kafka",
+ null,
+ new Configuration(),
+ ImmutableSet.of(
+ DROP_COLUMN,
+ ALTER_COLUMN_TYPE,
+ ADD_COLUMN,
+ CREATE_TABLE,
+ RENAME_COLUMN)),
Collections.emptyList(),
Collections.emptyList(),
Collections.emptyList(),
@@ -474,7 +495,16 @@ class YamlPipelineDefinitionParserTest {
private final PipelineDef pipelineDefWithUdf =
new PipelineDef(
new SourceDef("values", null, new Configuration()),
- new SinkDef("values", null, new Configuration()),
+ new SinkDef(
+ "values",
+ null,
+ new Configuration(),
+ ImmutableSet.of(
+ DROP_COLUMN,
+ ALTER_COLUMN_TYPE,
+ ADD_COLUMN,
+ CREATE_TABLE,
+ RENAME_COLUMN)),
Collections.emptyList(),
Collections.singletonList(
new TransformDef(
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/SchemaChangeEventType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/SchemaChangeEventType.java
index 668ea76a4..3e03ec4cf 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/SchemaChangeEventType.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/SchemaChangeEventType.java
@@ -22,11 +22,21 @@ import org.apache.flink.cdc.common.annotation.PublicEvolving;
/** An enumeration of schema change event types for {@link SchemaChangeEvent}. */
@PublicEvolving
public enum SchemaChangeEventType {
- ADD_COLUMN,
- ALTER_COLUMN_TYPE,
- CREATE_TABLE,
- DROP_COLUMN,
- RENAME_COLUMN;
+ ADD_COLUMN("add.column"),
+ ALTER_COLUMN_TYPE("alter.column.type"),
+ CREATE_TABLE("create.table"),
+ DROP_COLUMN("drop.column"),
+ RENAME_COLUMN("rename.column");
+
+ private final String tag;
+
+ SchemaChangeEventType(String tag) {
+ this.tag = tag;
+ }
+
+ public String getTag() {
+ return tag;
+ }
public static SchemaChangeEventType ofEvent(SchemaChangeEvent event) {
if (event instanceof AddColumnEvent) {
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
index 48a4fbb13..343e99270 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
@@ -48,7 +48,7 @@ public class PipelineOptions {
public static final ConfigOption<SchemaChangeBehavior> PIPELINE_SCHEMA_CHANGE_BEHAVIOR =
ConfigOptions.key("schema.change.behavior")
.enumType(SchemaChangeBehavior.class)
- .defaultValue(SchemaChangeBehavior.EVOLVE)
+ .defaultValue(SchemaChangeBehavior.LENIENT)
.withDescription(
Description.builder()
.text("Behavior for handling schema change events. ")
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/ChangeEventUtils.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/ChangeEventUtils.java
index ee72ef6e2..e2c289e35 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/ChangeEventUtils.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/ChangeEventUtils.java
@@ -95,12 +95,8 @@ public class ChangeEventUtils {
List<String> includedSchemaEvolutionTypes, List<String> excludedSchemaEvolutionTypes) {
List<SchemaChangeEventType> resultTypes = new ArrayList<>();
- if (includedSchemaEvolutionTypes.isEmpty()) {
- resultTypes.addAll(Arrays.asList(SchemaChangeEventTypeFamily.ALL));
- } else {
- for (String includeTag : includedSchemaEvolutionTypes) {
- resultTypes.addAll(resolveSchemaEvolutionTag(includeTag));
- }
+ for (String includeTag : includedSchemaEvolutionTypes) {
+ resultTypes.addAll(resolveSchemaEvolutionTag(includeTag));
}
for (String excludeTag : excludedSchemaEvolutionTypes) {
diff --git a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/ChangeEventUtilsTest.java b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/ChangeEventUtilsTest.java
index 0a34a7501..e159fb1d1 100644
--- a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/ChangeEventUtilsTest.java
+++ b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/ChangeEventUtilsTest.java
@@ -17,59 +17,67 @@
package org.apache.flink.cdc.common.utils;
-import org.assertj.core.api.Assertions;
+import org.apache.flink.cdc.common.event.SchemaChangeEventType;
+import org.apache.flink.cdc.common.event.SchemaChangeEventTypeFamily;
+
import org.assertj.core.util.Sets;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
import static org.apache.flink.cdc.common.event.SchemaChangeEventType.ADD_COLUMN;
import static org.apache.flink.cdc.common.event.SchemaChangeEventType.ALTER_COLUMN_TYPE;
import static org.apache.flink.cdc.common.event.SchemaChangeEventType.CREATE_TABLE;
import static org.apache.flink.cdc.common.event.SchemaChangeEventType.DROP_COLUMN;
import static org.apache.flink.cdc.common.event.SchemaChangeEventType.RENAME_COLUMN;
+import static org.apache.flink.cdc.common.testutils.assertions.EventAssertions.assertThat;
/** A test for the {@link org.apache.flink.cdc.common.utils.ChangeEventUtils}. */
public class ChangeEventUtilsTest {
@Test
public void testResolveSchemaEvolutionOptions() {
- Assertions.assertThat(
- ChangeEventUtils.resolveSchemaEvolutionOptions(
- Collections.emptyList(), Collections.emptyList()))
+
+ List<String> allTags =
+ Arrays.stream(SchemaChangeEventTypeFamily.ALL)
+ .map(SchemaChangeEventType::getTag)
+ .collect(Collectors.toList());
+ assertThat(ChangeEventUtils.resolveSchemaEvolutionOptions(allTags, Collections.emptyList()))
.isEqualTo(
Sets.set(
+ RENAME_COLUMN,
CREATE_TABLE,
- ADD_COLUMN,
ALTER_COLUMN_TYPE,
- DROP_COLUMN,
- RENAME_COLUMN));
+ ADD_COLUMN,
+ DROP_COLUMN));
- Assertions.assertThat(
+ assertThat(
ChangeEventUtils.resolveSchemaEvolutionOptions(
- Collections.emptyList(), Collections.singletonList("drop")))
- .isEqualTo(Sets.set(CREATE_TABLE, ADD_COLUMN, ALTER_COLUMN_TYPE, RENAME_COLUMN));
+ allTags, Collections.singletonList("drop")))
+ .isEqualTo(Sets.set(ADD_COLUMN, ALTER_COLUMN_TYPE, RENAME_COLUMN, CREATE_TABLE));
- Assertions.assertThat(
+ assertThat(
ChangeEventUtils.resolveSchemaEvolutionOptions(
Arrays.asList("create", "add"), Collections.emptyList()))
- .isEqualTo(Sets.set(CREATE_TABLE, ADD_COLUMN));
+ .isEqualTo(Sets.set(ADD_COLUMN, CREATE_TABLE));
- Assertions.assertThat(
+ assertThat(
ChangeEventUtils.resolveSchemaEvolutionOptions(
Collections.singletonList("column"),
Collections.singletonList("drop.column")))
.isEqualTo(Sets.set(ADD_COLUMN, ALTER_COLUMN_TYPE, RENAME_COLUMN));
- Assertions.assertThat(
+ assertThat(
ChangeEventUtils.resolveSchemaEvolutionOptions(
- Collections.emptyList(), Collections.singletonList("drop.column")))
- .isEqualTo(Sets.set(CREATE_TABLE, ADD_COLUMN, ALTER_COLUMN_TYPE, RENAME_COLUMN));
+ allTags, Collections.singletonList("drop.column")))
+ .isEqualTo(Sets.set(ADD_COLUMN, RENAME_COLUMN, ALTER_COLUMN_TYPE, CREATE_TABLE));
}
@Test
public void testResolveSchemaEvolutionTag() {
- Assertions.assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("all"))
+ assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("all"))
.isEqualTo(
Arrays.asList(
ADD_COLUMN,
@@ -78,41 +86,38 @@ public class ChangeEventUtilsTest {
DROP_COLUMN,
RENAME_COLUMN));
- Assertions.assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("column"))
+ assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("column"))
.isEqualTo(
Arrays.asList(ADD_COLUMN, ALTER_COLUMN_TYPE, DROP_COLUMN, RENAME_COLUMN));
- Assertions.assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("table"))
+ assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("table"))
.isEqualTo(Collections.singletonList(CREATE_TABLE));
- Assertions.assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("rename"))
- .isEqualTo(Collections.singletonList(RENAME_COLUMN));
-
- Assertions.assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("rename.column"))
+ assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("rename.column"))
.isEqualTo(Collections.singletonList(RENAME_COLUMN));
- Assertions.assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("drop"))
+ assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("drop"))
.isEqualTo(Collections.singletonList(DROP_COLUMN));
- Assertions.assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("drop.column"))
+ assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("drop.column"))
.isEqualTo(Collections.singletonList(DROP_COLUMN));
- Assertions.assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("create"))
+ assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("create"))
.isEqualTo(Collections.singletonList(CREATE_TABLE));
- Assertions.assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("create.table"))
+ assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("create.table"))
.isEqualTo(Collections.singletonList(CREATE_TABLE));
- Assertions.assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("alter"))
+ assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("alter"))
.isEqualTo(Collections.singletonList(ALTER_COLUMN_TYPE));
- Assertions.assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("alter.column.type"))
+ assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("alter.column.type"))
.isEqualTo(Collections.singletonList(ALTER_COLUMN_TYPE));
- Assertions.assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("add"))
+ assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("add"))
.isEqualTo(Collections.singletonList(ADD_COLUMN));
- Assertions.assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("add.column"))
+ assertThat(ChangeEventUtils.resolveSchemaEvolutionTag("add.column"))
.isEqualTo(Collections.singletonList(ADD_COLUMN));
}
}
diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java
index 8e471d886..4e16d2380 100644
--- a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposerITCase.java
@@ -26,6 +26,7 @@ import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.event.RenameColumnEvent;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.common.types.DataType;
@@ -134,6 +135,8 @@ class FlinkPipelineComposerITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -191,6 +194,8 @@ class FlinkPipelineComposerITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -313,6 +318,8 @@ class FlinkPipelineComposerITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -373,6 +380,8 @@ class FlinkPipelineComposerITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -441,6 +450,8 @@ class FlinkPipelineComposerITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -496,6 +507,8 @@ class FlinkPipelineComposerITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -569,6 +582,8 @@ class FlinkPipelineComposerITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -766,6 +781,8 @@ class FlinkPipelineComposerITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -972,6 +989,8 @@ class FlinkPipelineComposerITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -1035,6 +1054,8 @@ class FlinkPipelineComposerITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
index 322a3358e..54ae2c225 100644
--- a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
@@ -29,6 +29,7 @@ import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.event.RenameColumnEvent;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.common.types.DataType;
@@ -213,6 +214,8 @@ class FlinkPipelineTransformITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
pipelineConfig.set(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE, "America/Los_Angeles");
PipelineDef pipelineDef =
new PipelineDef(
@@ -266,6 +269,8 @@ class FlinkPipelineTransformITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -350,6 +355,8 @@ class FlinkPipelineTransformITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -436,6 +443,8 @@ class FlinkPipelineTransformITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -516,6 +525,8 @@ class FlinkPipelineTransformITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -608,6 +619,8 @@ class FlinkPipelineTransformITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java
index f9be7d7dd..c3b412dc7 100644
--- a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineUdfITCase.java
@@ -19,6 +19,7 @@ package org.apache.flink.cdc.composer.flink;
import org.apache.flink.cdc.common.configuration.Configuration;
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
+import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
import org.apache.flink.cdc.composer.PipelineExecution;
import org.apache.flink.cdc.composer.definition.PipelineDef;
import org.apache.flink.cdc.composer.definition.SinkDef;
@@ -139,6 +140,8 @@ public class FlinkPipelineUdfITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -207,6 +210,8 @@ public class FlinkPipelineUdfITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -273,6 +278,8 @@ public class FlinkPipelineUdfITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -341,6 +348,8 @@ public class FlinkPipelineUdfITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -411,6 +420,8 @@ public class FlinkPipelineUdfITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -472,6 +483,8 @@ public class FlinkPipelineUdfITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -565,6 +578,8 @@ public class FlinkPipelineUdfITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -632,6 +647,8 @@ public class FlinkPipelineUdfITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -697,6 +714,8 @@ public class FlinkPipelineUdfITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -758,6 +777,8 @@ public class FlinkPipelineUdfITCase {
// Setup pipeline
Configuration pipelineConfig = new Configuration();
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR, SchemaChangeBehavior.EVOLVE);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eITCase.java
index c16f6de9c..5cb22d089 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eITCase.java
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eITCase.java
@@ -210,7 +210,8 @@ public class MysqlE2eITCase extends PipelineTestEnvironment {
+ " type: values\n"
+ "\n"
+ "pipeline:\n"
- + " parallelism: %d",
+ + " parallelism: %d\n"
+ + " schema.change.behavior: evolve",
INTER_CONTAINER_MYSQL_ALIAS,
MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD,
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaEvolvingTransformE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaEvolvingTransformE2eITCase.java
index a4a21c992..8b306d1e7 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaEvolvingTransformE2eITCase.java
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/SchemaEvolvingTransformE2eITCase.java
@@ -228,11 +228,12 @@ public class SchemaEvolvingTransformE2eITCase extends PipelineTestEnvironment {
+ "\n"
+ "pipeline:\n"
+ " schema.change.behavior: unexpected\n"
- + " parallelism: 1",
+ + " parallelism: %d",
INTER_CONTAINER_MYSQL_ALIAS,
MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD,
- schemaEvolveDatabase.getDatabaseName());
+ schemaEvolveDatabase.getDatabaseName(),
+ parallelism);
Path mysqlCdcJar = TestUtils.getResource("mysql-cdc-pipeline-connector.jar");
Path valuesCdcJar = TestUtils.getResource("values-cdc-pipeline-connector.jar");
Path mysqlDriverJar = TestUtils.getResource("mysql-driver.jar");
@@ -305,7 +306,7 @@ public class SchemaEvolvingTransformE2eITCase extends PipelineTestEnvironment {
+ "\n"
+ "pipeline:\n"
+ " schema.change.behavior: %s\n"
- + " parallelism: 1",
+ + " parallelism: %d",
INTER_CONTAINER_MYSQL_ALIAS,
MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD,
@@ -313,7 +314,8 @@ public class SchemaEvolvingTransformE2eITCase extends PipelineTestEnvironment {
mergeTable ? "(members|new_members)" : "members",
dbName,
dbName,
- behavior);
+ behavior,
+ parallelism);
Path mysqlCdcJar = TestUtils.getResource("mysql-cdc-pipeline-connector.jar");
Path valuesCdcJar = TestUtils.getResource("values-cdc-pipeline-connector.jar");
Path mysqlDriverJar = TestUtils.getResource("mysql-driver.jar");
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java
index c1c8e8109..0a3b0aace 100644
--- a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/TransformE2eITCase.java
@@ -738,12 +738,14 @@ public class TransformE2eITCase extends PipelineTestEnvironment {
+ " projection: ID, 'id -> ' || ID AS UID, PRICEALPHA AS PRICE\n"
+ " filter: ID > 1008\n"
+ "pipeline:\n"
- + " parallelism: 1",
+ + " parallelism: %d\n"
+ + " schema.change.behavior: evolve",
INTER_CONTAINER_MYSQL_ALIAS,
MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD,
transformTestDatabase.getDatabaseName(),
- transformTestDatabase.getDatabaseName());
+ transformTestDatabase.getDatabaseName(),
+ parallelism);
Path mysqlCdcJar = TestUtils.getResource("mysql-cdc-pipeline-connector.jar");
Path valuesCdcJar = TestUtils.getResource("values-cdc-pipeline-connector.jar");
Path mysqlDriverJar = TestUtils.getResource("mysql-driver.jar");
@@ -835,12 +837,14 @@ public class TransformE2eITCase extends PipelineTestEnvironment {
+ " projection: \\*, 'id -> ' || ID AS UID\n"
+ " filter: ID > 1008\n"
+ "pipeline:\n"
- + " parallelism: 1",
+ + " parallelism: %d\n"
+ + " schema.change.behavior: evolve",
INTER_CONTAINER_MYSQL_ALIAS,
MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD,
transformTestDatabase.getDatabaseName(),
- transformTestDatabase.getDatabaseName());
+ transformTestDatabase.getDatabaseName(),
+ parallelism);
Path mysqlCdcJar = TestUtils.getResource("mysql-cdc-pipeline-connector.jar");
Path valuesCdcJar = TestUtils.getResource("values-cdc-pipeline-connector.jar");
Path mysqlDriverJar = TestUtils.getResource("mysql-driver.jar");
@@ -940,12 +944,14 @@ public class TransformE2eITCase extends PipelineTestEnvironment {
+ " projection: ID || ' <- id' AS UID, *\n"
+ " filter: ID > 1008\n"
+ "pipeline:\n"
- + " parallelism: 1",
+ + " parallelism: %d\n"
+ + " schema.change.behavior: evolve",
INTER_CONTAINER_MYSQL_ALIAS,
MYSQL_TEST_USER,
MYSQL_TEST_PASSWORD,
transformTestDatabase.getDatabaseName(),
- transformTestDatabase.getDatabaseName());
+ transformTestDatabase.getDatabaseName(),
+ parallelism);
Path mysqlCdcJar = TestUtils.getResource("mysql-cdc-pipeline-connector.jar");
Path valuesCdcJar = TestUtils.getResource("values-cdc-pipeline-connector.jar");
Path mysqlDriverJar = TestUtils.getResource("mysql-driver.jar");
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperator.java
index 6d66fa878..1171ad07b 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperator.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperator.java
@@ -28,6 +28,7 @@ import org.apache.flink.cdc.common.function.HashFunctionProvider;
import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.runtime.operators.schema.SchemaOperator;
import org.apache.flink.cdc.runtime.operators.sink.SchemaEvolutionClient;
+import org.apache.flink.cdc.runtime.serializer.event.EventSerializer;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.jobgraph.tasks.TaskOperatorEventGateway;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
@@ -108,7 +109,10 @@ public class PrePartitionOperator extends AbstractStreamOperator<PartitioningEve
private void broadcastEvent(Event toBroadcast) {
for (int i = 0; i < downstreamParallelism; i++) {
- output.collect(new StreamRecord<>(new PartitioningEvent(toBroadcast, i)));
+ // Deep-copying each event is required since downstream subTasks might run in the same
+ // JVM
+ Event copiedEvent = EventSerializer.INSTANCE.copy(toBroadcast);
+ output.collect(new StreamRecord<>(new PartitioningEvent(copiedEvent, i)));
}
}