You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2023/11/16 08:17:02 UTC
(incubator-paimon) branch master updated: [cdc] Debezium-json supports debezium records with included schema (#2280)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new e5b847a9d [cdc] Debezium-json supports debezium records with included schema (#2280)
e5b847a9d is described below
commit e5b847a9d7f886f8e38d4fcf4269c232a6bfae53
Author: Xiaojian Sun <su...@163.com>
AuthorDate: Thu Nov 16 16:16:57 2023 +0800
[cdc] Debezium-json supports debezium records with included schema (#2280)
---
.../flink/action/cdc/format/RecordParser.java | 2 +-
.../cdc/format/debezium/DebeziumRecordParser.java | 13 ++
...fkaDebeziumWithSchemaSyncTableActionITCase.java | 185 +++++++++++++++++++++
.../schema/computedcolumn/debezium-data-1.txt | 19 +++
.../schema/schemaevolution/debezium-data-1.txt | 20 +++
.../schema/schemaevolution/debezium-data-2.txt | 20 +++
.../schema/schemaevolution/debezium-data-3.txt | 22 +++
7 files changed, 280 insertions(+), 1 deletion(-)
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
index 96042fcb1..261d26308 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
@@ -219,7 +219,7 @@ public abstract class RecordParser implements FlatMapFunction<String, RichCdcMul
new CdcRecord(rowKind, data));
}
- private void setRoot(String record) {
+ protected void setRoot(String record) {
root = JsonSerdeUtil.fromJson(record, JsonNode.class);
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumRecordParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumRecordParser.java
index 8bdce72b1..1301bb532 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumRecordParser.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumRecordParser.java
@@ -23,6 +23,7 @@ import org.apache.paimon.flink.action.cdc.TypeMapping;
import org.apache.paimon.flink.action.cdc.format.RecordParser;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
import org.apache.paimon.types.RowKind;
+import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
@@ -52,6 +53,8 @@ import static org.apache.paimon.utils.JsonSerdeUtil.isNull;
*/
public class DebeziumRecordParser extends RecordParser {
+ private static final String FIELD_SCHEMA = "schema";
+ private static final String FIELD_PAYLOAD = "payload";
private static final String FIELD_BEFORE = "before";
private static final String FIELD_AFTER = "after";
private static final String FIELD_SOURCE = "source";
@@ -99,6 +102,16 @@ public class DebeziumRecordParser extends RecordParser {
return getAndCheck(FIELD_BEFORE, FIELD_TYPE, op);
}
+ @Override
+ protected void setRoot(String record) {
+ JsonNode node = JsonSerdeUtil.fromJson(record, JsonNode.class);
+ if (node.has(FIELD_SCHEMA)) {
+ root = node.get(FIELD_PAYLOAD);
+ } else {
+ root = node;
+ }
+ }
+
@Override
protected String primaryField() {
return FIELD_PRIMARY;
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaDebeziumWithSchemaSyncTableActionITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaDebeziumWithSchemaSyncTableActionITCase.java
new file mode 100644
index 000000000..cf2847951
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaDebeziumWithSchemaSyncTableActionITCase.java
@@ -0,0 +1,185 @@
+/*
+ * 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.paimon.flink.action.cdc.kafka;
+
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/** IT cases for {@link KafkaSyncTableAction}. */
+public class KafkaDebeziumWithSchemaSyncTableActionITCase extends KafkaActionITCaseBase {
+
+ @Test
+ @Timeout(300)
+ public void testSchemaEvolution() throws Exception {
+ runSingleTableSchemaEvolution("schemaevolution");
+ }
+
+ private void runSingleTableSchemaEvolution(String sourceDir) throws Exception {
+ final String topic = "schema_evolution";
+ createTestTopic(topic, 1, 1);
+ // ---------- Write the debezium json into Kafka -------------------
+ List<String> lines =
+ readLines(
+ String.format(
+ "kafka/debezium/table/schema/%s/debezium-data-1.txt", sourceDir));
+ try {
+ writeRecordsToKafka(topic, lines);
+ } catch (Exception e) {
+ throw new Exception("Failed to write debezium data to Kafka.", e);
+ }
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put("value.format", "debezium-json");
+ kafkaConfig.put("value.debezium-json.schema-include", "true");
+ kafkaConfig.put("topic", topic);
+ KafkaSyncTableAction action =
+ syncTableActionBuilder(kafkaConfig)
+ .withPrimaryKeys("id")
+ .withTableConfig(getBasicTableConfig())
+ .build();
+ runActionWithDefaultEnv(action);
+
+ testSchemaEvolutionImpl(topic, sourceDir);
+ }
+
+ private void testSchemaEvolutionImpl(String topic, String sourceDir) throws Exception {
+ FileStoreTable table = getFileStoreTable(tableName);
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight"});
+ List<String> primaryKeys = Collections.singletonList("id");
+ List<String> expected =
+ Arrays.asList(
+ "+I[101, scooter, Small 2-wheel scooter, 3.14]",
+ "+I[102, car battery, 12V car battery, 8.1]");
+ waitForResult(expected, table, rowType, primaryKeys);
+
+ try {
+ writeRecordsToKafka(
+ topic,
+ readLines(
+ String.format(
+ "kafka/debezium/table/schema/%s/debezium-data-2.txt",
+ sourceDir)));
+ } catch (Exception e) {
+ throw new Exception("Failed to write debezium data to Kafka.", e);
+ }
+ rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight", "age"});
+ expected =
+ Arrays.asList(
+ "+I[101, scooter, Small 2-wheel scooter, 3.14, NULL]",
+ "+I[102, car battery, 12V car battery, 8.1, NULL]",
+ "+I[103, 12-pack drill bits, 12-pack of drill bits with sizes ranging from #40 to #3, 0.8, 18]",
+ "+I[104, hammer, 12oz carpenter's hammer, 0.75, 24]");
+ waitForResult(expected, table, rowType, primaryKeys);
+
+ try {
+ writeRecordsToKafka(
+ topic,
+ readLines(
+ String.format(
+ "kafka/debezium/table/schema/%s/debezium-data-3.txt",
+ sourceDir)));
+ } catch (Exception e) {
+ throw new Exception("Failed to write debezium data to Kafka.", e);
+ }
+ rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"id", "name", "description", "weight", "age", "address"});
+ expected =
+ Arrays.asList(
+ "+I[102, car battery, 12V car battery, 8.1, NULL, NULL]",
+ "+I[103, 12-pack drill bits, 12-pack of drill bits with sizes ranging from #40 to #3, 0.8, 18, NULL]",
+ "+I[104, hammer, 12oz carpenter's hammer, 0.75, 24, NULL]",
+ "+I[105, hammer, 14oz carpenter's hammer, 0.875, NULL, Beijing]",
+ "+I[107, rocks, box of assorted rocks, 5.3, NULL, NULL]");
+ waitForResult(expected, table, rowType, primaryKeys);
+ }
+
+ @Test
+ @Timeout(60)
+ public void testComputedColumn() throws Exception {
+ String topic = "computed_column";
+ createTestTopic(topic, 1, 1);
+
+ List<String> lines =
+ readLines("kafka/debezium/table/schema/computedcolumn/debezium-data-1.txt");
+ try {
+ writeRecordsToKafka(topic, lines);
+ } catch (Exception e) {
+ throw new Exception("Failed to write canal data to Kafka.", e);
+ }
+ Map<String, String> kafkaConfig = getBasicKafkaConfig();
+ kafkaConfig.put("value.format", "debezium-json");
+ kafkaConfig.put("value.debezium-json.schema-include", "true");
+ kafkaConfig.put("topic", topic);
+ KafkaSyncTableAction action =
+ syncTableActionBuilder(kafkaConfig)
+ .withPrimaryKeys("id")
+ .withComputedColumnArgs("_year=year(date)")
+ .withTableConfig(getBasicTableConfig())
+ .build();
+ runActionWithDefaultEnv(action);
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.STRING().notNull(), DataTypes.STRING(), DataTypes.INT()
+ },
+ new String[] {"id", "date", "_year"});
+ waitForResult(
+ Collections.singletonList("+I[101, 2023-03-23, 2023]"),
+ getFileStoreTable(tableName),
+ rowType,
+ Collections.singletonList("id"));
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/computedcolumn/debezium-data-1.txt b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/computedcolumn/debezium-data-1.txt
new file mode 100644
index 000000000..d2c34a8f6
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/computedcolumn/debezium-data-1.txt
@@ -0,0 +1,19 @@
+/*
+ * 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.
+ */
+
+{"schema":null, "payload":{"before": null, "after": {"id": 101, "date": "2023-03-23"}, "source": {"version": "1.9.7.Final", "connector": "mysql", "name": "mysql_binlog_source", "ts_ms": 1596684883000, "snapshot": "false", "db": "test", "sequence": null, "table": "product", "server_id": 0, "gtid": null, "file": "", "pos": 0, "row": 0, "thread": null, "query": null}, "op": "c", "ts_ms": 1596684883000, "transaction": null}}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-1.txt b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-1.txt
new file mode 100644
index 000000000..3b34b6f8f
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-1.txt
@@ -0,0 +1,20 @@
+/*
+ * 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.
+ */
+
+{"schema":null, "payload":{"before": null, "after": {"id": 101, "name": "scooter", "description": "Small 2-wheel scooter", "weight": 3.14}, "source": {"version": "1.9.7.Final", "connector": "mysql", "name": "mysql_binlog_source", "ts_ms": 1596684883000, "snapshot": "false", "db": "test", "sequence": null, "table": "product", "server_id": 0, "gtid": null, "file": "", "pos": 0, "row": 0, "thread": null, "query": null}, "op": "c", "ts_ms": 1596684883000, "transaction": null}}
+{"schema":null, "payload":{"before": null, "after": {"id": 102, "name": "car battery", "description": "12V car battery", "weight": 8.1}, "source": {"version": "1.9.7.Final", "connector": "mysql", "name": "mysql_binlog_source", "ts_ms": 1596684883000, "snapshot": "false", "db": "test", "sequence": null, "table": "product", "server_id": 0, "gtid": null, "file": "", "pos": 0, "row": 0, "thread": null, "query": null}, "op": "c", "ts_ms": 1596684883000, "transaction": null}}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-2.txt b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-2.txt
new file mode 100644
index 000000000..3b0571d72
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-2.txt
@@ -0,0 +1,20 @@
+/*
+ * 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.
+ */
+
+{"schema":null, "payload":{"before": null, "after": {"id": 103, "name": "12-pack drill bits", "description": "12-pack of drill bits with sizes ranging from #40 to #3", "weight": 0.8, "age": 18}, "source": {"version": "1.9.7.Final", "connector": "mysql", "name": "mysql_binlog_source", "ts_ms": 1596684883000, "snapshot": "false", "db": "test", "sequence": null, "table": "product", "server_id": 0, "gtid": null, "file": "", "pos": 0, "row": 0, "thread": null, "query": null}, "op": "c", "ts_m [...]
+{"schema":null, "payload":{"before": null, "after": {"id": 104, "name": "hammer", "description": "12oz carpenter's hammer", "weight": 0.75, "age": 24}, "source": {"version": "1.9.7.Final", "connector": "mysql", "name": "mysql_binlog_source", "ts_ms": 1596684883000, "snapshot": "false", "db": "test", "sequence": null, "table": "product", "server_id": 0, "gtid": null, "file": "", "pos": 0, "row": 0, "thread": null, "query": null}, "op": "c", "ts_ms": 1596684883000, "transaction": null}}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-3.txt b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-3.txt
new file mode 100644
index 000000000..ea99eb71f
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/resources/kafka/debezium/table/schema/schemaevolution/debezium-data-3.txt
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+{"schema":null, "payload":{"before": null, "after": {"id": 105, "name": "hammer", "description": "14oz carpenter's hammer", "weight": 0.875, "address": "Shanghai"}, "source": {"version": "1.9.7.Final", "connector": "mysql", "name": "mysql_binlog_source", "ts_ms": 1596684883000, "snapshot": "false", "db": "test", "sequence": null, "table": "product", "server_id": 0, "gtid": null, "file": "", "pos": 0, "row": 0, "thread": null, "query": null}, "op": "c", "ts_ms": 1596684883000, "transactio [...]
+{"schema":null, "payload":{"before": {"id": 101, "name": "scooter", "description": "Small 2-wheel scooter", "weight": 3.14}, "after": null, "source": {"version": "1.9.7.Final", "connector": "mysql", "name": "mysql_binlog_source", "ts_ms": 1596684883000, "snapshot": "false", "db": "test", "sequence": null, "table": "product", "server_id": 0, "gtid": null, "file": "", "pos": 0, "row": 0, "thread": null, "query": null}, "op": "d", "ts_ms": 1596684883000, "transaction": null}}
+{"schema":null, "payload":{"before": {"address": "Shanghai"}, "after": {"id": 105, "name": "hammer", "description": "14oz carpenter's hammer", "weight": 0.875, "address": "Beijing"}, "source": {"version": "1.9.7.Final", "connector": "mysql", "name": "mysql_binlog_source", "ts_ms": 1596684906000, "snapshot": "false", "db": "test", "sequence": null, "table": "product", "server_id": 0, "gtid": null, "file": "", "pos": 0, "row": 0, "thread": null, "query": null}, "op": "u", "ts_ms": 15966849 [...]
+{"schema":null, "payload":{"before": null, "after": {"id": 107, "name": "rocks", "description": "box of assorted rocks", "weight": 5.3}, "source": {"version": "1.9.7.Final", "connector": "mysql", "name": "mysql_binlog_source", "ts_ms": 1596684883000, "snapshot": "false", "db": "test", "sequence": null, "table": "product", "server_id": 0, "gtid": null, "file": "", "pos": 0, "row": 0, "thread": null, "query": null}, "op": "c", "ts_ms": 1596684883000, "transaction": null}}