From 41c4eec9dd916cf15800ba315dc32cebe088856b Mon Sep 17 00:00:00 2001 From: sunxiaojian Date: Tue, 14 Nov 2023 11:04:22 +0800 Subject: [PATCH] Pulsar cdc add debezium-json test --- .../action/cdc/pulsar/PulsarActionUtils.java | 3 +- ... => PulsarCanalSyncTableActionITCase.java} | 2 +- .../PulsarDebeziumSyncTableActionITCase.java | 138 ++++++++++++++++++ 3 files changed, 140 insertions(+), 3 deletions(-) rename paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/{PulsarSyncTableActionITCase.java => PulsarCanalSyncTableActionITCase.java} (99%) create mode 100644 paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarDebeziumSyncTableActionITCase.java diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionUtils.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionUtils.java index da9a1d1f1ee53..ccf1fc7e7c973 100644 --- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionUtils.java +++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionUtils.java @@ -66,7 +66,7 @@ /** Utils for Pulsar synchronization. */ public class PulsarActionUtils { - + private static final String DEBEZIUM_JSON = "debezium-json"; static final ConfigOption VALUE_FORMAT = ConfigOptions.key("value.format") .stringType() @@ -359,7 +359,6 @@ static MessageQueueSchemaUtils.ConsumerWrapper createPulsarConsumer( } private static class PulsarConsumerWrapper implements MessageQueueSchemaUtils.ConsumerWrapper { - private final Consumer consumer; PulsarConsumerWrapper(Consumer consumer) { diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableActionITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarCanalSyncTableActionITCase.java similarity index 99% rename from paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableActionITCase.java rename to paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarCanalSyncTableActionITCase.java index a39507b5f5378..0e83bb800229d 100644 --- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableActionITCase.java +++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarCanalSyncTableActionITCase.java @@ -35,7 +35,7 @@ import static org.apache.paimon.flink.action.cdc.pulsar.PulsarActionUtils.VALUE_FORMAT; /** IT cases for {@link PulsarSyncTableAction}. */ -public class PulsarSyncTableActionITCase extends PulsarActionITCaseBase { +public class PulsarCanalSyncTableActionITCase extends PulsarActionITCaseBase { @Test @Timeout(120) diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarDebeziumSyncTableActionITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarDebeziumSyncTableActionITCase.java new file mode 100644 index 0000000000000..5fd5909a62e42 --- /dev/null +++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarDebeziumSyncTableActionITCase.java @@ -0,0 +1,138 @@ +/* + * 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.pulsar; + +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; + +import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS; +import static org.apache.paimon.flink.action.cdc.pulsar.PulsarActionUtils.TOPIC; +import static org.apache.paimon.flink.action.cdc.pulsar.PulsarActionUtils.VALUE_FORMAT; + +/** IT cases for {@link PulsarDebeziumSyncTableActionITCase}. */ +public class PulsarDebeziumSyncTableActionITCase extends PulsarActionITCaseBase { + + @Test + @Timeout(120) + public void testSchemaEvolution() throws Exception { + runSingleTableSchemaEvolution("schemaevolution"); + } + + private void runSingleTableSchemaEvolution(String sourceDir) throws Exception { + final String topic = "schema_evolution"; + createTopic(topic, 1); + // ---------- Write the debezium json into Pulsar ------------------- + sendMessages( + topic, + getMessages( + String.format("kafka/debezium/table/%s/debezium-data-1.txt", sourceDir))); + + Map pulsarConfig = getBasicPulsarConfig(); + pulsarConfig.put(PULSAR_PARTITION_DISCOVERY_INTERVAL_MS.key(), "-1"); + pulsarConfig.put(TOPIC.key(), topic); + pulsarConfig.put(VALUE_FORMAT.key(), "debezium-json"); + + PulsarSyncTableAction action = + syncTableActionBuilder(pulsarConfig) + .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 primaryKeys = Collections.singletonList("id"); + List 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); + + // ---------- Write the debezium json into Kafka ------------------- + sendMessages( + topic, + getMessages( + String.format("kafka/debezium/table/%s/debezium-data-2.txt", sourceDir))); + + 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); + + sendMessages( + topic, + getMessages( + String.format("kafka/debezium/table/%s/debezium-data-3.txt", sourceDir))); + + 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); + } +}