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 f68743c1b [CDC] Refactor cdc to reduce redundant code and set default
checkpoint interval (#2461)
f68743c1b is described below
commit f68743c1bf429db4b28fccd3e24e23bf0a5266f6
Author: yuzelin <[email protected]>
AuthorDate: Mon Dec 11 17:05:11 2023 +0800
[CDC] Refactor cdc to reduce redundant code and set default checkpoint
interval (#2461)
---
docs/content/cdc-ingestion/overview.md | 9 +-
.../flink/action/cdc/CdcActionCommonUtils.java | 13 ++
.../flink/action/cdc/MessageQueueSchemaUtils.java | 12 +-
.../cdc/MessageQueueSyncDatabaseActionBase.java | 81 --------
.../cdc/MessageQueueSyncTableActionBase.java | 28 +--
.../flink/action/cdc/SyncDatabaseActionBase.java | 113 +++--------
.../paimon/flink/action/cdc/SyncJobHandler.java | 222 +++++++++++++++++++++
.../flink/action/cdc/SyncTableActionBase.java | 155 ++++----------
.../action/cdc/SynchronizationActionBase.java | 154 ++++++++++++++
.../flink/action/cdc/kafka/KafkaActionUtils.java | 42 ++--
.../action/cdc/kafka/KafkaSyncDatabaseAction.java | 34 +---
.../cdc/kafka/KafkaSyncDatabaseActionFactory.java | 6 +-
.../action/cdc/kafka/KafkaSyncTableAction.java | 44 +---
.../action/cdc/mongodb/MongoDBActionUtils.java | 16 --
.../action/cdc/mongodb/MongoDBRecordParser.java | 5 -
.../cdc/mongodb/MongoDBSyncDatabaseAction.java | 44 +---
.../mongodb/MongoDBSyncDatabaseActionFactory.java | 6 +-
.../action/cdc/mongodb/MongoDBSyncTableAction.java | 48 ++---
.../flink/action/cdc/mysql/MySqlActionUtils.java | 25 ---
.../flink/action/cdc/mysql/MySqlRecordParser.java | 8 -
.../action/cdc/mysql/MySqlSyncDatabaseAction.java | 54 ++---
.../action/cdc/mysql/MySqlSyncTableAction.java | 45 ++---
.../flink/action/cdc/pulsar/PulsarActionUtils.java | 110 +++++-----
.../cdc/pulsar/PulsarSyncDatabaseAction.java | 34 +---
.../pulsar/PulsarSyncDatabaseActionFactory.java | 6 +-
.../action/cdc/pulsar/PulsarSyncTableAction.java | 48 +----
.../kafka/KafkaCanalSyncDatabaseActionITCase.java | 2 +-
.../KafkaMaxwellSyncDatabaseActionITCase.java | 2 +-
.../kafka/KafkaOggSyncDatabaseActionITCase.java | 2 +-
.../flink/action/cdc/kafka/KafkaSchemaITCase.java | 3 +-
.../cdc/mysql/MySqlSyncDatabaseActionITCase.java | 4 +-
.../cdc/mysql/MySqlSyncTableActionITCase.java | 38 ++++
.../action/cdc/pulsar/PulsarSchemaITCase.java | 3 +-
.../src/test/resources/mysql/sync_table_setup.sql | 13 ++
34 files changed, 674 insertions(+), 755 deletions(-)
diff --git a/docs/content/cdc-ingestion/overview.md
b/docs/content/cdc-ingestion/overview.md
index 031390104..d0de032f0 100644
--- a/docs/content/cdc-ingestion/overview.md
+++ b/docs/content/cdc-ingestion/overview.md
@@ -99,6 +99,13 @@ store up to 19 digits integer value. So you should ensure
the overflow won't occ
9. MySQL BINARY will be mapped to Paimon VARBINARY. This is because the binary
value is passed as bytes in binlog, so it
should be mapped to byte type (BYTES or VARBINARY). We choose VARBINARY
because it can retain the length information.
-## Setting Custom Job Name
+## Custom Job Settings
+
+### Checkpointing
+
+Use `-Dexecution.checkpointing.interval=<interval>` to enable checkpointing
and set interval. For 0.7 and later versions,
+if you haven't enabled checkpointing, Paimon will enable checkpointing by
default and set checkpoint interval to 180 seconds.
+
+### Job Name
Use `-Dpipeline.name=<job-name>` to set custom synchronization job name.
\ No newline at end of file
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java
index 97d4c9a10..d54b1ea36 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/CdcActionCommonUtils.java
@@ -26,6 +26,8 @@ import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -283,4 +285,15 @@ public class CdcActionCommonUtils {
excludingPattern = "?!" + excludingPattern;
return String.format("(%s)(%s)", excludingPattern, includingPattern);
}
+
+ public static void checkRequiredOptions(
+ Configuration config, String confName, ConfigOption<?>...
configOptions) {
+ for (ConfigOption<?> configOption : configOptions) {
+ checkArgument(
+ config.contains(configOption),
+ "%s [%s] must be specified.",
+ confName,
+ configOption.key());
+ }
+ }
}
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
index dc06a5a8c..326c73cca 100644
---
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
@@ -38,14 +38,13 @@ public class MessageQueueSchemaUtils {
* 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)
+ ConsumerWrapper consumer, DataFormat dataFormat, TypeMapping
typeMapping)
throws SchemaRetrievalException {
int retry = 0;
int retryInterval = 1000;
@@ -55,7 +54,7 @@ public class MessageQueueSchemaUtils {
while (true) {
Optional<Schema> schema =
- consumer.getRecords(topic, POLL_TIMEOUT_MILLIS).stream()
+ consumer.getRecords(POLL_TIMEOUT_MILLIS).stream()
.map(recordParser::buildSchema)
.filter(Objects::nonNull)
.findFirst();
@@ -66,7 +65,8 @@ public class MessageQueueSchemaUtils {
if (retry >= MAX_RETRY) {
throw new SchemaRetrievalException(
- String.format("Could not get metadata from server,
topic: %s", topic));
+ String.format(
+ "Could not get metadata from server, topic:
%s", consumer.topic()));
}
sleepSafely(retryInterval);
@@ -86,6 +86,8 @@ public class MessageQueueSchemaUtils {
/** Wrap the consumer for different message queues. */
public interface ConsumerWrapper extends AutoCloseable {
- List<String> getRecords(String topic, int pollTimeOutMills);
+ List<String> getRecords(int pollTimeOutMills);
+
+ String topic();
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncDatabaseActionBase.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncDatabaseActionBase.java
deleted file mode 100644
index cc869a85b..000000000
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncDatabaseActionBase.java
+++ /dev/null
@@ -1,81 +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;
-
-import org.apache.paimon.flink.action.Action;
-import org.apache.paimon.flink.action.MultiTablesSinkMode;
-import org.apache.paimon.flink.action.cdc.format.DataFormat;
-import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
-
-import org.apache.flink.api.common.functions.FlatMapFunction;
-
-import java.util.Collections;
-import java.util.Map;
-
-/**
- * An {@link Action} which synchronize the Multiple message queue topics into
one Paimon database.
- *
- * <p>For each message queue topic's table to be synchronized, if the
corresponding Paimon table
- * does not exist, this action will automatically create the table, and its
schema will be derived
- * from all specified message queue topic's tables. If the Paimon table
already exists and its
- * schema is different from that parsed from message queue record, this action
will try to preform
- * schema evolution.
- *
- * <p>This action supports a limited number of schema changes. Currently, the
framework can not drop
- * columns, so the behaviors of `DROP` will be ignored, `RENAME` will add a
new column. 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.
- * </ul>
- *
- * <p>To automatically synchronize new table, This action creates a single
sink for all Paimon
- * tables to be written. See {@link MultiTablesSinkMode#COMBINED}.
- */
-public abstract class MessageQueueSyncDatabaseActionBase extends
SyncDatabaseActionBase {
-
- public MessageQueueSyncDatabaseActionBase(
- String warehouse,
- String database,
- Map<String, String> catalogConfig,
- Map<String, String> mqConfig) {
- super(warehouse, database, catalogConfig, mqConfig);
- }
-
- protected abstract DataFormat getDataFormat();
-
- @Override
- protected FlatMapFunction<String, RichCdcMultiplexRecord> recordParse() {
- DataFormat format = getDataFormat();
- boolean caseSensitive = catalog.caseSensitive();
- return format.createParser(caseSensitive, typeMapping,
Collections.emptyList());
- }
-}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
index a0a976a8b..c3c6d49ff 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/MessageQueueSyncTableActionBase.java
@@ -19,12 +19,8 @@
package org.apache.paimon.flink.action.cdc;
import org.apache.paimon.flink.action.Action;
-import org.apache.paimon.flink.action.cdc.format.DataFormat;
-import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
import org.apache.paimon.schema.Schema;
-import org.apache.flink.api.common.functions.FlatMapFunction;
-
import java.util.Map;
/**
@@ -60,15 +56,16 @@ public abstract class MessageQueueSyncTableActionBase
extends SyncTableActionBas
String database,
String table,
Map<String, String> catalogConfig,
- Map<String, String> mqConfig) {
- super(warehouse, database, table, catalogConfig, mqConfig);
+ Map<String, String> mqConfig,
+ SyncJobHandler.SourceType sourceType) {
+ super(warehouse, database, table, catalogConfig, mqConfig, sourceType);
}
@Override
protected Schema retrieveSchema() throws Exception {
- String topic = topic();
- try (MessageQueueSchemaUtils.ConsumerWrapper consumer =
consumer(topic)) {
- return MessageQueueSchemaUtils.getSchema(consumer, topic,
getDataFormat(), typeMapping);
+ try (MessageQueueSchemaUtils.ConsumerWrapper consumer =
syncJobHandler.provideConsumer()) {
+ return MessageQueueSchemaUtils.getSchema(
+ consumer, syncJobHandler.provideDataFormat(), typeMapping);
}
}
@@ -83,17 +80,4 @@ public abstract class MessageQueueSyncTableActionBase
extends SyncTableActionBas
metadataConverters,
false);
}
-
- @Override
- protected FlatMapFunction<String, RichCdcMultiplexRecord> recordParse() {
- boolean caseSensitive = catalog.caseSensitive();
- DataFormat format = getDataFormat();
- return format.createParser(caseSensitive, typeMapping,
computedColumns);
- }
-
- protected abstract String topic();
-
- protected abstract MessageQueueSchemaUtils.ConsumerWrapper consumer(String
topic);
-
- protected abstract DataFormat getDataFormat();
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
index d31c6cc12..74c2f6fc7 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncDatabaseActionBase.java
@@ -18,10 +18,8 @@
package org.apache.paimon.flink.action.cdc;
-import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.catalog.AbstractCatalog;
import org.apache.paimon.flink.action.Action;
-import org.apache.paimon.flink.action.ActionBase;
import org.apache.paimon.flink.action.MultiTablesSinkMode;
import org.apache.paimon.flink.sink.cdc.EventParser;
import org.apache.paimon.flink.sink.cdc.FlinkCdcSyncDatabaseSinkBuilder;
@@ -31,51 +29,41 @@ import
org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecordEventParser;
import org.apache.paimon.table.FileStoreTable;
import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
import javax.annotation.Nullable;
import java.util.ArrayList;
-import java.util.HashMap;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.regex.Pattern;
import static org.apache.paimon.flink.action.MultiTablesSinkMode.COMBINED;
/** Base {@link Action} for synchronizing into one Paimon database. */
-public abstract class SyncDatabaseActionBase extends ActionBase {
+public abstract class SyncDatabaseActionBase extends SynchronizationActionBase
{
- protected final String database;
- protected final Configuration cdcSourceConfig;
- protected Map<String, String> tableConfig = new HashMap<>();
protected boolean mergeShards = true;
protected MultiTablesSinkMode mode = COMBINED;
protected String tablePrefix = "";
protected String tableSuffix = "";
protected String includingTables = ".*";
@Nullable protected String excludingTables;
- protected TypeMapping typeMapping = TypeMapping.defaultMapping();
-
- protected CdcMetadataConverter[] metadataConverters = new
CdcMetadataConverter[] {};
protected List<FileStoreTable> tables = new ArrayList<>();
public SyncDatabaseActionBase(
String warehouse,
String database,
Map<String, String> catalogConfig,
- Map<String, String> cdcSourceConfig) {
- super(warehouse, catalogConfig);
- this.database = database;
- this.cdcSourceConfig = Configuration.fromMap(cdcSourceConfig);
- }
-
- public SyncDatabaseActionBase withTableConfig(Map<String, String>
tableConfig) {
- this.tableConfig = tableConfig;
- return this;
+ Map<String, String> cdcSourceConfig,
+ SyncJobHandler.SourceType sourceType) {
+ super(
+ warehouse,
+ database,
+ catalogConfig,
+ cdcSourceConfig,
+ new SyncJobHandler(sourceType, cdcSourceConfig, database));
}
public SyncDatabaseActionBase mergeShards(boolean mergeShards) {
@@ -114,42 +102,21 @@ public abstract class SyncDatabaseActionBase extends
ActionBase {
return this;
}
- public SyncDatabaseActionBase withTypeMapping(TypeMapping typeMapping) {
- this.typeMapping = typeMapping;
- return this;
- }
-
- public SyncDatabaseActionBase withMetadataColumns(List<String>
metadataColumns) {
- this.metadataConverters =
- metadataColumns.stream()
- .map(this::metadataConverter)
- .filter(Optional::isPresent)
- .map(Optional::get)
- .toArray(CdcMetadataConverter[]::new);
- return this;
+ @Override
+ protected void validateCaseSensitivity() {
+ AbstractCatalog.validateCaseInsensitive(caseSensitive, "Database",
database);
+ AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table prefix",
tablePrefix);
+ AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table suffix",
tableSuffix);
}
- protected Optional<CdcMetadataConverter<?>> metadataConverter(String
column) {
- return Optional.empty();
+ @Override
+ protected FlatMapFunction<String, RichCdcMultiplexRecord> recordParse() {
+ return syncJobHandler.provideRecordParser(
+ caseSensitive, Collections.emptyList(), typeMapping,
metadataConverters);
}
- protected void checkCdcSourceArgument() {}
-
- protected abstract DataStreamSource<String> buildSource() throws Exception;
-
- protected abstract String sourceName();
-
- protected abstract FlatMapFunction<String, RichCdcMultiplexRecord>
recordParse();
-
@Override
- public void build() throws Exception {
- checkCdcSourceArgument();
- boolean caseSensitive = catalog.caseSensitive();
-
- validateCaseInsensitive(caseSensitive);
-
- catalog.createDatabase(database, true);
-
+ protected EventParser.Factory<RichCdcMultiplexRecord>
buildEventParserFactory() {
NewTableSchemaBuilder schemaBuilder = new
NewTableSchemaBuilder(tableConfig, caseSensitive);
Pattern includingPattern = Pattern.compile(includingTables);
Pattern excludingPattern =
@@ -157,16 +124,15 @@ public abstract class SyncDatabaseActionBase extends
ActionBase {
TableNameConverter tableNameConverter =
new TableNameConverter(caseSensitive, mergeShards,
tablePrefix, tableSuffix);
- DataStream<RichCdcMultiplexRecord> input =
- buildSource().flatMap(recordParse()).name("Parse");
- EventParser.Factory<RichCdcMultiplexRecord> parserFactory =
- () ->
- new RichCdcMultiplexRecordEventParser(
- schemaBuilder,
- includingPattern,
- excludingPattern,
- tableNameConverter);
+ return () ->
+ new RichCdcMultiplexRecordEventParser(
+ schemaBuilder, includingPattern, excludingPattern,
tableNameConverter);
+ }
+ @Override
+ protected void buildSink(
+ DataStream<RichCdcMultiplexRecord> input,
+ EventParser.Factory<RichCdcMultiplexRecord> parserFactory) {
new FlinkCdcSyncDatabaseSinkBuilder<RichCdcMultiplexRecord>()
.withInput(input)
.withParserFactory(parserFactory)
@@ -177,27 +143,4 @@ public abstract class SyncDatabaseActionBase extends
ActionBase {
.withTableOptions(tableConfig)
.build();
}
-
- protected void validateCaseInsensitive(boolean caseSensitive) {
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Database",
database);
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table prefix",
tablePrefix);
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table suffix",
tableSuffix);
- }
-
- @VisibleForTesting
- public Map<String, String> tableConfig() {
- return tableConfig;
- }
-
- // ------------------------------------------------------------------------
- // Flink run methods
- // ------------------------------------------------------------------------
-
- protected abstract String jobName();
-
- @Override
- public void run() throws Exception {
- build();
- execute(jobName());
- }
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncJobHandler.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncJobHandler.java
new file mode 100644
index 000000000..e469723b1
--- /dev/null
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncJobHandler.java
@@ -0,0 +1,222 @@
+/*
+ * 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.kafka.KafkaActionUtils;
+import org.apache.paimon.flink.action.cdc.mongodb.MongoDBRecordParser;
+import org.apache.paimon.flink.action.cdc.mysql.MySqlActionUtils;
+import org.apache.paimon.flink.action.cdc.mysql.MySqlRecordParser;
+import org.apache.paimon.flink.action.cdc.pulsar.PulsarActionUtils;
+import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
+
+import com.ververica.cdc.connectors.mongodb.source.config.MongoDBSourceOptions;
+import com.ververica.cdc.connectors.mysql.source.config.MySqlSourceOptions;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.pulsar.common.config.PulsarOptions;
+import org.apache.flink.connector.pulsar.source.PulsarSourceOptions;
+import org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions;
+
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.KAFKA_CONF;
+import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.MONGODB_CONF;
+import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.MYSQL_CONF;
+import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.PULSAR_CONF;
+import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.checkRequiredOptions;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Provide different function according to CDC source type. */
+public class SyncJobHandler {
+
+ private final SourceType sourceType;
+ private final Configuration cdcSourceConfig;
+ private final boolean isTableSync;
+ private final String sinkLocation;
+
+ public SyncJobHandler(
+ SourceType sourceType, Map<String, String> cdcSourceConfig, String
database) {
+ this.sourceType = sourceType;
+ this.cdcSourceConfig = Configuration.fromMap(cdcSourceConfig);
+ this.isTableSync = false;
+ this.sinkLocation = database;
+ }
+
+ public SyncJobHandler(
+ SourceType sourceType,
+ Map<String, String> cdcSourceConfig,
+ String database,
+ String table) {
+ this.sourceType = sourceType;
+ this.cdcSourceConfig = Configuration.fromMap(cdcSourceConfig);
+ this.isTableSync = true;
+ this.sinkLocation = database + "." + table;
+ }
+
+ public String provideSourceName() {
+ return sourceType.sourceName;
+ }
+
+ public String provideDefaultJobName() {
+ return String.format(
+ sourceType.defaultJobNameFormat, isTableSync ? "Table" :
"Database", sinkLocation);
+ }
+
+ public void registerJdbcDriver() {
+ if (sourceType == SourceType.MYSQL) {
+ MySqlActionUtils.registerJdbcDriver();
+ }
+ }
+
+ public void checkRequiredOption() {
+ switch (sourceType) {
+ case MYSQL:
+ checkRequiredOptions(
+ cdcSourceConfig,
+ MYSQL_CONF,
+ MySqlSourceOptions.HOSTNAME,
+ MySqlSourceOptions.PORT,
+ MySqlSourceOptions.USERNAME,
+ MySqlSourceOptions.PASSWORD,
+ MySqlSourceOptions.DATABASE_NAME);
+ if (isTableSync) {
+ checkRequiredOptions(
+ cdcSourceConfig, MYSQL_CONF,
MySqlSourceOptions.TABLE_NAME);
+ } else {
+ checkArgument(
+
!cdcSourceConfig.contains(MySqlSourceOptions.TABLE_NAME),
+ MySqlSourceOptions.TABLE_NAME.key()
+ + " cannot be set for mysql_sync_database.
"
+ + "If you want to sync several MySQL
tables into one Paimon table, "
+ + "use mysql_sync_table instead.");
+ }
+ break;
+ case KAFKA:
+ checkRequiredOptions(
+ cdcSourceConfig,
+ KAFKA_CONF,
+ KafkaConnectorOptions.VALUE_FORMAT,
+ KafkaConnectorOptions.TOPIC,
+ KafkaConnectorOptions.PROPS_BOOTSTRAP_SERVERS);
+ break;
+ case PULSAR:
+ checkRequiredOptions(
+ cdcSourceConfig,
+ PULSAR_CONF,
+ PulsarActionUtils.TOPIC,
+ PulsarActionUtils.VALUE_FORMAT,
+ PulsarOptions.PULSAR_SERVICE_URL,
+ PulsarOptions.PULSAR_ADMIN_URL,
+ PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME);
+ break;
+ case MONGODB:
+ checkRequiredOptions(
+ cdcSourceConfig,
+ MONGODB_CONF,
+ MongoDBSourceOptions.HOSTS,
+ MongoDBSourceOptions.DATABASE);
+ if (isTableSync) {
+ checkRequiredOptions(
+ cdcSourceConfig, MONGODB_CONF,
MongoDBSourceOptions.COLLECTION);
+ }
+ break;
+ default:
+ throw new UnsupportedOperationException("Unknown source type "
+ sourceType);
+ }
+ }
+
+ public Source<String, ?, ?> provideSource() {
+ switch (sourceType) {
+ case KAFKA:
+ return KafkaActionUtils.buildKafkaSource(cdcSourceConfig);
+ case PULSAR:
+ return PulsarActionUtils.buildPulsarSource(cdcSourceConfig);
+ default:
+ throw new UnsupportedOperationException(
+ "Cannot get source from source type" + sourceType);
+ }
+ }
+
+ public FlatMapFunction<String, RichCdcMultiplexRecord> provideRecordParser(
+ boolean caseSensitive,
+ List<ComputedColumn> computedColumns,
+ TypeMapping typeMapping,
+ CdcMetadataConverter[] metadataConverters) {
+ switch (sourceType) {
+ case MYSQL:
+ return new MySqlRecordParser(
+ cdcSourceConfig,
+ caseSensitive,
+ computedColumns,
+ typeMapping,
+ metadataConverters);
+ case KAFKA:
+ case PULSAR:
+ DataFormat dataFormat = provideDataFormat();
+ return dataFormat.createParser(caseSensitive, typeMapping,
computedColumns);
+ case MONGODB:
+ return new MongoDBRecordParser(caseSensitive, computedColumns,
cdcSourceConfig);
+ default:
+ throw new UnsupportedOperationException("Unknown source type "
+ sourceType);
+ }
+ }
+
+ public DataFormat provideDataFormat() {
+ switch (sourceType) {
+ case KAFKA:
+ return KafkaActionUtils.getDataFormat(cdcSourceConfig);
+ case PULSAR:
+ return PulsarActionUtils.getDataFormat(cdcSourceConfig);
+ default:
+ throw new UnsupportedOperationException(
+ "Cannot get DataFormat from source type" + sourceType);
+ }
+ }
+
+ public MessageQueueSchemaUtils.ConsumerWrapper provideConsumer() {
+ switch (sourceType) {
+ case KAFKA:
+ return
KafkaActionUtils.getKafkaEarliestConsumer(cdcSourceConfig);
+ case PULSAR:
+ return PulsarActionUtils.createPulsarConsumer(cdcSourceConfig);
+ default:
+ throw new UnsupportedOperationException(
+ "Cannot get consumer from source type" + sourceType);
+ }
+ }
+
+ /** CDC source type. */
+ public enum SourceType {
+ MYSQL("MySQL Source", "MySQL-Paimon %s Sync: %s"),
+ KAFKA("Kafka Source", "Kafka-Paimon %s Sync: %s"),
+ MONGODB("MongoDB Source", "MongoDB-Paimon %s Sync: %s"),
+ PULSAR("Pulsar Source", "Pulsar-Paimon %s Sync: %s");
+
+ private final String sourceName;
+ private final String defaultJobNameFormat;
+
+ SourceType(String sourceName, String defaultJobNameFormat) {
+ this.sourceName = sourceName;
+ this.defaultJobNameFormat = defaultJobNameFormat;
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionBase.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionBase.java
index 82510f15a..317304b28 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionBase.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SyncTableActionBase.java
@@ -23,7 +23,6 @@ import org.apache.paimon.catalog.AbstractCatalog;
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.sink.cdc.CdcSinkBuilder;
import org.apache.paimon.flink.sink.cdc.EventParser;
import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
@@ -31,60 +30,47 @@ import
org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecordEventParser;
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.common.functions.FlatMapFunction;
-import org.apache.flink.api.connector.source.Source;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
-import java.util.stream.Collectors;
import static
org.apache.paimon.flink.action.cdc.CdcActionCommonUtils.assertSchemaCompatible;
import static
org.apache.paimon.flink.action.cdc.ComputedColumnUtils.buildComputedColumns;
-import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkState;
/** Base {@link Action} for synchronizing into one Paimon table. */
-public abstract class SyncTableActionBase extends ActionBase {
+public abstract class SyncTableActionBase extends SynchronizationActionBase {
private static final Logger LOG =
LoggerFactory.getLogger(SyncTableActionBase.class);
- protected final String database;
protected final String table;
- protected final Configuration cdcSourceConfig;
- protected FileStoreTable fileStoreTable;
+ protected FileStoreTable fileStoreTable;
protected List<String> partitionKeys = new ArrayList<>();
protected List<String> primaryKeys = new ArrayList<>();
-
- protected Map<String, String> tableConfig = new HashMap<>();
protected List<String> computedColumnArgs = new ArrayList<>();
- protected TypeMapping typeMapping = TypeMapping.defaultMapping();
-
protected List<ComputedColumn> computedColumns = new ArrayList<>();
- protected CdcMetadataConverter[] metadataConverters = new
CdcMetadataConverter[] {};
public SyncTableActionBase(
String warehouse,
String database,
String table,
Map<String, String> catalogConfig,
- Map<String, String> cdcSourceConfig) {
- super(warehouse, catalogConfig);
- this.database = database;
+ Map<String, String> cdcSourceConfig,
+ SyncJobHandler.SourceType sourceType) {
+ super(
+ warehouse,
+ database,
+ catalogConfig,
+ cdcSourceConfig,
+ new SyncJobHandler(sourceType, cdcSourceConfig, database,
table));
this.table = table;
- this.cdcSourceConfig = Configuration.fromMap(cdcSourceConfig);
}
public SyncTableActionBase withPartitionKeys(String... partitionKeys) {
@@ -105,37 +91,11 @@ public abstract class SyncTableActionBase extends
ActionBase {
return this;
}
- public SyncTableActionBase withTableConfig(Map<String, String>
tableConfig) {
- this.tableConfig = tableConfig;
- return this;
- }
-
public SyncTableActionBase withComputedColumnArgs(List<String>
computedColumnArgs) {
this.computedColumnArgs = computedColumnArgs;
return this;
}
- public SyncTableActionBase withTypeMapping(TypeMapping typeMapping) {
- this.typeMapping = typeMapping;
- return this;
- }
-
- public SyncTableActionBase withMetadataColumns(List<String>
metadataColumns) {
- this.metadataConverters =
- metadataColumns.stream()
- .map(this::metadataConverter)
- .filter(Optional::isPresent)
- .map(Optional::get)
- .toArray(CdcMetadataConverter[]::new);
- return this;
- }
-
- protected Optional<CdcMetadataConverter<?>> metadataConverter(String
column) {
- return Optional.empty();
- }
-
- protected void checkCdcSourceArgument() {}
-
protected abstract Schema retrieveSchema() throws Exception;
protected Schema buildPaimonSchema(Schema retrievedSchema) {
@@ -149,26 +109,22 @@ public abstract class SyncTableActionBase extends
ActionBase {
true);
}
- protected abstract DataStreamSource<String> buildSource() throws Exception;
-
- protected abstract String sourceName();
-
- protected abstract FlatMapFunction<String, RichCdcMultiplexRecord>
recordParse();
-
@Override
- public void build() throws Exception {
- checkCdcSourceArgument();
- catalog.createDatabase(database, true);
-
- boolean caseSensitive = catalog.caseSensitive();
-
- validateCaseInsensitive(caseSensitive);
+ protected void validateCaseSensitivity() {
+ AbstractCatalog.validateCaseInsensitive(caseSensitive, "Database",
database);
+ AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table", table);
+ AbstractCatalog.validateCaseInsensitive(caseSensitive, "Partition
keys", partitionKeys);
+ AbstractCatalog.validateCaseInsensitive(caseSensitive, "Primary keys",
primaryKeys);
+ }
+ @Override
+ protected void beforeBuildingSourceSink() throws Exception {
Identifier identifier = new Identifier(database, table);
// Check if table exists before trying to get or create it
if (catalog.tableExists(identifier)) {
fileStoreTable = (FileStoreTable)
catalog.getTable(identifier).copy(tableConfig);
try {
+ // TODO: test case insensitive with computed columns
Schema retrievedSchema = retrieveSchema();
computedColumns =
buildComputedColumns(computedColumnArgs,
retrievedSchema.fields());
@@ -180,6 +136,7 @@ public abstract class SyncTableActionBase extends
ActionBase {
+ "Schema compatibility check will be skipped.
If you have specified computed columns, "
+ "here will use the existed Paimon table
schema to build them. Please make sure "
+ "the Paimon table has defined all the
argument columns used for computed columns.");
+ // schema evolution will add the computed columns
computedColumns =
buildComputedColumns(computedColumnArgs,
fileStoreTable.schema().fields());
// check partition keys and primary keys in case that user
specified them
@@ -192,20 +149,30 @@ public abstract class SyncTableActionBase extends
ActionBase {
catalog.createTable(identifier, paimonSchema, false);
fileStoreTable = (FileStoreTable)
catalog.getTable(identifier).copy(tableConfig);
}
+ }
- checkComputedColumns(computedColumns);
+ @Override
+ protected FlatMapFunction<String, RichCdcMultiplexRecord> recordParse() {
+ return syncJobHandler.provideRecordParser(
+ caseSensitive, computedColumns, typeMapping,
metadataConverters);
+ }
- DataStream<RichCdcMultiplexRecord> input =
- buildSource().flatMap(recordParse()).name("Parse");
- EventParser.Factory<RichCdcMultiplexRecord> parserFactory =
- () -> new RichCdcMultiplexRecordEventParser(caseSensitive);
+ @Override
+ protected EventParser.Factory<RichCdcMultiplexRecord>
buildEventParserFactory() {
+ boolean caseSensitive = this.caseSensitive;
+ return () -> new RichCdcMultiplexRecordEventParser(caseSensitive);
+ }
+ @Override
+ protected void buildSink(
+ DataStream<RichCdcMultiplexRecord> input,
+ EventParser.Factory<RichCdcMultiplexRecord> parserFactory) {
CdcSinkBuilder<RichCdcMultiplexRecord> sinkBuilder =
new CdcSinkBuilder<RichCdcMultiplexRecord>()
.withInput(input)
.withParserFactory(parserFactory)
.withTable(fileStoreTable)
- .withIdentifier(identifier)
+ .withIdentifier(new Identifier(database, table))
.withCatalogLoader(catalogLoader());
String sinkParallelism =
tableConfig.get(FlinkConnectorOptions.SINK_PARALLELISM.key());
if (sinkParallelism != null) {
@@ -214,39 +181,6 @@ public abstract class SyncTableActionBase extends
ActionBase {
sinkBuilder.build();
}
- protected DataStreamSource<String> buildDataStreamSource(Object source) {
- if (source instanceof Source) {
- return env.fromSource(
- (Source<String, ?, ?>) source,
WatermarkStrategy.noWatermarks(), sourceName());
- }
- if (source instanceof SourceFunction) {
- return env.addSource((SourceFunction<String>) source,
sourceName());
- }
- throw new UnsupportedOperationException("Unrecognized source type");
- }
-
- protected void validateCaseInsensitive(boolean caseSensitive) {
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Database",
database);
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Table", table);
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Partition
keys", partitionKeys);
- AbstractCatalog.validateCaseInsensitive(caseSensitive, "Primary keys",
primaryKeys);
- }
-
- protected void checkComputedColumns(List<ComputedColumn> computedColumns) {
- if (!computedColumns.isEmpty()) {
- List<String> computedFields =
- computedColumns.stream()
- .map(ComputedColumn::columnName)
- .collect(Collectors.toList());
- List<String> fieldNames = fileStoreTable.schema().fieldNames();
- checkArgument(
- new HashSet<>(fieldNames).containsAll(computedFields),
- " Exists Table should contain all computed columns %s, but
are %s.",
- computedFields,
- fieldNames);
- }
- }
-
private void checkConstraints() {
if (!partitionKeys.isEmpty()) {
List<String> actualPartitionKeys = fileStoreTable.partitionKeys();
@@ -271,28 +205,11 @@ public abstract class SyncTableActionBase extends
ActionBase {
}
}
- @VisibleForTesting
- public Map<String, String> tableConfig() {
- return tableConfig;
- }
-
@VisibleForTesting
public FileStoreTable fileStoreTable() {
return fileStoreTable;
}
- // ------------------------------------------------------------------------
- // Flink run methods
- // ------------------------------------------------------------------------
-
- protected abstract String jobName();
-
- @Override
- public void run() throws Exception {
- build();
- execute(jobName());
- }
-
/** Custom exception to indicate issues with schema retrieval. */
public static class SchemaRetrievalException extends Exception {
public SchemaRetrievalException(String message) {
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionBase.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionBase.java
new file mode 100644
index 000000000..3cc77e6f5
--- /dev/null
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/SynchronizationActionBase.java
@@ -0,0 +1,154 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.paimon.flink.action.Action;
+import org.apache.paimon.flink.action.ActionBase;
+import org.apache.paimon.flink.sink.cdc.EventParser;
+import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/** Base {@link Action} for table/database synchronizing job. */
+public abstract class SynchronizationActionBase extends ActionBase {
+
+ private static final long DEFAULT_CHECKPOINT_INTERVAL = 3 * 60 * 1000;
+
+ protected final String database;
+ protected final Configuration cdcSourceConfig;
+ protected final SyncJobHandler syncJobHandler;
+ protected final boolean caseSensitive;
+
+ protected Map<String, String> tableConfig = new HashMap<>();
+ protected TypeMapping typeMapping = TypeMapping.defaultMapping();
+ protected CdcMetadataConverter[] metadataConverters = new
CdcMetadataConverter[] {};
+
+ public SynchronizationActionBase(
+ String warehouse,
+ String database,
+ Map<String, String> catalogConfig,
+ Map<String, String> cdcSourceConfig,
+ SyncJobHandler syncJobHandler) {
+ super(warehouse, catalogConfig);
+ this.database = database;
+ this.cdcSourceConfig = Configuration.fromMap(cdcSourceConfig);
+ this.syncJobHandler = syncJobHandler;
+ this.caseSensitive = catalog.caseSensitive();
+
+ this.syncJobHandler.registerJdbcDriver();
+ }
+
+ public SynchronizationActionBase withTableConfig(Map<String, String>
tableConfig) {
+ this.tableConfig = tableConfig;
+ return this;
+ }
+
+ public SynchronizationActionBase withTypeMapping(TypeMapping typeMapping) {
+ this.typeMapping = typeMapping;
+ return this;
+ }
+
+ public SynchronizationActionBase withMetadataColumns(List<String>
metadataColumns) {
+ this.metadataConverters =
+ metadataColumns.stream()
+ .map(this::metadataConverter)
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .toArray(CdcMetadataConverter[]::new);
+ return this;
+ }
+
+ protected Optional<CdcMetadataConverter<?>> metadataConverter(String
column) {
+ return Optional.empty();
+ }
+
+ @VisibleForTesting
+ public Map<String, String> tableConfig() {
+ return tableConfig;
+ }
+
+ @Override
+ public void build() throws Exception {
+ syncJobHandler.checkRequiredOption();
+
+ catalog.createDatabase(database, true);
+
+ validateCaseSensitivity();
+
+ beforeBuildingSourceSink();
+
+ DataStream<RichCdcMultiplexRecord> input =
+
buildDataStreamSource(buildSource()).flatMap(recordParse()).name("Parse");
+
+ EventParser.Factory<RichCdcMultiplexRecord> parserFactory =
buildEventParserFactory();
+
+ buildSink(input, parserFactory);
+ }
+
+ protected abstract void validateCaseSensitivity();
+
+ protected void beforeBuildingSourceSink() throws Exception {}
+
+ protected Object buildSource() {
+ return syncJobHandler.provideSource();
+ }
+
+ private DataStreamSource<String> buildDataStreamSource(Object source) {
+ if (source instanceof Source) {
+ return env.fromSource(
+ (Source<String, ?, ?>) source,
+ WatermarkStrategy.noWatermarks(),
+ syncJobHandler.provideSourceName());
+ }
+ if (source instanceof SourceFunction) {
+ return env.addSource(
+ (SourceFunction<String>) source,
syncJobHandler.provideSourceName());
+ }
+ throw new UnsupportedOperationException("Unrecognized source type");
+ }
+
+ protected abstract FlatMapFunction<String, RichCdcMultiplexRecord>
recordParse();
+
+ protected abstract EventParser.Factory<RichCdcMultiplexRecord>
buildEventParserFactory();
+
+ protected abstract void buildSink(
+ DataStream<RichCdcMultiplexRecord> input,
+ EventParser.Factory<RichCdcMultiplexRecord> parserFactory);
+
+ @Override
+ public void run() throws Exception {
+ build();
+ if (!env.getCheckpointConfig().isCheckpointingEnabled()) {
+ env.enableCheckpointing(DEFAULT_CHECKPOINT_INTERVAL);
+ }
+ execute(syncJobHandler.provideDefaultJobName());
+ }
+}
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 c40fd5f39..8b98c8c06 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
@@ -32,7 +32,6 @@ import
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions;
import
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.ScanStartupMode;
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;
@@ -56,7 +55,6 @@ import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.SCAN_STARTUP_SPECIFIC_OFFSETS;
-import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Utils for Kafka Action. */
public class KafkaActionUtils {
@@ -67,7 +65,6 @@ public class KafkaActionUtils {
private static final String OFFSET = "offset";
public static KafkaSource<String> buildKafkaSource(Configuration
kafkaConfig) {
- validateKafkaConfig(kafkaConfig);
KafkaSourceBuilder<String> kafkaSourceBuilder = KafkaSource.builder();
List<String> topics =
@@ -228,25 +225,6 @@ public class KafkaActionUtils {
return offsetMap;
}
- private static void validateKafkaConfig(Configuration kafkaConfig) {
- checkArgument(
- kafkaConfig.get(KafkaConnectorOptions.VALUE_FORMAT) != null,
- String.format(
- "kafka-conf [%s] must be specified.",
- KafkaConnectorOptions.VALUE_FORMAT.key()));
-
- checkArgument(
-
!CollectionUtil.isNullOrEmpty(kafkaConfig.get(KafkaConnectorOptions.TOPIC)),
- String.format(
- "kafka-conf [%s] must be specified.",
KafkaConnectorOptions.TOPIC.key()));
-
- checkArgument(
- kafkaConfig.get(KafkaConnectorOptions.PROPS_BOOTSTRAP_SERVERS)
!= null,
- String.format(
- "kafka-conf [%s] must be specified.",
- KafkaConnectorOptions.PROPS_BOOTSTRAP_SERVERS.key()));
- }
-
private static String kafkaPropertiesGroupId(Configuration kafkaConfig) {
String groupId = kafkaConfig.get(KafkaConnectorOptions.PROPS_GROUP_ID);
if (StringUtils.isEmpty(groupId)) {
@@ -256,12 +234,12 @@ public class KafkaActionUtils {
return groupId;
}
- static DataFormat getDataFormat(Configuration kafkaConfig) {
+ public static DataFormat getDataFormat(Configuration kafkaConfig) {
return
DataFormat.fromConfigString(kafkaConfig.get(KafkaConnectorOptions.VALUE_FORMAT));
}
- static MessageQueueSchemaUtils.ConsumerWrapper getKafkaEarliestConsumer(
- Configuration kafkaConfig, String topic) {
+ public static MessageQueueSchemaUtils.ConsumerWrapper
getKafkaEarliestConsumer(
+ Configuration kafkaConfig) {
Properties props = new Properties();
props.put(
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
@@ -275,6 +253,7 @@ public class KafkaActionUtils {
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
+ String topic = kafkaConfig.get(KafkaConnectorOptions.TOPIC).get(0);
// the return may be null in older versions of the Kafka client
List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
if (partitionInfos == null || partitionInfos.isEmpty()) {
@@ -291,19 +270,21 @@ public class KafkaActionUtils {
consumer.assign(topicPartitions);
consumer.seekToBeginning(topicPartitions);
- return new KafkaConsumerWrapper(consumer);
+ return new KafkaConsumerWrapper(consumer, topic);
}
private static class KafkaConsumerWrapper implements
MessageQueueSchemaUtils.ConsumerWrapper {
private final KafkaConsumer<String, String> consumer;
+ private final String topic;
- KafkaConsumerWrapper(KafkaConsumer<String, String> kafkaConsumer) {
+ KafkaConsumerWrapper(KafkaConsumer<String, String> kafkaConsumer,
String topic) {
this.consumer = kafkaConsumer;
+ this.topic = topic;
}
@Override
- public List<String> getRecords(String topic, int pollTimeOutMills) {
+ public List<String> getRecords(int pollTimeOutMills) {
ConsumerRecords<String, String> consumerRecords =
consumer.poll(Duration.ofMillis(pollTimeOutMills));
return
StreamSupport.stream(consumerRecords.records(topic).spliterator(), false)
@@ -311,6 +292,11 @@ public class KafkaActionUtils {
.collect(Collectors.toList());
}
+ @Override
+ public String topic() {
+ return topic;
+ }
+
@Override
public void close() {
consumer.close();
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 88419bc1b..4c87307e3 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
@@ -18,45 +18,19 @@
package org.apache.paimon.flink.action.cdc.kafka;
-import org.apache.paimon.flink.action.cdc.MessageQueueSyncDatabaseActionBase;
-import org.apache.paimon.flink.action.cdc.format.DataFormat;
-
-import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.paimon.flink.action.cdc.SyncDatabaseActionBase;
+import org.apache.paimon.flink.action.cdc.SyncJobHandler;
import java.util.Map;
/** Synchronize database from Kafka. */
-public class KafkaSyncDatabaseAction extends
MessageQueueSyncDatabaseActionBase {
+public class KafkaSyncDatabaseAction extends SyncDatabaseActionBase {
public KafkaSyncDatabaseAction(
String warehouse,
String database,
Map<String, String> catalogConfig,
Map<String, String> kafkaConfig) {
- super(warehouse, database, catalogConfig, kafkaConfig);
- }
-
- @Override
- protected DataStreamSource<String> buildSource() throws Exception {
- return env.fromSource(
- KafkaActionUtils.buildKafkaSource(cdcSourceConfig),
- WatermarkStrategy.noWatermarks(),
- sourceName());
- }
-
- @Override
- protected String sourceName() {
- return "Kafka Source";
- }
-
- @Override
- protected DataFormat getDataFormat() {
- return KafkaActionUtils.getDataFormat(cdcSourceConfig);
- }
-
- @Override
- protected String jobName() {
- return String.format("Kafka-Paimon Database Sync: %s", database);
+ super(warehouse, database, catalogConfig, kafkaConfig,
SyncJobHandler.SourceType.KAFKA);
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseActionFactory.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseActionFactory.java
index 700720ba1..7f787196b 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseActionFactory.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseActionFactory.java
@@ -53,11 +53,11 @@ public class KafkaSyncDatabaseActionFactory implements
ActionFactory {
optionalConfigMap(params, CATALOG_CONF),
optionalConfigMap(params, KAFKA_CONF));
- action.withTableConfig(optionalConfigMap(params, TABLE_CONF))
- .withTablePrefix(params.get(TABLE_PREFIX))
+ action.withTablePrefix(params.get(TABLE_PREFIX))
.withTableSuffix(params.get(TABLE_SUFFIX))
.includingTables(params.get(INCLUDING_TABLES))
- .excludingTables(params.get(EXCLUDING_TABLES));
+ .excludingTables(params.get(EXCLUDING_TABLES))
+ .withTableConfig(optionalConfigMap(params, TABLE_CONF));
if (params.has(TYPE_MAPPING)) {
String[] options = params.get(TYPE_MAPPING).split(",");
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 0e90bda54..0c2d85407 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,12 +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.MessageQueueSyncTableActionBase;
-import org.apache.paimon.flink.action.cdc.format.DataFormat;
-
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
-import org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions;
+import org.apache.paimon.flink.action.cdc.SyncJobHandler;
import java.util.Map;
@@ -36,36 +32,12 @@ public class KafkaSyncTableAction extends
MessageQueueSyncTableActionBase {
String table,
Map<String, String> catalogConfig,
Map<String, String> kafkaConfig) {
- super(warehouse, database, table, catalogConfig, kafkaConfig);
- }
-
- @Override
- protected DataStreamSource<String> buildSource() {
- return
buildDataStreamSource(KafkaActionUtils.buildKafkaSource(cdcSourceConfig));
- }
-
- @Override
- protected String topic() {
- return cdcSourceConfig.get(KafkaConnectorOptions.TOPIC).get(0);
- }
-
- @Override
- protected MessageQueueSchemaUtils.ConsumerWrapper consumer(String topic) {
- return KafkaActionUtils.getKafkaEarliestConsumer(cdcSourceConfig,
topic);
- }
-
- @Override
- protected DataFormat getDataFormat() {
- return KafkaActionUtils.getDataFormat(cdcSourceConfig);
- }
-
- @Override
- protected String sourceName() {
- return "Kafka Source";
- }
-
- @Override
- protected String jobName() {
- return String.format("Kafka-Paimon Table Sync: %s.%s", database,
table);
+ super(
+ warehouse,
+ database,
+ table,
+ catalogConfig,
+ kafkaConfig,
+ SyncJobHandler.SourceType.KAFKA);
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBActionUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBActionUtils.java
index f667b3bbb..c1da23b0b 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBActionUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBActionUtils.java
@@ -33,8 +33,6 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
-import static org.apache.paimon.utils.Preconditions.checkArgument;
-
/**
* Utility class for MongoDB-related actions.
*
@@ -87,7 +85,6 @@ public class MongoDBActionUtils {
public static MongoDBSource<String> buildMongodbSource(
Configuration mongodbConfig, String tableList) {
- validateMongodbConfig(mongodbConfig);
MongoDBSourceBuilder<String> sourceBuilder = MongoDBSource.builder();
if (mongodbConfig.contains(MongoDBSourceOptions.USERNAME)
@@ -140,17 +137,4 @@ public class MongoDBActionUtils {
return sourceBuilder.deserializer(schema).build();
}
-
- private static void validateMongodbConfig(Configuration mongodbConfig) {
- checkArgument(
- mongodbConfig.get(MongoDBSourceOptions.HOSTS) != null,
- String.format(
- "mongodb-conf [%s] must be specified.",
MongoDBSourceOptions.HOSTS.key()));
-
- checkArgument(
- mongodbConfig.get(MongoDBSourceOptions.DATABASE) != null,
- String.format(
- "mongodb-conf [%s] must be specified.",
- MongoDBSourceOptions.DATABASE.key()));
- }
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBRecordParser.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBRecordParser.java
index 31ce83df1..1307c070a 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBRecordParser.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBRecordParser.java
@@ -30,7 +30,6 @@ import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
-import java.util.Collections;
import java.util.List;
/**
@@ -63,10 +62,6 @@ public class MongoDBRecordParser implements
FlatMapFunction<String, RichCdcMulti
private final Configuration mongodbConfig;
private JsonNode root;
- public MongoDBRecordParser(boolean caseSensitive, Configuration
mongodbConfig) {
- this(caseSensitive, Collections.emptyList(), mongodbConfig);
- }
-
public MongoDBRecordParser(
boolean caseSensitive,
List<ComputedColumn> computedColumns,
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
index 2ced0edb9..3fcfccd07 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseAction.java
@@ -18,19 +18,14 @@
package org.apache.paimon.flink.action.cdc.mongodb;
-import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.action.cdc.CdcActionCommonUtils;
import org.apache.paimon.flink.action.cdc.SyncDatabaseActionBase;
-import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
+import org.apache.paimon.flink.action.cdc.SyncJobHandler;
import com.ververica.cdc.connectors.mongodb.source.MongoDBSource;
import com.ververica.cdc.connectors.mongodb.source.config.MongoDBSourceOptions;
-import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.Collections;
import java.util.Map;
/**
@@ -59,35 +54,16 @@ public class MongoDBSyncDatabaseAction extends
SyncDatabaseActionBase {
String database,
Map<String, String> catalogConfig,
Map<String, String> mongodbConfig) {
- super(warehouse, database, catalogConfig, mongodbConfig);
+ super(warehouse, database, catalogConfig, mongodbConfig,
SyncJobHandler.SourceType.MONGODB);
}
@Override
- protected DataStreamSource<String> buildSource() throws Exception {
- List<Identifier> excludedTables = new ArrayList<>();
- MongoDBSource<String> source =
- MongoDBActionUtils.buildMongodbSource(
- cdcSourceConfig,
- CdcActionCommonUtils.combinedModeTableList(
-
cdcSourceConfig.get(MongoDBSourceOptions.DATABASE),
- includingTables,
- excludedTables));
- return env.fromSource(source, WatermarkStrategy.noWatermarks(),
sourceName());
- }
-
- @Override
- protected String sourceName() {
- return "MongoDB Source";
- }
-
- @Override
- protected FlatMapFunction<String, RichCdcMultiplexRecord> recordParse() {
- boolean caseSensitive = catalog.caseSensitive();
- return new MongoDBRecordParser(caseSensitive, cdcSourceConfig);
- }
-
- @Override
- protected String jobName() {
- return String.format("MongoDB-Paimon Database Sync: %s", database);
+ protected MongoDBSource<String> buildSource() {
+ return MongoDBActionUtils.buildMongodbSource(
+ cdcSourceConfig,
+ CdcActionCommonUtils.combinedModeTableList(
+ cdcSourceConfig.get(MongoDBSourceOptions.DATABASE),
+ includingTables,
+ Collections.emptyList()));
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseActionFactory.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseActionFactory.java
index e9c61582c..654572c48 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseActionFactory.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseActionFactory.java
@@ -50,11 +50,11 @@ public class MongoDBSyncDatabaseActionFactory implements
ActionFactory {
optionalConfigMap(params, CATALOG_CONF),
optionalConfigMap(params, MONGODB_CONF));
- action.withTableConfig(optionalConfigMap(params, TABLE_CONF))
- .withTablePrefix(params.get(TABLE_PREFIX))
+ action.withTablePrefix(params.get(TABLE_PREFIX))
.withTableSuffix(params.get(TABLE_SUFFIX))
.includingTables(params.get(INCLUDING_TABLES))
- .excludingTables(params.get(EXCLUDING_TABLES));
+ .excludingTables(params.get(EXCLUDING_TABLES))
+ .withTableConfig(optionalConfigMap(params, TABLE_CONF));
return Optional.of(action);
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java
index 08aacf9cc..7ee875f8d 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncTableAction.java
@@ -18,18 +18,15 @@
package org.apache.paimon.flink.action.cdc.mongodb;
+import org.apache.paimon.flink.action.cdc.SyncJobHandler;
import org.apache.paimon.flink.action.cdc.SyncTableActionBase;
-import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
import org.apache.paimon.schema.Schema;
+import com.ververica.cdc.connectors.mongodb.source.MongoDBSource;
import com.ververica.cdc.connectors.mongodb.source.config.MongoDBSourceOptions;
-import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
import java.util.Map;
-import static org.apache.paimon.utils.Preconditions.checkArgument;
-
/**
* Represents an action to synchronize a specific MongoDB table with a target
system.
*
@@ -57,47 +54,26 @@ public class MongoDBSyncTableAction extends
SyncTableActionBase {
String table,
Map<String, String> catalogConfig,
Map<String, String> mongodbConfig) {
- super(warehouse, database, table, catalogConfig, mongodbConfig);
- }
-
- @Override
- protected void checkCdcSourceArgument() {
- checkArgument(
- cdcSourceConfig.contains(MongoDBSourceOptions.COLLECTION),
- String.format(
- "mongodb-conf [%s] must be specified.",
- MongoDBSourceOptions.COLLECTION.key()));
+ super(
+ warehouse,
+ database,
+ table,
+ catalogConfig,
+ mongodbConfig,
+ SyncJobHandler.SourceType.MONGODB);
}
@Override
- protected Schema retrieveSchema() throws Exception {
- boolean caseSensitive = catalog.caseSensitive();
+ protected Schema retrieveSchema() {
return MongodbSchemaUtils.getMongodbSchema(cdcSourceConfig,
caseSensitive);
}
@Override
- protected DataStreamSource<String> buildSource() throws Exception {
+ protected MongoDBSource<String> buildSource() {
String tableList =
cdcSourceConfig.get(MongoDBSourceOptions.DATABASE)
+ "\\."
+ cdcSourceConfig.get(MongoDBSourceOptions.COLLECTION);
- return buildDataStreamSource(
- MongoDBActionUtils.buildMongodbSource(cdcSourceConfig,
tableList));
- }
-
- @Override
- protected String sourceName() {
- return "MongoDB Source";
- }
-
- @Override
- protected FlatMapFunction<String, RichCdcMultiplexRecord> recordParse() {
- boolean caseSensitive = catalog.caseSensitive();
- return new MongoDBRecordParser(caseSensitive, computedColumns,
cdcSourceConfig);
- }
-
- @Override
- protected String jobName() {
- return String.format("MongoDB-Paimon Table Sync: %s.%s", database,
table);
+ return MongoDBActionUtils.buildMongodbSource(cdcSourceConfig,
tableList);
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
index b5f5047b2..9cfa8eca8 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
@@ -53,7 +53,6 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import static
org.apache.paimon.flink.action.cdc.TypeMapping.TypeMappingMode.TINYINT1_NOT_BOOL;
-import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Utils for MySQL Action. */
public class MySqlActionUtils {
@@ -133,7 +132,6 @@ public class MySqlActionUtils {
public static MySqlSource<String> buildMySqlSource(
Configuration mySqlConfig, String tableList) {
- validateMySqlConfig(mySqlConfig);
MySqlSourceBuilder<String> sourceBuilder = MySqlSource.builder();
sourceBuilder
@@ -241,27 +239,4 @@ public class MySqlActionUtils {
}
}
}
-
- private static void validateMySqlConfig(Configuration mySqlConfig) {
- checkArgument(
- mySqlConfig.get(MySqlSourceOptions.HOSTNAME) != null,
- String.format(
- "mysql-conf [%s] must be specified.",
MySqlSourceOptions.HOSTNAME.key()));
-
- checkArgument(
- mySqlConfig.get(MySqlSourceOptions.USERNAME) != null,
- String.format(
- "mysql-conf [%s] must be specified.",
MySqlSourceOptions.USERNAME.key()));
-
- checkArgument(
- mySqlConfig.get(MySqlSourceOptions.PASSWORD) != null,
- String.format(
- "mysql-conf [%s] must be specified.",
MySqlSourceOptions.PASSWORD.key()));
-
- checkArgument(
- mySqlConfig.get(MySqlSourceOptions.DATABASE_NAME) != null,
- String.format(
- "mysql-conf [%s] must be specified.",
- MySqlSourceOptions.DATABASE_NAME.key()));
- }
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlRecordParser.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlRecordParser.java
index 9f407173a..e5ba73e07 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlRecordParser.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlRecordParser.java
@@ -103,14 +103,6 @@ public class MySqlRecordParser implements
FlatMapFunction<String, RichCdcMultipl
private final Set<String> nonPkTables = new HashSet<>();
- public MySqlRecordParser(
- Configuration mySqlConfig,
- boolean caseSensitive,
- TypeMapping typeMapping,
- CdcMetadataConverter[] metadataConverters) {
- this(mySqlConfig, caseSensitive, Collections.emptyList(), typeMapping,
metadataConverters);
- }
-
public MySqlRecordParser(
Configuration mySqlConfig,
boolean caseSensitive,
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
index e44aeb452..67bf7927b 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseAction.java
@@ -25,10 +25,10 @@ import org.apache.paimon.flink.action.Action;
import org.apache.paimon.flink.action.cdc.CdcActionCommonUtils;
import org.apache.paimon.flink.action.cdc.CdcMetadataConverter;
import org.apache.paimon.flink.action.cdc.SyncDatabaseActionBase;
+import org.apache.paimon.flink.action.cdc.SyncJobHandler;
import org.apache.paimon.flink.action.cdc.TableNameConverter;
import org.apache.paimon.flink.action.cdc.mysql.schema.MySqlSchemasInfo;
import org.apache.paimon.flink.action.cdc.mysql.schema.MySqlTableInfo;
-import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
@@ -36,9 +36,6 @@ import org.apache.paimon.utils.Preconditions;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.config.MySqlSourceOptions;
-import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -109,9 +106,8 @@ public class MySqlSyncDatabaseAction extends
SyncDatabaseActionBase {
String database,
Map<String, String> catalogConfig,
Map<String, String> mySqlConfig) {
- super(warehouse, database, catalogConfig, mySqlConfig);
+ super(warehouse, database, catalogConfig, mySqlConfig,
SyncJobHandler.SourceType.MYSQL);
this.mode = DIVIDED;
- MySqlActionUtils.registerJdbcDriver();
}
public MySqlSyncDatabaseAction ignoreIncompatible(boolean
ignoreIncompatible) {
@@ -125,17 +121,7 @@ public class MySqlSyncDatabaseAction extends
SyncDatabaseActionBase {
}
@Override
- protected void checkCdcSourceArgument() {
- checkArgument(
- !cdcSourceConfig.contains(MySqlSourceOptions.TABLE_NAME),
- MySqlSourceOptions.TABLE_NAME.key()
- + " cannot be set for mysql-sync-database. "
- + "If you want to sync several MySQL tables into one
Paimon table, "
- + "use mysql-sync-table instead.");
- }
-
- @Override
- protected DataStreamSource<String> buildSource() throws Exception {
+ protected void beforeBuildingSourceSink() throws Exception {
boolean caseSensitive = catalog.caseSensitive();
Pattern includingPattern = Pattern.compile(includingTables);
Pattern excludingPattern =
@@ -197,34 +183,18 @@ public class MySqlSyncDatabaseAction extends
SyncDatabaseActionBase {
!monitoredTables.isEmpty(),
"No tables to be synchronized. Possible cause is the schemas
of all tables in specified "
+ "MySQL database are not compatible with those of
existed Paimon tables. Please check the log.");
-
- MySqlSource<String> source =
- MySqlActionUtils.buildMySqlSource(
- cdcSourceConfig,
- tableList(
- mode,
-
cdcSourceConfig.get(MySqlSourceOptions.DATABASE_NAME),
- includingTables,
- monitoredTables,
- excludedTables));
- return env.fromSource(source, WatermarkStrategy.noWatermarks(),
sourceName());
- }
-
- @Override
- protected String sourceName() {
- return "MySQL Source";
- }
-
- @Override
- protected FlatMapFunction<String, RichCdcMultiplexRecord> recordParse() {
- boolean caseSensitive = catalog.caseSensitive();
- return new MySqlRecordParser(
- cdcSourceConfig, caseSensitive, typeMapping,
metadataConverters);
}
@Override
- protected String jobName() {
- return String.format("MySQL-Paimon Database Sync: %s", database);
+ protected MySqlSource<String> buildSource() {
+ return MySqlActionUtils.buildMySqlSource(
+ cdcSourceConfig,
+ tableList(
+ mode,
+ cdcSourceConfig.get(MySqlSourceOptions.DATABASE_NAME),
+ includingTables,
+ monitoredTables,
+ excludedTables));
}
private void logNonPkTables(List<Identifier> nonPkTables) {
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
index 4d0f05664..b7510d08a 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableAction.java
@@ -21,15 +21,14 @@ package org.apache.paimon.flink.action.cdc.mysql;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.action.Action;
import org.apache.paimon.flink.action.cdc.CdcMetadataConverter;
+import org.apache.paimon.flink.action.cdc.SyncJobHandler;
import org.apache.paimon.flink.action.cdc.SyncTableActionBase;
import org.apache.paimon.flink.action.cdc.mysql.schema.MySqlSchemasInfo;
import org.apache.paimon.flink.action.cdc.mysql.schema.MySqlTableInfo;
-import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
import org.apache.paimon.schema.Schema;
+import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.config.MySqlSourceOptions;
-import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
import java.util.ArrayList;
import java.util.List;
@@ -82,8 +81,13 @@ public class MySqlSyncTableAction extends
SyncTableActionBase {
String table,
Map<String, String> catalogConfig,
Map<String, String> mySqlConfig) {
- super(warehouse, database, table, catalogConfig, mySqlConfig);
- MySqlActionUtils.registerJdbcDriver();
+ super(
+ warehouse,
+ database,
+ table,
+ catalogConfig,
+ mySqlConfig,
+ SyncJobHandler.SourceType.MYSQL);
}
@Override
@@ -91,14 +95,6 @@ public class MySqlSyncTableAction extends
SyncTableActionBase {
return Optional.of(MySqlMetadataProcessor.converter(column));
}
- @Override
- protected void checkCdcSourceArgument() {
- checkArgument(
- cdcSourceConfig.contains(MySqlSourceOptions.TABLE_NAME),
- String.format(
- "mysql-conf [%s] must be specified.",
MySqlSourceOptions.TABLE_NAME.key()));
- }
-
@Override
protected Schema retrieveSchema() throws Exception {
this.mySqlSchemasInfo =
@@ -107,36 +103,19 @@ public class MySqlSyncTableAction extends
SyncTableActionBase {
monitorTablePredication(),
new ArrayList<>(),
typeMapping,
- catalog.caseSensitive());
+ caseSensitive);
validateMySqlTableInfos(mySqlSchemasInfo);
MySqlTableInfo tableInfo = mySqlSchemasInfo.mergeAll();
return tableInfo.schema();
}
@Override
- protected DataStreamSource<String> buildSource() throws Exception {
+ protected MySqlSource<String> buildSource() {
String tableList =
mySqlSchemasInfo.pkTables().stream()
.map(i -> i.getDatabaseName() + "\\." +
i.getObjectName())
.collect(Collectors.joining("|"));
- return
buildDataStreamSource(MySqlActionUtils.buildMySqlSource(cdcSourceConfig,
tableList));
- }
-
- @Override
- protected String sourceName() {
- return "MySQL Source";
- }
-
- @Override
- protected FlatMapFunction<String, RichCdcMultiplexRecord> recordParse() {
- boolean caseSensitive = catalog.caseSensitive();
- return new MySqlRecordParser(
- cdcSourceConfig, caseSensitive, computedColumns, typeMapping,
metadataConverters);
- }
-
- @Override
- protected String jobName() {
- return String.format("MySQL-Paimon Table Sync: %s.%s", database,
table);
+ return MySqlActionUtils.buildMySqlSource(cdcSourceConfig, tableList);
}
private void validateMySqlTableInfos(MySqlSchemasInfo mySqlSchemasInfo) {
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionUtils.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionUtils.java
index bc8bc3f01..11e979c8b 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionUtils.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarActionUtils.java
@@ -67,13 +67,13 @@ import static
org.apache.pulsar.client.api.KeySharedPolicy.stickyHashRange;
/** Utils for Pulsar synchronization. */
public class PulsarActionUtils {
- static final ConfigOption<String> VALUE_FORMAT =
+ public 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 =
+ public static final ConfigOption<String> TOPIC =
ConfigOptions.key("topic")
.stringType()
.noDefaultValue()
@@ -151,9 +151,8 @@ public class PulsarActionUtils {
.defaultValue(true)
.withDescription("To specify the boundedness of a
stream.");
- static PulsarSource<String> buildPulsarSource(Configuration rawConfig) {
+ public static PulsarSource<String> buildPulsarSource(Configuration
rawConfig) {
Configuration pulsarConfig = preprocessPulsarConfig(rawConfig);
- validatePulsarConfig(pulsarConfig);
PulsarSourceBuilder<String> pulsarSourceBuilder =
PulsarSource.builder();
@@ -261,29 +260,6 @@ public class PulsarActionUtils {
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;
@@ -317,56 +293,63 @@ public class PulsarActionUtils {
return cloned;
}
- static DataFormat getDataFormat(Configuration pulsarConfig) {
+ public 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(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);
- }
+ public static MessageQueueSchemaUtils.ConsumerWrapper createPulsarConsumer(
+ Configuration pulsarConfig) {
+ try {
+ SourceConfiguration pulsarSourceConfiguration =
toSourceConfiguration(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);
+
+ String topic =
pulsarConfig.get(PulsarActionUtils.TOPIC).split(",")[0];
+ 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();
+ // Create the consumer configuration by using common utils.
+ Consumer<String> consumer = consumerBuilder.subscribe();
- return new PulsarConsumerWrapper(consumer);
+ return new PulsarConsumerWrapper(consumer, topic);
+ } catch (PulsarClientException e) {
+ throw new RuntimeException(e);
+ }
}
private static class PulsarConsumerWrapper implements
MessageQueueSchemaUtils.ConsumerWrapper {
private final Consumer<String> consumer;
+ private final String topic;
- PulsarConsumerWrapper(Consumer<String> consumer) {
+ PulsarConsumerWrapper(Consumer<String> consumer, String topic) {
this.consumer = consumer;
+ this.topic = topic;
}
@Override
- public List<String> getRecords(String topic, int pollTimeOutMills) {
+ public List<String> getRecords(int pollTimeOutMills) {
try {
Message<String> message = consumer.receive(pollTimeOutMills,
TimeUnit.MILLISECONDS);
return message == null
@@ -377,6 +360,11 @@ public class PulsarActionUtils {
}
}
+ @Override
+ public String topic() {
+ return topic;
+ }
+
@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/PulsarSyncDatabaseAction.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseAction.java
index 589332cdf..fa7b7353e 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseAction.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseAction.java
@@ -18,45 +18,19 @@
package org.apache.paimon.flink.action.cdc.pulsar;
-import org.apache.paimon.flink.action.cdc.MessageQueueSyncDatabaseActionBase;
-import org.apache.paimon.flink.action.cdc.format.DataFormat;
-
-import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.paimon.flink.action.cdc.SyncDatabaseActionBase;
+import org.apache.paimon.flink.action.cdc.SyncJobHandler;
import java.util.Map;
/** Synchronize database from Pulsar. */
-public class PulsarSyncDatabaseAction extends
MessageQueueSyncDatabaseActionBase {
+public class PulsarSyncDatabaseAction extends SyncDatabaseActionBase {
public PulsarSyncDatabaseAction(
String warehouse,
String database,
Map<String, String> catalogConfig,
Map<String, String> pulsarConfig) {
- super(warehouse, database, catalogConfig, pulsarConfig);
- }
-
- @Override
- protected DataStreamSource<String> buildSource() throws Exception {
- return env.fromSource(
- PulsarActionUtils.buildPulsarSource(cdcSourceConfig),
- WatermarkStrategy.noWatermarks(),
- sourceName());
- }
-
- @Override
- protected DataFormat getDataFormat() {
- return PulsarActionUtils.getDataFormat(cdcSourceConfig);
- }
-
- @Override
- protected String sourceName() {
- return "Pulsar Source";
- }
-
- @Override
- protected String jobName() {
- return String.format("Pulsar-Paimon Database Sync: %s", database);
+ super(warehouse, database, catalogConfig, pulsarConfig,
SyncJobHandler.SourceType.PULSAR);
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseActionFactory.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseActionFactory.java
index f042798ff..45301254a 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseActionFactory.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarSyncDatabaseActionFactory.java
@@ -53,11 +53,11 @@ public class PulsarSyncDatabaseActionFactory implements
ActionFactory {
optionalConfigMap(params, CATALOG_CONF),
optionalConfigMap(params, PULSAR_CONF));
- action.withTableConfig(optionalConfigMap(params, TABLE_CONF))
- .withTablePrefix(params.get(TABLE_PREFIX))
+ action.withTablePrefix(params.get(TABLE_PREFIX))
.withTableSuffix(params.get(TABLE_SUFFIX))
.includingTables(params.get(INCLUDING_TABLES))
- .excludingTables(params.get(EXCLUDING_TABLES));
+ .excludingTables(params.get(EXCLUDING_TABLES))
+ .withTableConfig(optionalConfigMap(params, TABLE_CONF));
if (params.has(TYPE_MAPPING)) {
String[] options = params.get(TYPE_MAPPING).split(",");
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
index d21713828..c94c4581a 100644
---
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
@@ -18,12 +18,8 @@
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.flink.streaming.api.datastream.DataStreamSource;
-import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.paimon.flink.action.cdc.SyncJobHandler;
import java.util.Map;
@@ -36,40 +32,12 @@ public class PulsarSyncTableAction extends
MessageQueueSyncTableActionBase {
String table,
Map<String, String> catalogConfig,
Map<String, String> pulsarConfig) {
- super(warehouse, database, table, catalogConfig, pulsarConfig);
- }
-
- @Override
- protected DataStreamSource<String> buildSource() {
- return
buildDataStreamSource(PulsarActionUtils.buildPulsarSource(cdcSourceConfig));
- }
-
- @Override
- protected String topic() {
- return
cdcSourceConfig.get(PulsarActionUtils.TOPIC).split(",")[0].trim();
- }
-
- @Override
- protected MessageQueueSchemaUtils.ConsumerWrapper consumer(String topic) {
- try {
- return PulsarActionUtils.createPulsarConsumer(cdcSourceConfig,
topic);
- } catch (PulsarClientException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- protected DataFormat getDataFormat() {
- return PulsarActionUtils.getDataFormat(cdcSourceConfig);
- }
-
- @Override
- protected String sourceName() {
- return "Pulsar Source";
- }
-
- @Override
- protected String jobName() {
- return String.format("Pulsar-Paimon Table Sync: %s.%s", database,
table);
+ super(
+ warehouse,
+ database,
+ table,
+ catalogConfig,
+ pulsarConfig,
+ SyncJobHandler.SourceType.PULSAR);
}
}
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncDatabaseActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncDatabaseActionITCase.java
index 2a0b6ef3c..fdc946565 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncDatabaseActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaCanalSyncDatabaseActionITCase.java
@@ -252,7 +252,7 @@ public class KafkaCanalSyncDatabaseActionITCase extends
KafkaActionITCaseBase {
.satisfies(
anyCauseMatches(
IllegalArgumentException.class,
- "kafka-conf [topic] must be specified."));
+ "kafka_conf [topic] must be specified."));
}
@Test
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaMaxwellSyncDatabaseActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaMaxwellSyncDatabaseActionITCase.java
index d85217844..60ce40773 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaMaxwellSyncDatabaseActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaMaxwellSyncDatabaseActionITCase.java
@@ -214,7 +214,7 @@ public class KafkaMaxwellSyncDatabaseActionITCase extends
KafkaActionITCaseBase
.satisfies(
AssertionUtils.anyCauseMatches(
IllegalArgumentException.class,
- "kafka-conf [topic] must be specified."));
+ "kafka_conf [topic] must be specified."));
}
@Test
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaOggSyncDatabaseActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaOggSyncDatabaseActionITCase.java
index d13c61b99..c930e1054 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaOggSyncDatabaseActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaOggSyncDatabaseActionITCase.java
@@ -215,7 +215,7 @@ public class KafkaOggSyncDatabaseActionITCase extends
KafkaActionITCaseBase {
.satisfies(
AssertionUtils.anyCauseMatches(
IllegalArgumentException.class,
- "kafka-conf [topic] must be specified."));
+ "kafka_conf [topic] must be specified."));
}
@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 a62a4380e..35eecbfad 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
@@ -61,8 +61,7 @@ public class KafkaSchemaITCase extends KafkaActionITCaseBase {
Schema kafkaSchema =
MessageQueueSchemaUtils.getSchema(
- getKafkaEarliestConsumer(kafkaConfig, topic),
- topic,
+ getKafkaEarliestConsumer(kafkaConfig),
getDataFormat(kafkaConfig),
TypeMapping.defaultMapping());
List<DataField> fields = new ArrayList<>();
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
index 1dd87b121..5074e759d 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncDatabaseActionITCase.java
@@ -212,9 +212,9 @@ public class MySqlSyncDatabaseActionITCase extends
MySqlActionITCaseBase {
assertThatThrownBy(action::run)
.isInstanceOf(IllegalArgumentException.class)
.hasMessage(
- "table-name cannot be set for mysql-sync-database. "
+ "table-name cannot be set for mysql_sync_database. "
+ "If you want to sync several MySQL tables
into one Paimon table, "
- + "use mysql-sync-table instead.");
+ + "use mysql_sync_table instead.");
}
@Test
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
index c7a04fc97..e7abc8902 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
+++
b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
@@ -26,14 +26,18 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.CommonTestUtils;
import org.apache.paimon.utils.JsonSerdeUtil;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import java.sql.Statement;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -960,4 +964,38 @@ public class MySqlSyncTableActionITCase extends
MySqlActionITCaseBase {
private FileStoreTable getFileStoreTable() throws Exception {
return getFileStoreTable(tableName);
}
+
+ @Test
+ @Timeout(60)
+ public void testDefaultCheckpointInterval() throws Exception {
+ Map<String, String> mySqlConfig = getBasicMySqlConfig();
+ mySqlConfig.put("database-name", "default_checkpoint");
+ mySqlConfig.put("table-name", "t");
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setRestartStrategy(RestartStrategies.noRestart());
+
+ MySqlSyncTableAction action =
syncTableActionBuilder(mySqlConfig).build();
+ action.withStreamExecutionEnvironment(env);
+
+ Thread thread =
+ new Thread(
+ () -> {
+ try {
+ action.run();
+ } catch (Exception ignore) {
+ }
+ });
+ thread.start();
+
+ CommonTestUtils.waitUtil(
+ () -> env.getCheckpointConfig().isCheckpointingEnabled(),
+ Duration.ofSeconds(5),
+ Duration.ofMillis(100));
+
+ assertThat(env.getCheckpointInterval()).isEqualTo(3 * 60 * 1000);
+
+ thread.interrupt();
+ env.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
index 789f6b414..98dfea01a 100644
---
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
@@ -56,8 +56,7 @@ public class PulsarSchemaITCase extends
PulsarActionITCaseBase {
Schema pulsarSchema =
MessageQueueSchemaUtils.getSchema(
- createPulsarConsumer(pulsarConfig, topic),
- topic,
+ createPulsarConsumer(pulsarConfig),
getDataFormat(pulsarConfig),
TypeMapping.defaultMapping());
List<DataField> fields = new ArrayList<>();
diff --git
a/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/sync_table_setup.sql
b/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/sync_table_setup.sql
index 3e67e9909..7990a0330 100644
---
a/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/sync_table_setup.sql
+++
b/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/sync_table_setup.sql
@@ -334,4 +334,17 @@ CREATE TABLE test_metadata_columns (
pk INT,
_date VARCHAR(10),
PRIMARY KEY (pk)
+);
+
+--
################################################################################
+-- testDefaultCheckpointInterval
+--
################################################################################
+
+CREATE DATABASE default_checkpoint;
+USE default_checkpoint;
+
+CREATE TABLE t (
+ pk INT,
+ v VARCHAR(10),
+ PRIMARY KEY (pk)
);
\ No newline at end of file
