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/10/09 05:30:23 UTC
[incubator-paimon] branch master updated: [flink] Introduce Pulsar synchronizing table action (#2099)
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 df2ebe53e [flink] Introduce Pulsar synchronizing table action (#2099)
df2ebe53e is described below
commit df2ebe53eac87ab6c2d9fbd8329f13bafbac67c6
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Mon Oct 9 13:30:18 2023 +0800
[flink] Introduce Pulsar synchronizing table action (#2099)
---
paimon-flink/paimon-flink-cdc/pom.xml | 16 +
.../flink/action/cdc/MessageQueueSchemaUtils.java | 97 ++++++
...n.java => MessageQueueSyncTableActionBase.java} | 98 +++---
.../action/cdc/{kafka => }/format/DataFormat.java | 30 +-
.../cdc/{kafka => }/format/RecordParser.java | 4 +-
.../{kafka => }/format/RecordParserFactory.java | 2 +-
.../{kafka => }/format/canal/CanalFieldParser.java | 2 +-
.../format/canal/CanalRecordParser.java | 4 +-
.../format/maxwell/MaxwellRecordParser.java | 4 +-
.../{kafka => }/format/ogg/OggRecordParser.java | 4 +-
.../flink/action/cdc/kafka/KafkaActionUtils.java | 73 +++-
.../flink/action/cdc/kafka/KafkaSchemaUtils.java | 146 --------
.../action/cdc/kafka/KafkaSyncDatabaseAction.java | 7 +-
.../action/cdc/kafka/KafkaSyncTableAction.java | 198 ++---------
.../cdc/kafka/KafkaSyncTableActionFactory.java | 12 +-
.../flink/action/cdc/pulsar/PulsarActionUtils.java | 386 +++++++++++++++++++++
.../action/cdc/pulsar/PulsarSyncTableAction.java | 72 ++++
.../cdc/pulsar/PulsarSyncTableActionFactory.java | 85 +++++
.../services/org.apache.paimon.factories.Factory | 1 +
.../action/cdc/kafka/KafkaActionITCaseBase.java | 2 +-
.../cdc/kafka/KafkaCanalSyncTableActionITCase.java | 2 +-
.../flink/action/cdc/kafka/KafkaSchemaITCase.java | 19 +-
.../action/cdc/pulsar/PulsarActionITCaseBase.java | 318 +++++++++++++++++
.../action/cdc/pulsar/PulsarSchemaITCase.java | 69 ++++
.../cdc/pulsar/PulsarSyncTableActionITCase.java | 188 ++++++++++
25 files changed, 1419 insertions(+), 420 deletions(-)
diff --git a/paimon-flink/paimon-flink-cdc/pom.xml b/paimon-flink/paimon-flink-cdc/pom.xml
index 3054993b3..6caf71461 100644
--- a/paimon-flink/paimon-flink-cdc/pom.xml
+++ b/paimon-flink/paimon-flink-cdc/pom.xml
@@ -41,6 +41,7 @@ under the License.
<geometry.version>2.2.0</geometry.version>
<json-path.version>2.8.0</json-path.version>
<mongodb.testcontainers.version>1.18.3</mongodb.testcontainers.version>
+ <flink.connector.pulsar.version>4.0.0-1.17</flink.connector.pulsar.version>
</properties>
<dependencies>
@@ -102,6 +103,14 @@ under the License.
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-pulsar</artifactId>
+ <version>${flink.connector.pulsar.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+
<dependency>
<groupId>com.ververica</groupId>
<artifactId>flink-connector-mongodb-cdc</artifactId>
@@ -276,6 +285,13 @@ under the License.
<version>${mongodb.testcontainers.version}</version>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>pulsar</artifactId>
+ <version>${testcontainers.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSchemaUtils.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSchemaUtils.java
new file mode 100644
index 000000000..f8273a93a
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSchemaUtils.java
@@ -0,0 +1,97 @@
+/*
+ * 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;
+
+import org.apache.paimon.flink.action.cdc.format.DataFormat;
+import org.apache.paimon.flink.action.cdc.format.RecordParser;
+import org.apache.paimon.schema.Schema;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/** Utility class to build schema by trying to read and parse records from message queue. */
+public class MessageQueueSchemaUtils {
+
+ private static final int MAX_RETRY = 5;
+ private static final int POLL_TIMEOUT_MILLIS = 1000;
+
+ /**
+ * Retrieves the Kafka schema for a given topic.
+ *
+ * @param consumer The wrapper of message queue consumer to fetch messages.
+ * @param topic The topic to retrieve the schema for.
+ * @param dataFormat The data format for the messages in the message queue.
+ * @param typeMapping Data type mapping options.
+ * @return The schema for the topic.
+ * @throws SchemaRetrievalException If unable to retrieve the schema after max retries.
+ */
+ public static Schema getSchema(
+ ConsumerWrapper consumer, String topic, DataFormat dataFormat, TypeMapping typeMapping)
+ throws SchemaRetrievalException {
+ int retry = 0;
+ int retryInterval = 1000;
+
+ RecordParser recordParser =
+ dataFormat.createParser(true, typeMapping, Collections.emptyList());
+
+ while (true) {
+ Optional<Schema> schema =
+ consumer.getRecords(topic, POLL_TIMEOUT_MILLIS).stream()
+ .map(recordParser::buildSchema)
+ .filter(Objects::nonNull)
+ .findFirst();
+
+ if (schema.isPresent()) {
+ return schema.get();
+ }
+
+ if (retry >= MAX_RETRY) {
+ throw new SchemaRetrievalException(
+ String.format("Could not get metadata from server, topic: %s", topic));
+ }
+
+ sleepSafely(retryInterval);
+ retryInterval *= 2;
+ retry++;
+ }
+ }
+
+ private static void sleepSafely(int duration) {
+ try {
+ Thread.sleep(duration);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /** Custom exception to indicate issues with schema retrieval. */
+ public static class SchemaRetrievalException extends Exception {
+ public SchemaRetrievalException(String message) {
+ super(message);
+ }
+ }
+
+ /** Wrap the consumer for different message queues. */
+ public interface ConsumerWrapper extends AutoCloseable {
+
+ List<String> getRecords(String topic, int pollTimeOutMills);
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
similarity index 68%
copy from paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java
copy to paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
index 4de611b30..15571d74d 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.action.cdc.kafka;
+package org.apache.paimon.flink.action.cdc;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.catalog.Catalog;
@@ -24,11 +24,8 @@ import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.flink.action.Action;
import org.apache.paimon.flink.action.ActionBase;
-import org.apache.paimon.flink.action.cdc.CdcMetadataConverter;
-import org.apache.paimon.flink.action.cdc.ComputedColumn;
-import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.format.DataFormat;
-import org.apache.paimon.flink.action.cdc.kafka.format.RecordParser;
+import org.apache.paimon.flink.action.cdc.format.DataFormat;
+import org.apache.paimon.flink.action.cdc.format.RecordParser;
import org.apache.paimon.flink.sink.cdc.CdcSinkBuilder;
import org.apache.paimon.flink.sink.cdc.EventParser;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
@@ -37,10 +34,9 @@ import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.connector.source.Source;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions;
import java.util.ArrayList;
import java.util.Arrays;
@@ -53,17 +49,11 @@ import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.buildPaimo
import static org.apache.paimon.flink.action.cdc.ComputedColumnUtils.buildComputedColumns;
/**
- * An {@link Action} which synchronize one kafka topic into one Paimon table.
- *
- * <p>This topic must be from canal-json format.
- *
- * <p>You should specify Kafka source topic in {@code kafkaConfig}. See <a
- * href="https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/table/kafka/">document
- * of flink-connectors</a> for detailed keys and values.
+ * Base {@link Action} for synchronizing one message queue topic into one Paimon table.
*
* <p>If the specified Paimon table does not exist, this action will automatically create the table.
- * Its schema will be derived from all specified Kafka topic. If the Paimon table already exists,
- * its schema will be compared against the schema of all specified Kafka topic.
+ * Its schema will be derived from all specified topics. If the Paimon table already exists, its
+ * schema will be compared against the schema of all specified topics.
*
* <p>This action supports a limited number of schema changes. Unsupported schema changes will be
* ignored. Currently supported schema changes includes:
@@ -84,98 +74,106 @@ import static org.apache.paimon.flink.action.cdc.ComputedColumnUtils.buildComput
* are supported. Other type changes will cause exceptions.
* </ul>
*/
-public class KafkaSyncTableAction extends ActionBase {
+public abstract class MessageQueueSyncTableActionBase extends ActionBase {
- private final String database;
- private final String table;
- private final Configuration kafkaConfig;
+ protected final String database;
+ protected final String table;
+ protected final Configuration mqConfig;
private FileStoreTable fileStoreTable;
- private List<String> partitionKeys = new ArrayList<>();
- private List<String> primaryKeys = new ArrayList<>();
+ protected List<String> partitionKeys = new ArrayList<>();
+ protected List<String> primaryKeys = new ArrayList<>();
- private Map<String, String> tableConfig = new HashMap<>();
- private List<String> computedColumnArgs = new ArrayList<>();
- private TypeMapping typeMapping = TypeMapping.defaultMapping();
+ protected Map<String, String> tableConfig = new HashMap<>();
+ protected List<String> computedColumnArgs = new ArrayList<>();
+ protected TypeMapping typeMapping = TypeMapping.defaultMapping();
- public KafkaSyncTableAction(
+ public MessageQueueSyncTableActionBase(
String warehouse,
String database,
String table,
Map<String, String> catalogConfig,
- Map<String, String> kafkaConfig) {
+ Map<String, String> mqConfig) {
super(warehouse, catalogConfig);
this.database = database;
this.table = table;
- this.kafkaConfig = Configuration.fromMap(kafkaConfig);
+ this.mqConfig = Configuration.fromMap(mqConfig);
}
- public KafkaSyncTableAction withPartitionKeys(String... partitionKeys) {
+ public MessageQueueSyncTableActionBase withPartitionKeys(String... partitionKeys) {
return withPartitionKeys(Arrays.asList(partitionKeys));
}
- public KafkaSyncTableAction withPartitionKeys(List<String> partitionKeys) {
+ public MessageQueueSyncTableActionBase withPartitionKeys(List<String> partitionKeys) {
this.partitionKeys = partitionKeys;
return this;
}
- public KafkaSyncTableAction withPrimaryKeys(String... primaryKeys) {
+ public MessageQueueSyncTableActionBase withPrimaryKeys(String... primaryKeys) {
return withPrimaryKeys(Arrays.asList(primaryKeys));
}
- public KafkaSyncTableAction withPrimaryKeys(List<String> primaryKeys) {
+ public MessageQueueSyncTableActionBase withPrimaryKeys(List<String> primaryKeys) {
this.primaryKeys = primaryKeys;
return this;
}
- public KafkaSyncTableAction withTableConfig(Map<String, String> tableConfig) {
+ public MessageQueueSyncTableActionBase withTableConfig(Map<String, String> tableConfig) {
this.tableConfig = tableConfig;
return this;
}
- public KafkaSyncTableAction withComputedColumnArgs(List<String> computedColumnArgs) {
+ public MessageQueueSyncTableActionBase withComputedColumnArgs(List<String> computedColumnArgs) {
this.computedColumnArgs = computedColumnArgs;
return this;
}
- public KafkaSyncTableAction withTypeMapping(TypeMapping typeMapping) {
+ public MessageQueueSyncTableActionBase withTypeMapping(TypeMapping typeMapping) {
this.typeMapping = typeMapping;
return this;
}
+ protected abstract Source<String, ?, ?> buildSource();
+
+ protected abstract Schema buildSchema();
+
+ protected abstract DataFormat getDataFormat();
+
+ protected abstract String sourceName();
+
@Override
public void build(StreamExecutionEnvironment env) throws Exception {
- KafkaSource<String> source = KafkaActionUtils.buildKafkaSource(kafkaConfig);
- String topic = kafkaConfig.get(KafkaConnectorOptions.TOPIC).get(0);
+ Source<String, ?, ?> source = buildSource();
catalog.createDatabase(database, true);
boolean caseSensitive = catalog.caseSensitive();
- Schema kafkaSchema =
- KafkaSchemaUtils.getKafkaSchema(kafkaConfig, topic, typeMapping, caseSensitive);
+ // TODO: add case validate
+
+ Schema mqSchema = buildSchema();
Identifier identifier = new Identifier(database, table);
- List<ComputedColumn> computedColumns =
- buildComputedColumns(computedColumnArgs, kafkaSchema);
- Schema fromKafka =
+ List<ComputedColumn> computedColumns = buildComputedColumns(computedColumnArgs, mqSchema);
+ Schema fromMq =
buildPaimonSchema(
partitionKeys,
primaryKeys,
computedColumns,
tableConfig,
- kafkaSchema,
+ mqSchema,
new CdcMetadataConverter[] {});
try {
fileStoreTable = (FileStoreTable) catalog.getTable(identifier);
fileStoreTable = fileStoreTable.copy(tableConfig);
- assertSchemaCompatible(fileStoreTable.schema(), fromKafka.fields());
+ assertSchemaCompatible(fileStoreTable.schema(), fromMq.fields());
} catch (Catalog.TableNotExistException e) {
- catalog.createTable(identifier, fromKafka, false);
+ catalog.createTable(identifier, fromMq, false);
fileStoreTable = (FileStoreTable) catalog.getTable(identifier);
}
- DataFormat format = DataFormat.getDataFormat(kafkaConfig);
+ DataFormat format = getDataFormat();
RecordParser recordParser =
format.createParser(caseSensitive, typeMapping, computedColumns);
+
EventParser.Factory<RichCdcMultiplexRecord> parserFactory =
() -> new RichCdcMultiplexRecordEventParser(caseSensitive);
@@ -185,7 +183,7 @@ public class KafkaSyncTableAction extends ActionBase {
env.fromSource(
source,
WatermarkStrategy.noWatermarks(),
- "Kafka Source")
+ sourceName())
.flatMap(recordParser))
.withParserFactory(parserFactory)
.withTable(fileStoreTable)
@@ -212,10 +210,12 @@ public class KafkaSyncTableAction extends ActionBase {
// Flink run methods
// ------------------------------------------------------------------------
+ protected abstract String jobName();
+
@Override
public void run() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
build(env);
- env.execute(String.format("Kafka-Paimon Table Sync: %s.%s", database, table));
+ env.execute(jobName());
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/DataFormat.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/DataFormat.java
similarity index 65%
rename from paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/DataFormat.java
rename to paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/DataFormat.java
index e178fec40..fd5f99602 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/DataFormat.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/DataFormat.java
@@ -16,22 +16,19 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.action.cdc.kafka.format;
+package org.apache.paimon.flink.action.cdc.format;
import org.apache.paimon.flink.action.cdc.ComputedColumn;
import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.format.canal.CanalRecordParser;
-import org.apache.paimon.flink.action.cdc.kafka.format.maxwell.MaxwellRecordParser;
-import org.apache.paimon.flink.action.cdc.kafka.format.ogg.OggRecordParser;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions;
+import org.apache.paimon.flink.action.cdc.format.canal.CanalRecordParser;
+import org.apache.paimon.flink.action.cdc.format.maxwell.MaxwellRecordParser;
+import org.apache.paimon.flink.action.cdc.format.ogg.OggRecordParser;
import java.util.List;
/**
- * Enumerates the supported data formats and provides a mechanism to create their associated {@link
- * RecordParser}.
+ * Enumerates the supported data formats for message queue and provides a mechanism to create their
+ * associated {@link RecordParser}.
*
* <p>Each data format is associated with a specific implementation of {@link RecordParserFactory},
* which can be used to create instances of {@link RecordParser} for that format.
@@ -61,21 +58,12 @@ public enum DataFormat {
return parser.createParser(caseSensitive, typeMapping, computedColumns);
}
- /**
- * Determines the appropriate {@link DataFormat} based on the provided Kafka configuration.
- *
- * @param kafkaConfig The Kafka configuration containing the desired data format.
- * @return The corresponding {@link DataFormat}.
- * @throws UnsupportedOperationException If the specified format in the configuration is not
- * supported.
- */
- public static DataFormat getDataFormat(Configuration kafkaConfig) {
- String formatStr = kafkaConfig.get(KafkaConnectorOptions.VALUE_FORMAT);
+ public static DataFormat fromConfigString(String format) {
try {
- return DataFormat.valueOf(formatStr.replace("-", "_").toUpperCase());
+ return DataFormat.valueOf(format.replace("-", "_").toUpperCase());
} catch (Exception e) {
throw new UnsupportedOperationException(
- String.format("This format: %s is not supported.", formatStr));
+ String.format("This format: %s is not supported.", format));
}
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/RecordParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
similarity index 98%
rename from paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/RecordParser.java
rename to paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
index 2c0ffb19a..643f5d4ad 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/RecordParser.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParser.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.action.cdc.kafka.format;
+package org.apache.paimon.flink.action.cdc.format;
import org.apache.paimon.flink.action.cdc.ComputedColumn;
import org.apache.paimon.flink.action.cdc.TypeMapping;
@@ -90,7 +90,7 @@ public abstract class RecordParser implements FlatMapFunction<String, RichCdcMul
}
@Nullable
- public Schema getKafkaSchema(String record) {
+ public Schema buildSchema(String record) {
this.parseRootJson(record);
if (this.isDDL()) {
return null;
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/RecordParserFactory.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParserFactory.java
similarity index 96%
rename from paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/RecordParserFactory.java
rename to paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParserFactory.java
index e83d4b0d7..29c3aafb2 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/RecordParserFactory.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/RecordParserFactory.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.action.cdc.kafka.format;
+package org.apache.paimon.flink.action.cdc.format;
import org.apache.paimon.flink.action.cdc.ComputedColumn;
import org.apache.paimon.flink.action.cdc.TypeMapping;
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/canal/CanalFieldParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalFieldParser.java
similarity index 98%
rename from paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/canal/CanalFieldParser.java
rename to paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalFieldParser.java
index 619da6b79..185e02750 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/canal/CanalFieldParser.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalFieldParser.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.action.cdc.kafka.format.canal;
+package org.apache.paimon.flink.action.cdc.format.canal;
/** Converts some special types such as enum、set、geometry. */
public class CanalFieldParser {
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/canal/CanalRecordParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
similarity index 98%
rename from paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/canal/CanalRecordParser.java
rename to paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
index f0404fa46..8507b8cbc 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/canal/CanalRecordParser.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/canal/CanalRecordParser.java
@@ -16,11 +16,11 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.action.cdc.kafka.format.canal;
+package org.apache.paimon.flink.action.cdc.format.canal;
import org.apache.paimon.flink.action.cdc.ComputedColumn;
import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.format.RecordParser;
+import org.apache.paimon.flink.action.cdc.format.RecordParser;
import org.apache.paimon.flink.action.cdc.mysql.MySqlTypeUtils;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
import org.apache.paimon.types.DataType;
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/maxwell/MaxwellRecordParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/maxwell/MaxwellRecordParser.java
similarity index 96%
rename from paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/maxwell/MaxwellRecordParser.java
rename to paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/maxwell/MaxwellRecordParser.java
index f116d5d27..34f72bcd0 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/maxwell/MaxwellRecordParser.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/maxwell/MaxwellRecordParser.java
@@ -16,11 +16,11 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.action.cdc.kafka.format.maxwell;
+package org.apache.paimon.flink.action.cdc.format.maxwell;
import org.apache.paimon.flink.action.cdc.ComputedColumn;
import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.format.RecordParser;
+import org.apache.paimon.flink.action.cdc.format.RecordParser;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
import org.apache.paimon.types.RowKind;
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/ogg/OggRecordParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/ogg/OggRecordParser.java
similarity index 97%
rename from paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/ogg/OggRecordParser.java
rename to paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/ogg/OggRecordParser.java
index bce6199a9..04cdf03c0 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/format/ogg/OggRecordParser.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/ogg/OggRecordParser.java
@@ -16,11 +16,11 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.action.cdc.kafka.format.ogg;
+package org.apache.paimon.flink.action.cdc.format.ogg;
import org.apache.paimon.flink.action.cdc.ComputedColumn;
import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.format.RecordParser;
+import org.apache.paimon.flink.action.cdc.format.RecordParser;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
import org.apache.paimon.types.RowKind;
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionUtils.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionUtils.java
index 7a5f46bf7..3eace2574 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionUtils.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionUtils.java
@@ -18,6 +18,8 @@
package org.apache.paimon.flink.action.cdc.kafka;
+import org.apache.paimon.flink.action.cdc.MessageQueueSchemaUtils;
+import org.apache.paimon.flink.action.cdc.format.DataFormat;
import org.apache.paimon.utils.StringUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
@@ -32,10 +34,19 @@ import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.util.CollectionUtil;
import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
+import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
@@ -177,7 +188,7 @@ class KafkaActionUtils {
*
* @return specificOffsets with Map format, key is partition, and value is offset
*/
- public static Map<Integer, Long> parseSpecificOffsets(
+ private static Map<Integer, Long> parseSpecificOffsets(
String specificOffsetsStr, String optionKey) {
final Map<Integer, Long> offsetMap = new HashMap<>();
final String[] pairs = specificOffsetsStr.split(";");
@@ -235,7 +246,7 @@ class KafkaActionUtils {
KafkaConnectorOptions.PROPS_BOOTSTRAP_SERVERS.key()));
}
- public static String kafkaPropertiesGroupId(Configuration kafkaConfig) {
+ private static String kafkaPropertiesGroupId(Configuration kafkaConfig) {
String groupId = kafkaConfig.get(KafkaConnectorOptions.PROPS_GROUP_ID);
if (StringUtils.isEmpty(groupId)) {
groupId = UUID.randomUUID().toString();
@@ -243,4 +254,62 @@ class KafkaActionUtils {
}
return groupId;
}
+
+ static DataFormat getDataFormat(Configuration kafkaConfig) {
+ return DataFormat.fromConfigString(kafkaConfig.get(KafkaConnectorOptions.VALUE_FORMAT));
+ }
+
+ static MessageQueueSchemaUtils.ConsumerWrapper getKafkaEarliestConsumer(
+ Configuration kafkaConfig, String topic) {
+ Properties props = new Properties();
+ props.put(
+ ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
+ kafkaConfig.get(KafkaConnectorOptions.PROPS_BOOTSTRAP_SERVERS));
+ props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaPropertiesGroupId(kafkaConfig));
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ props.put(
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+
+ KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
+
+ List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
+ if (partitionInfos.isEmpty()) {
+ throw new IllegalArgumentException(
+ "Failed to find partition information for topic " + topic);
+ }
+ int firstPartition =
+ partitionInfos.stream().map(PartitionInfo::partition).sorted().findFirst().get();
+ Collection<TopicPartition> topicPartitions =
+ Collections.singletonList(new TopicPartition(topic, firstPartition));
+ consumer.assign(topicPartitions);
+ consumer.seekToBeginning(topicPartitions);
+
+ return new KafkaConsumerWrapper(consumer);
+ }
+
+ private static class KafkaConsumerWrapper implements MessageQueueSchemaUtils.ConsumerWrapper {
+
+ private final KafkaConsumer<String, String> consumer;
+
+ KafkaConsumerWrapper(KafkaConsumer<String, String> kafkaConsumer) {
+ this.consumer = kafkaConsumer;
+ }
+
+ @Override
+ public List<String> getRecords(String topic, int pollTimeOutMills) {
+ ConsumerRecords<String, String> consumerRecords =
+ consumer.poll(Duration.ofMillis(pollTimeOutMills));
+ Iterable<ConsumerRecord<String, String>> records = consumerRecords.records(topic);
+ List<String> result = new ArrayList<>();
+ records.forEach(r -> result.add(r.value()));
+ return result;
+ }
+
+ @Override
+ public void close() {
+ consumer.close();
+ }
+ }
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSchemaUtils.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSchemaUtils.java
deleted file mode 100644
index 7ed9a4aab..000000000
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSchemaUtils.java
+++ /dev/null
@@ -1,146 +0,0 @@
-/*
- * 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.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.format.DataFormat;
-import org.apache.paimon.flink.action.cdc.kafka.format.RecordParser;
-import org.apache.paimon.schema.Schema;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.common.PartitionInfo;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.serialization.StringDeserializer;
-
-import java.time.Duration;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.Properties;
-import java.util.stream.Stream;
-import java.util.stream.StreamSupport;
-
-import static org.apache.paimon.flink.action.cdc.kafka.KafkaActionUtils.kafkaPropertiesGroupId;
-import static org.apache.paimon.flink.action.cdc.kafka.format.DataFormat.getDataFormat;
-
-/** Utility class to load kafka schema. */
-public class KafkaSchemaUtils {
-
- private static final int MAX_RETRY = 5;
- private static final int POLL_TIMEOUT_MILLIS = 1000;
-
- private static KafkaConsumer<String, String> getKafkaEarliestConsumer(
- Configuration kafkaConfig, String topic) {
- Properties props = new Properties();
- props.put(
- ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
- kafkaConfig.get(KafkaConnectorOptions.PROPS_BOOTSTRAP_SERVERS));
- props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaPropertiesGroupId(kafkaConfig));
- props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
- props.put(
- ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
- props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
- props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
-
- KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
-
- List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
- if (partitionInfos.isEmpty()) {
- throw new IllegalArgumentException(
- "Failed to find partition information for topic " + topic);
- }
- int firstPartition =
- partitionInfos.stream().map(PartitionInfo::partition).sorted().findFirst().get();
- Collection<TopicPartition> topicPartitions =
- Collections.singletonList(new TopicPartition(topic, firstPartition));
- consumer.assign(topicPartitions);
- consumer.seekToBeginning(topicPartitions);
-
- return consumer;
- }
-
- /**
- * Retrieves the Kafka schema for a given topic.
- *
- * @param kafkaConfig The configuration for Kafka.
- * @param topic The topic to retrieve the schema for.
- * @param typeMapping data type mapping options.
- * @return The Kafka schema for the topic.
- * @throws KafkaSchemaRetrievalException If unable to retrieve the schema after max retries.
- */
- public static Schema getKafkaSchema(
- Configuration kafkaConfig, String topic, TypeMapping typeMapping, boolean caseSensitive)
- throws KafkaSchemaRetrievalException {
- KafkaConsumer<String, String> consumer = getKafkaEarliestConsumer(kafkaConfig, topic);
- int retry = 0;
- int retryInterval = 1000;
-
- DataFormat format = getDataFormat(kafkaConfig);
- RecordParser recordParser = format.createParser(true, typeMapping, Collections.emptyList());
-
- while (true) {
- ConsumerRecords<String, String> consumerRecords =
- consumer.poll(Duration.ofMillis(POLL_TIMEOUT_MILLIS));
- Iterable<ConsumerRecord<String, String>> records = consumerRecords.records(topic);
- Stream<ConsumerRecord<String, String>> recordStream =
- StreamSupport.stream(records.spliterator(), false);
-
- Optional<Schema> kafkaSchema =
- recordStream
- .map(record -> recordParser.getKafkaSchema(record.value()))
- .filter(Objects::nonNull)
- .findFirst();
-
- if (kafkaSchema.isPresent()) {
- return kafkaSchema.get();
- }
-
- if (retry >= MAX_RETRY) {
- throw new KafkaSchemaRetrievalException(
- String.format("Could not get metadata from server, topic: %s", topic));
- }
-
- sleepSafely(retryInterval);
- retryInterval *= 2;
- retry++;
- }
- }
-
- private static void sleepSafely(int duration) {
- try {
- Thread.sleep(duration);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- }
-
- /** Custom exception to indicate issues with Kafka schema retrieval. */
- public static class KafkaSchemaRetrievalException extends Exception {
- public KafkaSchemaRetrievalException(String message) {
- super(message);
- }
- }
-}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
index e8df881fb..e7156dd3a 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
@@ -24,8 +24,8 @@ import org.apache.paimon.flink.action.ActionBase;
import org.apache.paimon.flink.action.MultiTablesSinkMode;
import org.apache.paimon.flink.action.cdc.TableNameConverter;
import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.format.DataFormat;
-import org.apache.paimon.flink.action.cdc.kafka.format.RecordParser;
+import org.apache.paimon.flink.action.cdc.format.DataFormat;
+import org.apache.paimon.flink.action.cdc.format.RecordParser;
import org.apache.paimon.flink.sink.cdc.EventParser;
import org.apache.paimon.flink.sink.cdc.FlinkCdcSyncDatabaseSinkBuilder;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
@@ -44,6 +44,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.regex.Pattern;
+import static org.apache.paimon.flink.action.cdc.kafka.KafkaActionUtils.getDataFormat;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/**
@@ -151,7 +152,7 @@ public class KafkaSyncDatabaseAction extends ActionBase {
KafkaSource<String> source = KafkaActionUtils.buildKafkaSource(kafkaConfig);
- DataFormat format = DataFormat.getDataFormat(kafkaConfig);
+ DataFormat format = getDataFormat(kafkaConfig);
RecordParser recordParser =
format.createParser(caseSensitive, typeMapping, Collections.emptyList());
RichCdcMultiplexRecordSchemaBuilder schemaBuilder =
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java
index 4de611b30..06a51aafd 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java
@@ -18,85 +18,18 @@
package org.apache.paimon.flink.action.cdc.kafka;
-import org.apache.paimon.annotation.VisibleForTesting;
-import org.apache.paimon.catalog.Catalog;
-import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.flink.FlinkConnectorOptions;
-import org.apache.paimon.flink.action.Action;
-import org.apache.paimon.flink.action.ActionBase;
-import org.apache.paimon.flink.action.cdc.CdcMetadataConverter;
-import org.apache.paimon.flink.action.cdc.ComputedColumn;
-import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.format.DataFormat;
-import org.apache.paimon.flink.action.cdc.kafka.format.RecordParser;
-import org.apache.paimon.flink.sink.cdc.CdcSinkBuilder;
-import org.apache.paimon.flink.sink.cdc.EventParser;
-import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
-import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecordEventParser;
+import org.apache.paimon.flink.action.cdc.MessageQueueSchemaUtils;
+import org.apache.paimon.flink.action.cdc.MessageQueueSyncTableActionBase;
+import org.apache.paimon.flink.action.cdc.format.DataFormat;
import org.apache.paimon.schema.Schema;
-import org.apache.paimon.table.FileStoreTable;
-import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.kafka.source.KafkaSource;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.api.connector.source.Source;
import org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
import java.util.Map;
-import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.assertSchemaCompatible;
-import static org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.buildPaimonSchema;
-import static org.apache.paimon.flink.action.cdc.ComputedColumnUtils.buildComputedColumns;
-
-/**
- * An {@link Action} which synchronize one kafka topic into one Paimon table.
- *
- * <p>This topic must be from canal-json format.
- *
- * <p>You should specify Kafka source topic in {@code kafkaConfig}. See <a
- * href="https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/table/kafka/">document
- * of flink-connectors</a> for detailed keys and values.
- *
- * <p>If the specified Paimon table does not exist, this action will automatically create the table.
- * Its schema will be derived from all specified Kafka topic. If the Paimon table already exists,
- * its schema will be compared against the schema of all specified Kafka topic.
- *
- * <p>This action supports a limited number of schema changes. Unsupported schema changes will be
- * ignored. Currently supported schema changes includes:
- *
- * <ul>
- * <li>Adding columns.
- * <li>Altering column types. More specifically,
- * <ul>
- * <li>altering from a string type (char, varchar, text) to another string type with longer
- * length,
- * <li>altering from a binary type (binary, varbinary, blob) to another binary type with
- * longer length,
- * <li>altering from an integer type (tinyint, smallint, int, bigint) to another integer
- * type with wider range,
- * <li>altering from a floating-point type (float, double) to another floating-point type
- * with wider range,
- * </ul>
- * are supported. Other type changes will cause exceptions.
- * </ul>
- */
-public class KafkaSyncTableAction extends ActionBase {
-
- private final String database;
- private final String table;
- private final Configuration kafkaConfig;
- private FileStoreTable fileStoreTable;
-
- private List<String> partitionKeys = new ArrayList<>();
- private List<String> primaryKeys = new ArrayList<>();
-
- private Map<String, String> tableConfig = new HashMap<>();
- private List<String> computedColumnArgs = new ArrayList<>();
- private TypeMapping typeMapping = TypeMapping.defaultMapping();
+/** Synchronize table from Kafka. */
+public class KafkaSyncTableAction extends MessageQueueSyncTableActionBase {
public KafkaSyncTableAction(
String warehouse,
@@ -104,118 +37,37 @@ public class KafkaSyncTableAction extends ActionBase {
String table,
Map<String, String> catalogConfig,
Map<String, String> kafkaConfig) {
- super(warehouse, catalogConfig);
- this.database = database;
- this.table = table;
- this.kafkaConfig = Configuration.fromMap(kafkaConfig);
- }
-
- public KafkaSyncTableAction withPartitionKeys(String... partitionKeys) {
- return withPartitionKeys(Arrays.asList(partitionKeys));
- }
-
- public KafkaSyncTableAction withPartitionKeys(List<String> partitionKeys) {
- this.partitionKeys = partitionKeys;
- return this;
- }
-
- public KafkaSyncTableAction withPrimaryKeys(String... primaryKeys) {
- return withPrimaryKeys(Arrays.asList(primaryKeys));
+ super(warehouse, database, table, catalogConfig, kafkaConfig);
}
- public KafkaSyncTableAction withPrimaryKeys(List<String> primaryKeys) {
- this.primaryKeys = primaryKeys;
- return this;
- }
-
- public KafkaSyncTableAction withTableConfig(Map<String, String> tableConfig) {
- this.tableConfig = tableConfig;
- return this;
- }
-
- public KafkaSyncTableAction withComputedColumnArgs(List<String> computedColumnArgs) {
- this.computedColumnArgs = computedColumnArgs;
- return this;
- }
-
- public KafkaSyncTableAction withTypeMapping(TypeMapping typeMapping) {
- this.typeMapping = typeMapping;
- return this;
+ @Override
+ protected Source<String, ?, ?> buildSource() {
+ return KafkaActionUtils.buildKafkaSource(mqConfig);
}
@Override
- public void build(StreamExecutionEnvironment env) throws Exception {
- KafkaSource<String> source = KafkaActionUtils.buildKafkaSource(kafkaConfig);
- String topic = kafkaConfig.get(KafkaConnectorOptions.TOPIC).get(0);
-
- catalog.createDatabase(database, true);
- boolean caseSensitive = catalog.caseSensitive();
- Schema kafkaSchema =
- KafkaSchemaUtils.getKafkaSchema(kafkaConfig, topic, typeMapping, caseSensitive);
-
- Identifier identifier = new Identifier(database, table);
- List<ComputedColumn> computedColumns =
- buildComputedColumns(computedColumnArgs, kafkaSchema);
- Schema fromKafka =
- buildPaimonSchema(
- partitionKeys,
- primaryKeys,
- computedColumns,
- tableConfig,
- kafkaSchema,
- new CdcMetadataConverter[] {});
-
- try {
- fileStoreTable = (FileStoreTable) catalog.getTable(identifier);
- fileStoreTable = fileStoreTable.copy(tableConfig);
- assertSchemaCompatible(fileStoreTable.schema(), fromKafka.fields());
- } catch (Catalog.TableNotExistException e) {
- catalog.createTable(identifier, fromKafka, false);
- fileStoreTable = (FileStoreTable) catalog.getTable(identifier);
- }
- DataFormat format = DataFormat.getDataFormat(kafkaConfig);
- RecordParser recordParser =
- format.createParser(caseSensitive, typeMapping, computedColumns);
- EventParser.Factory<RichCdcMultiplexRecord> parserFactory =
- () -> new RichCdcMultiplexRecordEventParser(caseSensitive);
-
- CdcSinkBuilder<RichCdcMultiplexRecord> sinkBuilder =
- new CdcSinkBuilder<RichCdcMultiplexRecord>()
- .withInput(
- env.fromSource(
- source,
- WatermarkStrategy.noWatermarks(),
- "Kafka Source")
- .flatMap(recordParser))
- .withParserFactory(parserFactory)
- .withTable(fileStoreTable)
- .withIdentifier(identifier)
- .withCatalogLoader(catalogLoader());
- String sinkParallelism = tableConfig.get(FlinkConnectorOptions.SINK_PARALLELISM.key());
- if (sinkParallelism != null) {
- sinkBuilder.withParallelism(Integer.parseInt(sinkParallelism));
+ protected Schema buildSchema() {
+ String topic = mqConfig.get(KafkaConnectorOptions.TOPIC).get(0);
+ try (MessageQueueSchemaUtils.ConsumerWrapper consumer =
+ KafkaActionUtils.getKafkaEarliestConsumer(mqConfig, topic)) {
+ return MessageQueueSchemaUtils.getSchema(consumer, topic, getDataFormat(), typeMapping);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
}
- sinkBuilder.build();
}
- @VisibleForTesting
- public Map<String, String> tableConfig() {
- return tableConfig;
+ @Override
+ protected DataFormat getDataFormat() {
+ return KafkaActionUtils.getDataFormat(mqConfig);
}
- @VisibleForTesting
- public FileStoreTable fileStoreTable() {
- return fileStoreTable;
+ @Override
+ protected String sourceName() {
+ return "Kafka Source";
}
- // ------------------------------------------------------------------------
- // Flink run methods
- // ------------------------------------------------------------------------
-
@Override
- public void run() throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- build(env);
- env.execute(String.format("Kafka-Paimon Table Sync: %s.%s", database, table));
+ protected String jobName() {
+ return String.format("Kafka-Paimon Table Sync: %s.%s", database, table);
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionFactory.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionFactory.java
index 3af27df4c..08663d634 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionFactory.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionFactory.java
@@ -45,12 +45,12 @@ public class KafkaSyncTableActionFactory implements ActionFactory {
KafkaSyncTableAction action =
new KafkaSyncTableAction(
- tablePath.f0,
- tablePath.f1,
- tablePath.f2,
- optionalConfigMap(params, "catalog-conf"),
- optionalConfigMap(params, "kafka-conf"))
- .withTableConfig(optionalConfigMap(params, "table-conf"));
+ tablePath.f0,
+ tablePath.f1,
+ tablePath.f2,
+ optionalConfigMap(params, "catalog-conf"),
+ optionalConfigMap(params, "kafka-conf"));
+ action.withTableConfig(optionalConfigMap(params, "table-conf"));
if (params.has("partition-keys")) {
action.withPartitionKeys(params.get("partition-keys").split(","));
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
new file mode 100644
index 000000000..da9a1d1f1
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionUtils.java
@@ -0,0 +1,386 @@
+/*
+ * 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.flink.action.cdc.MessageQueueSchemaUtils;
+import org.apache.paimon.flink.action.cdc.format.DataFormat;
+
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.PulsarClientFactory;
+import org.apache.flink.connector.pulsar.common.config.PulsarOptions;
+import org.apache.flink.connector.pulsar.source.PulsarSource;
+import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
+import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
+import org.apache.flink.connector.pulsar.source.reader.PulsarPartitionSplitReader;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.KeySharedPolicy;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.internal.DefaultImplementation;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.CLIENT_CONFIG_PREFIX;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAMS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_CONSUMER_NAME;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.createConsumerBuilder;
+import static org.apache.flink.connector.pulsar.source.enumerator.topic.range.TopicRangeUtils.isFullTopicRanges;
+import static org.apache.paimon.flink.action.ActionFactory.parseCommaSeparatedKeyValues;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.pulsar.client.api.KeySharedPolicy.stickyHashRange;
+
+/** Utils for Pulsar synchronization. */
+public class PulsarActionUtils {
+
+ static final ConfigOption<String> VALUE_FORMAT =
+ ConfigOptions.key("value.format")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Defines the format identifier for encoding value data.");
+
+ static final ConfigOption<String> TOPIC =
+ ConfigOptions.key("topic")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Topic names from which the table is read. Either 'topic' or 'topic-pattern' must be set for source. "
+ + "Option 'topic' is required for sink.");
+
+ static final ConfigOption<String> PULSAR_AUTH_PARAM_MAP =
+ ConfigOptions.key(CLIENT_CONFIG_PREFIX + "authParamMap")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Parameters for the authentication plugin.");
+
+ static final ConfigOption<String> PULSAR_START_CURSOR_FROM_MESSAGE_ID =
+ ConfigOptions.key("pulsar.startCursor.fromMessageId")
+ .stringType()
+ .defaultValue("EARLIEST")
+ .withDescription(
+ "Using a unique identifier of a single message to seek the start position. "
+ + "The common format is a triple '<long>ledgerId,<long>entryId,<int>partitionIndex'. "
+ + "Specially, you can set it to EARLIEST (-1, -1, -1) or LATEST (Long.MAX_VALUE, Long.MAX_VALUE, -1).");
+
+ static final ConfigOption<Long> PULSAR_START_CURSOR_FORM_PUBLISH_TIME =
+ ConfigOptions.key("pulsar.startCursor.fromPublishTime")
+ .longType()
+ .noDefaultValue()
+ .withDescription("Using the message publish time to seek the start position.");
+
+ static final ConfigOption<Boolean> PULSAR_START_CURSOR_FROM_MESSAGE_ID_INCLUSIVE =
+ ConfigOptions.key("pulsar.startCursor.fromMessageIdInclusive")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Whether to include the given message id. This option only works when the message id is not EARLIEST or LATEST.");
+
+ static final ConfigOption<String> PULSAR_STOP_CURSOR_AT_MESSAGE_ID =
+ ConfigOptions.key("pulsar.stopCursor.atMessageId")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Stop consuming when the message id is equal or greater than the specified message id. "
+ + "Message that is equal to the specified message id will not be consumed. "
+ + "The common format is a triple '<long>ledgerId,<long>entryId,<int>partitionIndex'. "
+ + "Specially, you can set it to LATEST (Long.MAX_VALUE, Long.MAX_VALUE, -1).");
+
+ static final ConfigOption<String> PULSAR_STOP_CURSOR_AFTER_MESSAGE_ID =
+ ConfigOptions.key("pulsar.stopCursor.afterMessageId")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Stop consuming when the message id is greater than the specified message id. "
+ + "Message that is equal to the specified message id will be consumed. "
+ + "The common format is a triple '<long>ledgerId,<long>entryId,<int>partitionIndex'. "
+ + "Specially, you can set it to LATEST (Long.MAX_VALUE, Long.MAX_VALUE, -1).");
+
+ static final ConfigOption<Long> PULSAR_STOP_CURSOR_AT_EVENT_TIME =
+ ConfigOptions.key("pulsar.stopCursor.atEventTime")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "Stop consuming when message event time is greater than or equals the specified timestamp. "
+ + "Message that even time is equal to the specified timestamp will not be consumed.");
+
+ static final ConfigOption<Long> PULSAR_STOP_CURSOR_AFTER_EVENT_TIME =
+ ConfigOptions.key("pulsar.stopCursor.afterEventTime")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "Stop consuming when message event time is greater than the specified timestamp. "
+ + "Message that even time is equal to the specified timestamp will be consumed.");
+
+ static final ConfigOption<Boolean> PULSAR_SOURCE_UNBOUNDED =
+ ConfigOptions.key("pulsar.source.unbounded")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription("To specify the boundedness of a stream.");
+
+ static PulsarSource<String> buildPulsarSource(Configuration rawConfig) {
+ Configuration pulsarConfig = preprocessPulsarConfig(rawConfig);
+ validatePulsarConfig(pulsarConfig);
+
+ PulsarSourceBuilder<String> pulsarSourceBuilder = PulsarSource.builder();
+
+ // the minimum setup
+ pulsarSourceBuilder
+ .setServiceUrl(pulsarConfig.get(PULSAR_SERVICE_URL))
+ .setAdminUrl(pulsarConfig.get(PULSAR_ADMIN_URL))
+ .setSubscriptionName(pulsarConfig.get(PULSAR_SUBSCRIPTION_NAME))
+ .setTopics(
+ Arrays.stream(pulsarConfig.get(TOPIC).split(","))
+ .map(String::trim)
+ .collect(Collectors.toList()))
+ .setDeserializationSchema(new SimpleStringSchema());
+
+ // other settings
+
+ // consumer name
+ pulsarConfig
+ .getOptional(PULSAR_CONSUMER_NAME)
+ .ifPresent(pulsarSourceBuilder::setConsumerName);
+
+ // start cursor
+ if (pulsarConfig.contains(PULSAR_START_CURSOR_FORM_PUBLISH_TIME)) {
+ checkArgument(!pulsarConfig.contains(PULSAR_START_CURSOR_FROM_MESSAGE_ID), "");
+ pulsarSourceBuilder.setStartCursor(
+ StartCursor.fromPublishTime(
+ pulsarConfig.get(PULSAR_START_CURSOR_FORM_PUBLISH_TIME)));
+ } else {
+ String messageId = pulsarConfig.get(PULSAR_START_CURSOR_FROM_MESSAGE_ID);
+ if (messageId.equalsIgnoreCase("EARLIEST")) {
+ pulsarSourceBuilder.setStartCursor(StartCursor.earliest());
+ } else if (messageId.equalsIgnoreCase("LATEST")) {
+ pulsarSourceBuilder.setStartCursor(StartCursor.latest());
+ } else {
+ StartCursor startCursor =
+ StartCursor.fromMessageId(
+ toMessageId(messageId),
+ pulsarConfig.get(PULSAR_START_CURSOR_FROM_MESSAGE_ID_INCLUSIVE));
+ pulsarSourceBuilder.setStartCursor(startCursor);
+ }
+ }
+
+ // boundedness and stop cursor
+ StopCursor stopCursor = StopCursor.never();
+ int stopCursorSet = 0;
+ if (pulsarConfig.contains(PULSAR_STOP_CURSOR_AT_MESSAGE_ID)) {
+ stopCursor =
+ StopCursor.atMessageId(
+ toMessageId(pulsarConfig.get(PULSAR_STOP_CURSOR_AT_MESSAGE_ID)));
+ stopCursorSet++;
+ }
+ if (pulsarConfig.contains(PULSAR_STOP_CURSOR_AFTER_MESSAGE_ID)) {
+ stopCursor =
+ StopCursor.afterMessageId(
+ toMessageId(pulsarConfig.get(PULSAR_STOP_CURSOR_AT_MESSAGE_ID)));
+ stopCursorSet++;
+ }
+ if (pulsarConfig.contains(PULSAR_STOP_CURSOR_AT_EVENT_TIME)) {
+ stopCursor = StopCursor.atEventTime(pulsarConfig.get(PULSAR_STOP_CURSOR_AT_EVENT_TIME));
+ stopCursorSet++;
+ }
+ if (pulsarConfig.contains(PULSAR_STOP_CURSOR_AFTER_EVENT_TIME)) {
+ stopCursor =
+ StopCursor.atEventTime(pulsarConfig.get(PULSAR_STOP_CURSOR_AFTER_EVENT_TIME));
+ stopCursorSet++;
+ }
+ checkArgument(stopCursorSet <= 1, "You can set at most one of the stop cursor options.");
+
+ if (pulsarConfig.get(PULSAR_SOURCE_UNBOUNDED)) {
+ pulsarSourceBuilder.setUnboundedStopCursor(stopCursor);
+ } else {
+ pulsarSourceBuilder.setBoundedStopCursor(stopCursor);
+ }
+
+ // auth
+ String authPluginClassName = pulsarConfig.get(PULSAR_AUTH_PLUGIN_CLASS_NAME);
+ if (authPluginClassName != null) {
+ String authParamsString = pulsarConfig.get(PULSAR_AUTH_PARAMS);
+ Map<String, String> authParamsMap =
+ pulsarConfig.get(PulsarOptions.PULSAR_AUTH_PARAM_MAP);
+
+ checkArgument(
+ authParamsString != null || authParamsMap != null,
+ "You should set '%s' or '%s'",
+ PULSAR_AUTH_PARAMS.key(),
+ PULSAR_AUTH_PARAM_MAP.key());
+ checkArgument(
+ authParamsString == null || authParamsMap == null,
+ "You can only set one of '%s' and '%s'",
+ PULSAR_AUTH_PARAMS.key(),
+ PULSAR_AUTH_PARAM_MAP.key());
+
+ if (authParamsString != null) {
+ pulsarSourceBuilder.setAuthentication(authPluginClassName, authParamsString);
+ } else {
+ pulsarSourceBuilder.setAuthentication(authPluginClassName, authParamsMap);
+ }
+ }
+
+ // set all options as additional pulsar options and pulsar source options
+ pulsarSourceBuilder.setConfig(pulsarConfig);
+
+ // TODO set RangeGenerator; set PulsarCrypto
+
+ return pulsarSourceBuilder.build();
+ }
+
+ private static void validatePulsarConfig(Configuration pulsarConfig) {
+ checkArgument(
+ pulsarConfig.contains(VALUE_FORMAT),
+ String.format("pulsar-conf [%s] must be specified.", VALUE_FORMAT.key()));
+
+ checkArgument(
+ pulsarConfig.contains(PULSAR_SERVICE_URL),
+ String.format("pulsar-conf [%s] must be specified.", PULSAR_SERVICE_URL.key()));
+
+ checkArgument(
+ pulsarConfig.contains(PULSAR_ADMIN_URL),
+ String.format("pulsar-conf [%s] must be specified.", PULSAR_ADMIN_URL.key()));
+
+ checkArgument(
+ pulsarConfig.contains(PULSAR_SUBSCRIPTION_NAME),
+ String.format(
+ "pulsar-conf [%s] must be specified.", PULSAR_SUBSCRIPTION_NAME.key()));
+
+ checkArgument(
+ pulsarConfig.contains(TOPIC),
+ String.format("pulsar-conf [%s] must be specified.", TOPIC.key()));
+ }
+
+ private static MessageId toMessageId(String messageIdString) {
+ if (messageIdString.equalsIgnoreCase("EARLIEST")) {
+ return MessageId.earliest;
+ } else if (messageIdString.equalsIgnoreCase("LATEST")) {
+ return MessageId.latest;
+ } else {
+ String[] splits = messageIdString.split(",");
+ checkArgument(
+ splits.length == 3,
+ "Please use format '<long>ledgerId,<long>entryId,<int>partitionIndex' for message id");
+ return DefaultImplementation.getDefaultImplementation()
+ .newMessageId(
+ Long.parseLong(splits[0].trim()),
+ Long.parseLong(splits[1].trim()),
+ Integer.parseInt(splits[2].trim()));
+ }
+ }
+
+ static SourceConfiguration toSourceConfiguration(Configuration rawConfig) {
+ return new SourceConfiguration(preprocessPulsarConfig(rawConfig));
+ }
+
+ private static Configuration preprocessPulsarConfig(Configuration rawConfig) {
+ Configuration cloned = new Configuration(rawConfig);
+ if (cloned.contains(PULSAR_AUTH_PARAM_MAP)) {
+ Map<String, String> authParamsMap =
+ parseCommaSeparatedKeyValues(cloned.get(PULSAR_AUTH_PARAM_MAP));
+ cloned.removeConfig(PULSAR_AUTH_PARAM_MAP);
+ cloned.set(PulsarOptions.PULSAR_AUTH_PARAM_MAP, authParamsMap);
+ }
+ return cloned;
+ }
+
+ static DataFormat getDataFormat(Configuration pulsarConfig) {
+ return DataFormat.fromConfigString(pulsarConfig.get(VALUE_FORMAT));
+ }
+
+ /** Referenced to {@link PulsarPartitionSplitReader#createPulsarConsumer}. */
+ static MessageQueueSchemaUtils.ConsumerWrapper createPulsarConsumer(
+ Configuration pulsarConfig, String topic) throws PulsarClientException {
+ SourceConfiguration pulsarSourceConfiguration =
+ toSourceConfiguration(preprocessPulsarConfig(pulsarConfig));
+ PulsarClient pulsarClient = PulsarClientFactory.createClient(pulsarSourceConfiguration);
+
+ ConsumerBuilder<String> consumerBuilder =
+ createConsumerBuilder(
+ pulsarClient,
+ org.apache.pulsar.client.api.Schema.STRING,
+ pulsarSourceConfiguration);
+
+ // The default position is Latest
+ consumerBuilder.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+
+ TopicPartition topicPartition = new TopicPartition(topic);
+ consumerBuilder.topic(topicPartition.getFullTopicName());
+
+ // TODO currently, PulsarCrypto is not supported
+
+ // Add KeySharedPolicy for partial keys subscription.
+ if (!isFullTopicRanges(topicPartition.getRanges())) {
+ KeySharedPolicy policy = stickyHashRange().ranges(topicPartition.getPulsarRanges());
+ // We may enable out of order delivery for speeding up. It was turned off by
+ // default.
+ policy.setAllowOutOfOrderDelivery(
+ pulsarSourceConfiguration.isAllowKeySharedOutOfOrderDelivery());
+ consumerBuilder.keySharedPolicy(policy);
+ }
+
+ // Create the consumer configuration by using common utils.
+ Consumer<String> consumer = consumerBuilder.subscribe();
+
+ return new PulsarConsumerWrapper(consumer);
+ }
+
+ private static class PulsarConsumerWrapper implements MessageQueueSchemaUtils.ConsumerWrapper {
+
+ private final Consumer<String> consumer;
+
+ PulsarConsumerWrapper(Consumer<String> consumer) {
+ this.consumer = consumer;
+ }
+
+ @Override
+ public List<String> getRecords(String topic, int pollTimeOutMills) {
+ try {
+ Message<String> message = consumer.receive(pollTimeOutMills, TimeUnit.MILLISECONDS);
+ return message == null
+ ? Collections.emptyList()
+ : Collections.singletonList(message.getValue());
+ } catch (PulsarClientException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void close() throws PulsarClientException {
+ consumer.close();
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableAction.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableAction.java
new file mode 100644
index 000000000..6221ff8d4
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableAction.java
@@ -0,0 +1,72 @@
+/*
+ * 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.flink.action.cdc.MessageQueueSchemaUtils;
+import org.apache.paimon.flink.action.cdc.MessageQueueSyncTableActionBase;
+import org.apache.paimon.flink.action.cdc.format.DataFormat;
+import org.apache.paimon.schema.Schema;
+
+import org.apache.flink.api.connector.source.Source;
+
+import java.util.Map;
+
+/** Synchronize table from Pulsar. */
+public class PulsarSyncTableAction extends MessageQueueSyncTableActionBase {
+
+ public PulsarSyncTableAction(
+ String warehouse,
+ String database,
+ String table,
+ Map<String, String> catalogConfig,
+ Map<String, String> pulsarConfig) {
+ super(warehouse, database, table, catalogConfig, pulsarConfig);
+ }
+
+ @Override
+ protected Source<String, ?, ?> buildSource() {
+ return PulsarActionUtils.buildPulsarSource(mqConfig);
+ }
+
+ @Override
+ protected Schema buildSchema() {
+ String topic = mqConfig.get(PulsarActionUtils.TOPIC).split(",")[0].trim();
+ try (MessageQueueSchemaUtils.ConsumerWrapper consumer =
+ PulsarActionUtils.createPulsarConsumer(mqConfig, topic)) {
+ return MessageQueueSchemaUtils.getSchema(consumer, topic, getDataFormat(), typeMapping);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ protected DataFormat getDataFormat() {
+ return PulsarActionUtils.getDataFormat(mqConfig);
+ }
+
+ @Override
+ protected String sourceName() {
+ return "Pulsar Source";
+ }
+
+ @Override
+ protected String jobName() {
+ return String.format("Pulsar-Paimon Table Sync: %s.%s", database, table);
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableActionFactory.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableActionFactory.java
new file mode 100644
index 000000000..0a19763be
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableActionFactory.java
@@ -0,0 +1,85 @@
+/*
+ * 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.flink.action.Action;
+import org.apache.paimon.flink.action.ActionFactory;
+import org.apache.paimon.flink.action.cdc.TypeMapping;
+
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.utils.MultipleParameterTool;
+
+import java.util.ArrayList;
+import java.util.Optional;
+
+/** Factory to create {@link PulsarSyncTableAction}. */
+public class PulsarSyncTableActionFactory implements ActionFactory {
+
+ public static final String IDENTIFIER = "pulsar-sync-table";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Optional<Action> create(MultipleParameterTool params) {
+ Tuple3<String, String, String> tablePath = getTablePath(params);
+ checkRequiredArgument(params, "pulsar-conf");
+
+ PulsarSyncTableAction action =
+ new PulsarSyncTableAction(
+ tablePath.f0,
+ tablePath.f1,
+ tablePath.f2,
+ optionalConfigMap(params, "catalog-conf"),
+ optionalConfigMap(params, "pulsar-conf"));
+ action.withTableConfig(optionalConfigMap(params, "table-conf"));
+
+ if (params.has("partition-keys")) {
+ action.withPartitionKeys(params.get("partition-keys").split(","));
+ }
+
+ if (params.has("primary-keys")) {
+ action.withPrimaryKeys(params.get("primary-keys").split(","));
+ }
+
+ if (params.has("computed-column")) {
+ action.withComputedColumnArgs(
+ new ArrayList<>(params.getMultiParameter("computed-column")));
+ }
+
+ if (params.has("type-mapping")) {
+ String[] options = params.get("type-mapping").split(",");
+ action.withTypeMapping(TypeMapping.parse(options));
+ }
+
+ return Optional.of(action);
+ }
+
+ @Override
+ public void printHelp() {
+ System.out.println(
+ "Action \"pulsar-sync-table\" creates a streaming job "
+ + "with a Flink pulsar CDC source and a Paimon table sink to consume CDC events.");
+ System.out.println();
+
+ // TODO
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory b/paimon-flink/paimon-flink-cdc/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index 1e3a2ff65..d536d8c52 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++ b/paimon-flink/paimon-flink-cdc/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -20,5 +20,6 @@ org.apache.paimon.flink.action.cdc.mysql.MySqlSyncTableActionFactory
org.apache.paimon.flink.action.cdc.mysql.MySqlSyncDatabaseActionFactory
org.apache.paimon.flink.action.cdc.kafka.KafkaSyncTableActionFactory
org.apache.paimon.flink.action.cdc.kafka.KafkaSyncDatabaseActionFactory
+org.apache.paimon.flink.action.cdc.pulsar.PulsarSyncTableActionFactory
org.apache.paimon.flink.action.cdc.mongodb.MongoDBSyncTableActionFactory
org.apache.paimon.flink.action.cdc.mongodb.MongoDBSyncDatabaseActionFactory
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java
index 6dc85ce23..dfd698078 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java
@@ -69,7 +69,7 @@ import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
-/** Base test class for {@link org.apache.paimon.flink.action.Action}s related to MySQL. */
+/** Base test class for Kafka synchronization. */
public abstract class KafkaActionITCaseBase extends CdcActionITCaseBase {
private final ObjectMapper objectMapper = new ObjectMapper();
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
index 8d3440345..7738106fd 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncTableActionITCase.java
@@ -38,7 +38,7 @@ import static org.apache.paimon.testutils.assertj.AssertionUtils.anyCauseMatches
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-/** IT cases for {@link KafkaCanalSyncTableActionITCase}. */
+/** IT cases for {@link KafkaSyncTableAction}. */
public class KafkaCanalSyncTableActionITCase extends KafkaActionITCaseBase {
@Test
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSchemaITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSchemaITCase.java
index d3b80686a..e7113feb0 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSchemaITCase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSchemaITCase.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.action.cdc.kafka;
+import org.apache.paimon.flink.action.cdc.MessageQueueSchemaUtils;
import org.apache.paimon.flink.action.cdc.TypeMapping;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
@@ -34,9 +35,11 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static org.apache.paimon.flink.action.cdc.kafka.KafkaActionUtils.getDataFormat;
+import static org.apache.paimon.flink.action.cdc.kafka.KafkaActionUtils.getKafkaEarliestConsumer;
import static org.assertj.core.api.Assertions.assertThat;
-/** Tests for {@link KafkaSchemaUtils}. */
+/** Tests for building schema from Kafka. */
public class KafkaSchemaITCase extends KafkaActionITCaseBase {
@Test
@Timeout(60)
@@ -50,16 +53,16 @@ public class KafkaSchemaITCase extends KafkaActionITCaseBase {
} catch (Exception e) {
throw new Exception("Failed to write canal data to Kafka.", e);
}
- Map<String, String> kafkaConfig = getBasicKafkaConfig();
- kafkaConfig.put("value.format", "canal-json");
- kafkaConfig.put("topic", topic);
+ Configuration kafkaConfig = Configuration.fromMap(getBasicKafkaConfig());
+ kafkaConfig.setString("value.format", "canal-json");
+ kafkaConfig.setString("topic", topic);
Schema kafkaSchema =
- KafkaSchemaUtils.getKafkaSchema(
- Configuration.fromMap(kafkaConfig),
+ MessageQueueSchemaUtils.getSchema(
+ getKafkaEarliestConsumer(kafkaConfig, topic),
topic,
- TypeMapping.defaultMapping(),
- true);
+ getDataFormat(kafkaConfig),
+ TypeMapping.defaultMapping());
List<DataField> fields = new ArrayList<>();
fields.add(new DataField(0, "pt", DataTypes.INT()));
fields.add(new DataField(1, "_id", DataTypes.INT().notNull()));
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionITCaseBase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionITCaseBase.java
new file mode 100644
index 000000000..8172cd652
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionITCaseBase.java
@@ -0,0 +1,318 @@
+/*
+ * 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.flink.action.cdc.CdcActionITCaseBase;
+import org.apache.paimon.utils.StringUtils;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.flink.api.java.utils.MultipleParameterTool;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.AfterAllCallback;
+import org.junit.jupiter.api.extension.BeforeAllCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Network;
+import org.testcontainers.containers.PulsarContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
+import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicName;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.pulsar.client.api.ProducerAccessMode.Shared;
+import static org.apache.pulsar.common.partition.PartitionedTopicMetadata.NON_PARTITIONED;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.testcontainers.containers.PulsarContainer.BROKER_HTTP_PORT;
+import static org.testcontainers.containers.wait.strategy.Wait.forHttp;
+
+/**
+ * Base test class for Pulsar synchronization. *
+ *
+ * <p>Using Kafka test data.
+ */
+public class PulsarActionITCaseBase extends CdcActionITCaseBase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PulsarActionITCaseBase.class);
+
+ private static final Network NETWORK = Network.newNetwork();
+ private static final String INTER_CONTAINER_PULSAR_ALIAS = "pulsar";
+
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
+ private PulsarAdmin admin;
+ private PulsarClient client;
+
+ @RegisterExtension
+ public static final PulsarContainerExtension PULSAR_CONTAINER =
+ createPulsarContainerExtension();
+
+ private static PulsarContainerExtension createPulsarContainerExtension() {
+ PulsarContainerExtension container =
+ new PulsarContainerExtension("3.0.0") {
+ @Override
+ protected void doStart() {
+ super.doStart();
+ if (LOG.isInfoEnabled()) {
+ this.followOutput(new Slf4jLogConsumer(LOG));
+ }
+ }
+ };
+
+ container.withNetwork(NETWORK);
+ container.withNetworkAliases(INTER_CONTAINER_PULSAR_ALIAS);
+
+ // Override the default standalone configuration by system environments
+ container.withEnv("PULSAR_PREFIX_transactionCoordinatorEnabled", "true");
+ container.withEnv("PULSAR_PREFIX_acknowledgmentAtBatchIndexLevelEnabled", "true");
+ container.withEnv("PULSAR_PREFIX_systemTopicEnabled", "true");
+ container.withEnv("PULSAR_PREFIX_brokerDeduplicationEnabled", "true");
+ container.withEnv("PULSAR_PREFIX_defaultNumberOfNamespaceBundles", "1");
+
+ // Change the default bootstrap script, it will override the default configuration
+ // and start a standalone Pulsar without streaming storage and function worker
+ container.withCommand(
+ "sh",
+ "-c",
+ "/pulsar/bin/apply-config-from-env.py /pulsar/conf/standalone.conf && /pulsar/bin/pulsar standalone --no-functions-worker -nss");
+
+ // Waiting for the Pulsar broker and the transaction is ready after the container started
+ container.waitingFor(
+ forHttp(
+ "/admin/v2/persistent/pulsar/system/transaction_coordinator_assign/partitions")
+ .forPort(BROKER_HTTP_PORT)
+ .forStatusCode(200)
+ .withStartupTimeout(Duration.ofMinutes(5)));
+
+ return container;
+ }
+
+ @BeforeEach
+ public void setup() throws Exception {
+ admin = PulsarAdmin.builder().serviceHttpUrl(PULSAR_CONTAINER.getHttpServiceUrl()).build();
+ client =
+ PulsarClient.builder()
+ .serviceUrl(PULSAR_CONTAINER.getPulsarBrokerUrl())
+ .enableTransaction(true)
+ .build();
+ }
+
+ @AfterEach
+ public void after() throws Exception {
+ super.after();
+ // Delete topics for avoid reusing topics of pulsar cluster
+ deleteTopics();
+ admin.close();
+ client.close();
+ }
+
+ protected Map<String, String> getBasicPulsarConfig() {
+ Map<String, String> config = new HashMap<>();
+ config.put(PULSAR_SERVICE_URL.key(), PULSAR_CONTAINER.getPulsarBrokerUrl());
+ config.put(PULSAR_ADMIN_URL.key(), PULSAR_CONTAINER.getHttpServiceUrl());
+ config.put(PULSAR_SUBSCRIPTION_NAME.key(), "paimon-tests");
+ return config;
+ }
+
+ protected void createTopic(String topic) {
+ createTopic(topic, 0);
+ }
+
+ /**
+ * Create a pulsar topic with given partition number if the topic doesn't exist.
+ *
+ * @param topic The name of the topic.
+ * @param numberOfPartitions The number of partitions. We would create a non-partitioned topic
+ * if this number is zero.
+ */
+ protected void createTopic(String topic, int numberOfPartitions) {
+ checkArgument(numberOfPartitions >= 0);
+ String topicName = topicName(topic);
+ if (numberOfPartitions == 0) {
+ createNonPartitionedTopic(topicName);
+ } else {
+ createPartitionedTopic(topicName, numberOfPartitions);
+ }
+ }
+
+ private void createNonPartitionedTopic(String topic) {
+ try {
+ admin.topics().createNonPartitionedTopic(topic);
+ } catch (PulsarAdminException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void createPartitionedTopic(String topic, int numberOfPartitions) {
+ try {
+ admin.topics().createPartitionedTopic(topic, numberOfPartitions);
+ } catch (PulsarAdminException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void deleteTopics() throws Exception {
+ List<String> topics = admin.topics().getList("public/default");
+ for (String topic : topics) {
+ String topicName = topicName(topic);
+ PartitionedTopicMetadata metadata =
+ admin.topics().getPartitionedTopicMetadata(topicName);
+
+ if (metadata.partitions == NON_PARTITIONED) {
+ admin.topics().delete(topicName, true);
+ } else {
+ admin.topics().deletePartitionedTopic(topicName, true);
+ }
+ }
+ }
+
+ protected List<String> getMessages(String resource) throws IOException {
+ URL url = PulsarActionITCaseBase.class.getClassLoader().getResource(resource);
+ assertThat(url).isNotNull();
+ java.nio.file.Path path = new File(url.getFile()).toPath();
+ List<String> lines = Files.readAllLines(path);
+ List<String> messages = new ArrayList<>();
+ for (String line : lines) {
+ try {
+ objectMapper.readTree(line);
+ if (!StringUtils.isEmpty(line)) {
+ messages.add(line);
+ }
+ } catch (Exception e) {
+ // ignore
+ }
+ }
+
+ return messages;
+ }
+
+ /**
+ * Send a list messages to Pulsar, return the message id set after the ack from Pulsar.
+ *
+ * @param topic The name of the topic.
+ * @param messages The records need to be sent.
+ */
+ protected void sendMessages(String topic, List<String> messages) {
+ try (Producer<String> producer = createProducer(topic)) {
+ for (String message : messages) {
+ producer.newMessage().value(message).send();
+ }
+ producer.flush();
+ } catch (PulsarClientException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Producer<String> createProducer(String topic) {
+ try {
+ return client.newProducer(Schema.STRING)
+ .topic(topic)
+ .enableBatching(false)
+ .enableMultiSchema(true)
+ .accessMode(Shared)
+ .create();
+ } catch (PulsarClientException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected PulsarSyncTableActionBuilder syncTableActionBuilder(
+ Map<String, String> pulsarConfig) {
+ return new PulsarSyncTableActionBuilder(pulsarConfig);
+ }
+
+ /** Builder to build {@link PulsarSyncTableAction} from action arguments. */
+ protected class PulsarSyncTableActionBuilder
+ extends SyncTableActionBuilder<PulsarSyncTableAction> {
+
+ public PulsarSyncTableActionBuilder(Map<String, String> pulsarConfig) {
+ super(pulsarConfig);
+ }
+
+ public PulsarSyncTableAction build() {
+ List<String> args =
+ new ArrayList<>(
+ Arrays.asList(
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName));
+
+ args.addAll(mapToArgs("--pulsar-conf", sourceConfig));
+ args.addAll(mapToArgs("--catalog-conf", catalogConfig));
+ args.addAll(mapToArgs("--table-conf", tableConfig));
+
+ args.addAll(listToArgs("--partition-keys", partitionKeys));
+ args.addAll(listToArgs("--primary-keys", primaryKeys));
+ args.addAll(listToArgs("--type-mapping", typeMappingModes));
+
+ args.addAll(listToMultiArgs("--computed-column", computedColumnArgs));
+
+ MultipleParameterTool params =
+ MultipleParameterTool.fromArgs(args.toArray(args.toArray(new String[0])));
+ return (PulsarSyncTableAction)
+ new PulsarSyncTableActionFactory()
+ .create(params)
+ .orElseThrow(RuntimeException::new);
+ }
+ }
+
+ /** Pulsar container extension for junit5. */
+ private static class PulsarContainerExtension extends PulsarContainer
+ implements BeforeAllCallback, AfterAllCallback {
+ private PulsarContainerExtension(String pulsarVersion) {
+ super(pulsarVersion);
+ }
+
+ @Override
+ public void beforeAll(ExtensionContext extensionContext) {
+ this.doStart();
+ }
+
+ @Override
+ public void afterAll(ExtensionContext extensionContext) {
+ this.close();
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSchemaITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSchemaITCase.java
new file mode 100644
index 000000000..789f6b414
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSchemaITCase.java
@@ -0,0 +1,69 @@
+/*
+ * 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.flink.action.cdc.MessageQueueSchemaUtils;
+import org.apache.paimon.flink.action.cdc.TypeMapping;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+
+import org.apache.flink.configuration.Configuration;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.paimon.flink.action.cdc.pulsar.PulsarActionUtils.TOPIC;
+import static org.apache.paimon.flink.action.cdc.pulsar.PulsarActionUtils.VALUE_FORMAT;
+import static org.apache.paimon.flink.action.cdc.pulsar.PulsarActionUtils.createPulsarConsumer;
+import static org.apache.paimon.flink.action.cdc.pulsar.PulsarActionUtils.getDataFormat;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for building schema from Pulsar. */
+public class PulsarSchemaITCase extends PulsarActionITCaseBase {
+
+ @Test
+ @Timeout(60)
+ public void testPulsarSchema() throws Exception {
+ String topic = "test_pulsar_schema";
+ createTopic(topic);
+
+ // ---------- Write the Canal json into pulsar -------------------
+ List<String> messages = getMessages("kafka/canal/table/schemaevolution/canal-data-1.txt");
+ sendMessages(topic, messages);
+
+ Configuration pulsarConfig = Configuration.fromMap(getBasicPulsarConfig());
+ pulsarConfig.set(TOPIC, topic);
+ pulsarConfig.set(VALUE_FORMAT, "canal-json");
+
+ Schema pulsarSchema =
+ MessageQueueSchemaUtils.getSchema(
+ createPulsarConsumer(pulsarConfig, topic),
+ topic,
+ getDataFormat(pulsarConfig),
+ TypeMapping.defaultMapping());
+ List<DataField> fields = new ArrayList<>();
+ fields.add(new DataField(0, "pt", DataTypes.INT()));
+ fields.add(new DataField(1, "_id", DataTypes.INT().notNull()));
+ fields.add(new DataField(2, "v1", DataTypes.VARCHAR(10)));
+ assertThat(pulsarSchema.fields()).isEqualTo(fields);
+ }
+}
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/PulsarSyncTableActionITCase.java
new file mode 100644
index 000000000..a39507b5f
--- /dev/null
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncTableActionITCase.java
@@ -0,0 +1,188 @@
+/*
+ * 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.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 PulsarSyncTableAction}. */
+public class PulsarSyncTableActionITCase 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 Canal json into Pulsar -------------------
+ sendMessages(
+ topic,
+ getMessages(String.format("kafka/canal/table/%s/canal-data-1.txt", sourceDir)));
+
+ Map<String, String> pulsarConfig = getBasicPulsarConfig();
+ pulsarConfig.put(PULSAR_PARTITION_DISCOVERY_INTERVAL_MS.key(), "-1");
+ pulsarConfig.put(TOPIC.key(), topic);
+ pulsarConfig.put(VALUE_FORMAT.key(), "canal-json");
+
+ PulsarSyncTableAction action =
+ syncTableActionBuilder(pulsarConfig)
+ .withPartitionKeys("pt")
+ .withPrimaryKeys("pt", "_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.INT().notNull(),
+ DataTypes.INT().notNull(),
+ DataTypes.VARCHAR(10)
+ },
+ new String[] {"pt", "_id", "v1"});
+ List<String> primaryKeys = Arrays.asList("pt", "_id");
+ List<String> expected = Arrays.asList("+I[1, 1, one]", "+I[1, 2, two]", "+I[2, 4, four]");
+ waitForResult(expected, table, rowType, primaryKeys);
+
+ sendMessages(
+ topic,
+ getMessages(String.format("kafka/canal/table/%s/canal-data-2.txt", sourceDir)));
+
+ rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.INT().notNull(),
+ DataTypes.VARCHAR(10),
+ DataTypes.INT()
+ },
+ new String[] {"pt", "_id", "v1", "v2"});
+ expected =
+ Arrays.asList(
+ "+I[1, 1, one, NULL]",
+ "+I[1, 2, second, NULL]",
+ "+I[2, 3, three, 30]",
+ "+I[2, 4, four, NULL]",
+ "+I[1, 5, five, 50]",
+ "+I[1, 6, six, 60]");
+ waitForResult(expected, table, rowType, primaryKeys);
+
+ sendMessages(
+ topic,
+ getMessages(String.format("kafka/canal/table/%s/canal-data-3.txt", sourceDir)));
+
+ rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.INT().notNull(),
+ DataTypes.VARCHAR(10),
+ DataTypes.BIGINT()
+ },
+ new String[] {"pt", "_id", "v1", "v2"});
+ expected =
+ Arrays.asList(
+ "+I[1, 1, one, NULL]",
+ "+I[1, 2, second, NULL]",
+ "+I[2, 3, three, 30000000000]",
+ "+I[2, 4, four, NULL]",
+ "+I[1, 6, six, 60]",
+ "+I[2, 7, seven, 70000000000]",
+ "+I[2, 8, eight, 80000000000]");
+ waitForResult(expected, table, rowType, primaryKeys);
+
+ sendMessages(
+ topic,
+ getMessages(String.format("kafka/canal/table/%s/canal-data-4.txt", sourceDir)));
+
+ rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.INT().notNull(),
+ DataTypes.VARCHAR(20),
+ DataTypes.BIGINT(),
+ DataTypes.DECIMAL(8, 3),
+ DataTypes.VARBINARY(10),
+ DataTypes.FLOAT()
+ },
+ new String[] {"pt", "_id", "v1", "v2", "v3", "v4", "v5"});
+ expected =
+ Arrays.asList(
+ "+I[1, 1, one, NULL, NULL, NULL, NULL]",
+ "+I[1, 2, second, NULL, NULL, NULL, NULL]",
+ "+I[2, 3, three, 30000000000, NULL, NULL, NULL]",
+ "+I[2, 4, four, NULL, NULL, NULL, NULL]",
+ "+I[1, 6, six, 60, NULL, NULL, NULL]",
+ "+I[2, 7, seven, 70000000000, NULL, NULL, NULL]",
+ "+I[2, 8, very long string, 80000000000, NULL, NULL, NULL]",
+ "+I[1, 9, nine, 90000000000, 99999.999, [110, 105, 110, 101, 46, 98, 105, 110], 9.9]");
+ waitForResult(expected, table, rowType, primaryKeys);
+
+ sendMessages(
+ topic,
+ getMessages(String.format("kafka/canal/table/%s/canal-data-5.txt", sourceDir)));
+
+ rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(),
+ DataTypes.INT().notNull(),
+ DataTypes.VARCHAR(20),
+ DataTypes.BIGINT(),
+ DataTypes.DECIMAL(8, 3),
+ DataTypes.VARBINARY(20),
+ DataTypes.DOUBLE()
+ },
+ new String[] {"pt", "_id", "v1", "v2", "v3", "v4", "v5"});
+ expected =
+ Arrays.asList(
+ "+I[1, 1, one, NULL, NULL, NULL, NULL]",
+ "+I[1, 2, second, NULL, NULL, NULL, NULL]",
+ "+I[2, 3, three, 30000000000, NULL, NULL, NULL]",
+ "+I[2, 4, four, NULL, NULL, [102, 111, 117, 114, 46, 98, 105, 110, 46, 108, 111, 110, 103], 4.00000000004]",
+ "+I[1, 6, six, 60, NULL, NULL, NULL]",
+ "+I[2, 7, seven, 70000000000, NULL, NULL, NULL]",
+ "+I[2, 8, very long string, 80000000000, NULL, NULL, NULL]",
+ "+I[1, 9, nine, 90000000000, 99999.999, [110, 105, 110, 101, 46, 98, 105, 110, 46, 108, 111, 110, 103], 9.00000000009]");
+ waitForResult(expected, table, rowType, primaryKeys);
+ }
+}