result = new ArrayList<>();
+ StringBuilder current = new StringBuilder();
+ int parenDepth = 0;
+
+ for (int i = 0; i < str.length(); i++) {
+ char c = str.charAt(i);
+ if (c == '(') {
+ parenDepth++;
+ current.append(c);
+ } else if (c == ')') {
+ parenDepth--;
+ current.append(c);
+ } else if (c == ',' && parenDepth == 0) {
+ result.add(current.toString());
+ current = new StringBuilder();
+ } else {
+ current.append(c);
+ }
+ }
+
+ if (current.length() > 0) {
+ result.add(current.toString());
+ }
+
+ return result;
+ }
+
+ private String extractTableName(String str) {
+ if (str == null) {
+ return null;
+ }
+ str = str.trim();
+ if (str.startsWith("`") && str.endsWith("`")) {
+ return str.substring(1, str.length() - 1);
+ }
+ return str;
+ }
+
+ private static DataType mapMySqlType(String typeName, int precision, int scale) {
+ switch (typeName.toUpperCase()) {
+ case "TINYINT":
+ return DataTypes.TINYINT();
+ case "SMALLINT":
+ return DataTypes.SMALLINT();
+ case "INT":
+ case "INTEGER":
+ return DataTypes.INT();
+ case "BIGINT":
+ return DataTypes.BIGINT();
+ case "FLOAT":
+ return DataTypes.FLOAT();
+ case "DOUBLE":
+ return DataTypes.DOUBLE();
+ case "DECIMAL":
+ case "NUMERIC":
+ return DataTypes.DECIMAL(precision > 0 ? precision : 10, scale);
+ case "DATE":
+ return DataTypes.DATE();
+ case "TIME":
+ return DataTypes.TIME(3);
+ case "DATETIME":
+ return DataTypes.TIMESTAMP(3);
+ case "TIMESTAMP":
+ return DataTypes.TIMESTAMP(3);
+ case "CHAR":
+ return DataTypes.CHAR(precision > 0 ? precision : 1);
+ case "VARCHAR":
+ return DataTypes.VARCHAR(precision > 0 ? precision : 255);
+ case "TEXT":
+ case "LONGTEXT":
+ case "MEDIUMTEXT":
+ case "TINYTEXT":
+ return DataTypes.STRING();
+ case "BLOB":
+ case "LONGBLOB":
+ case "MEDIUMBLOB":
+ case "TINYBLOB":
+ return DataTypes.BYTES();
+ case "BOOLEAN":
+ case "BIT":
+ return DataTypes.BOOLEAN();
+ case "ENUM":
+ case "SET":
+ return DataTypes.STRING();
+ case "JSON":
+ return DataTypes.STRING();
+ default:
+ LOG.warn("Unknown MySQL type '{}', defaulting to STRING", typeName);
+ return DataTypes.STRING();
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonDeserializationSchema.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonDeserializationSchema.java
new file mode 100644
index 00000000000..e2a60356196
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonDeserializationSchema.java
@@ -0,0 +1,430 @@
+/*
+ * 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.flink.cdc.connectors.kafka.json.canal;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.cdc.common.annotation.VisibleForTesting;
+import org.apache.flink.cdc.common.data.DecimalData;
+import org.apache.flink.cdc.common.data.GenericRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.DecimalType;
+import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
+import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
+import org.apache.flink.util.Collector;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Deserialization schema that parses Canal JSON bytes from Kafka into Flink CDC {@link Event}
+ * objects.
+ *
+ * Canal JSON format has the following structure:
+ *
+ *
{@code
+ * {
+ * "old": [{"col1": "old_value"}],
+ * "data": [{"col1": "new_value"}],
+ * "type": "INSERT",
+ * "database": "mydb",
+ * "table": "mytable",
+ * "pkNames": ["col1"]
+ * }
+ * }
+ *
+ * The deserializer infers the table schema from the first message for each table and emits a
+ * {@link CreateTableEvent} followed by {@link DataChangeEvent}s.
+ *
+ * @see Alibaba Canal
+ */
+public class CanalJsonDeserializationSchema implements KafkaRecordDeserializationSchema {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(CanalJsonDeserializationSchema.class);
+
+ // Canal JSON field names
+ private static final String FIELD_DATA = "data";
+ private static final String FIELD_OLD = "old";
+ private static final String FIELD_TYPE = "type";
+ private static final String FIELD_DATABASE = "database";
+ private static final String FIELD_TABLE = "table";
+ private static final String FIELD_PK_NAMES = "pkNames";
+ private static final String FIELD_IS_DDL = "isDdl";
+ private static final String FIELD_SQL = "sql";
+
+ // Canal operation types
+ private static final String OP_INSERT = "INSERT";
+ private static final String OP_UPDATE = "UPDATE";
+ private static final String OP_DELETE = "DELETE";
+
+ private final boolean schemaInferEnabled;
+ @Nullable private final String defaultDatabaseName;
+ @Nullable private final String defaultSchemaName;
+ @Nullable private final String defaultTableName;
+
+ private transient ObjectMapper objectMapper;
+
+ // Track seen tables to emit CreateTableEvent only once
+ private transient Map tableSchemas;
+
+ // DDL parser for parsing schema change events
+ private transient CanalDdlParser ddlParser;
+
+ /** Holds the inferred schema and column order for a table. */
+ static class TableSchema implements Serializable {
+ private static final long serialVersionUID = 1L;
+ final List columnNames;
+ final List columnTypes;
+ final List primaryKeys;
+
+ TableSchema(
+ List columnNames, List columnTypes, List primaryKeys) {
+ this.columnNames = columnNames;
+ this.columnTypes = columnTypes;
+ this.primaryKeys = primaryKeys;
+ }
+ }
+
+ public CanalJsonDeserializationSchema(
+ boolean schemaInferEnabled,
+ @Nullable String defaultDatabaseName,
+ @Nullable String defaultSchemaName,
+ @Nullable String defaultTableName) {
+ this.schemaInferEnabled = schemaInferEnabled;
+ this.defaultDatabaseName = defaultDatabaseName;
+ this.defaultSchemaName = defaultSchemaName;
+ this.defaultTableName = defaultTableName;
+ }
+
+ @Override
+ public void open(DeserializationSchema.InitializationContext context) throws Exception {
+ this.objectMapper = new ObjectMapper();
+ this.tableSchemas = new HashMap<>();
+ this.ddlParser = new CanalDdlParser();
+ }
+
+ @Override
+ public void deserialize(ConsumerRecord record, Collector out)
+ throws IOException {
+ if (record.value() == null) {
+ LOG.debug("Received null value from Kafka, skipping.");
+ return;
+ }
+
+ JsonNode root = objectMapper.readTree(record.value());
+ if (root == null || root.isNull()) {
+ LOG.debug("Received empty JSON from Kafka, skipping.");
+ return;
+ }
+
+ boolean isDdl = root.has(FIELD_IS_DDL) && root.get(FIELD_IS_DDL).asBoolean();
+
+ if (isDdl) {
+ processDdlEvent(root, out);
+ return;
+ }
+
+ processDmlEvent(root, out);
+ }
+
+ private void processDdlEvent(JsonNode root, Collector out) {
+ String database =
+ root.has(FIELD_DATABASE) && !root.get(FIELD_DATABASE).isNull()
+ ? root.get(FIELD_DATABASE).asText()
+ : defaultDatabaseName;
+ String table =
+ root.has(FIELD_TABLE) && !root.get(FIELD_TABLE).isNull()
+ ? root.get(FIELD_TABLE).asText()
+ : defaultTableName;
+ String sql =
+ root.has(FIELD_SQL) && !root.get(FIELD_SQL).isNull()
+ ? root.get(FIELD_SQL).asText()
+ : null;
+
+ if (sql == null || sql.trim().isEmpty()) {
+ LOG.warn("DDL message without SQL statement, skipping.");
+ return;
+ }
+
+ List events = ddlParser.parse(sql, database, table, defaultSchemaName);
+ for (SchemaChangeEvent event : events) {
+ out.collect(event);
+ if (event instanceof CreateTableEvent) {
+ tableSchemas.put(event.tableId(), null);
+ } else if (event instanceof org.apache.flink.cdc.common.event.DropTableEvent) {
+ tableSchemas.remove(event.tableId());
+ }
+ }
+ }
+
+ private void processDmlEvent(JsonNode root, Collector out) {
+ TableId tableId = extractTableId(root);
+ String type = root.has(FIELD_TYPE) ? root.get(FIELD_TYPE).asText() : OP_INSERT;
+
+ List primaryKeys = extractPrimaryKeys(root);
+
+ TableSchema tableSchema = tableSchemas.get(tableId);
+ if (tableSchema == null) {
+ JsonNode dataNode = root.get(FIELD_DATA);
+ if (dataNode != null && dataNode.isArray() && dataNode.size() > 0) {
+ JsonNode firstRow = dataNode.get(0);
+ tableSchema = inferSchema(firstRow, primaryKeys);
+ } else {
+ JsonNode oldNode = root.get(FIELD_OLD);
+ if (oldNode != null && oldNode.isArray() && oldNode.size() > 0) {
+ JsonNode firstRow = oldNode.get(0);
+ tableSchema = inferSchema(firstRow, primaryKeys);
+ } else {
+ LOG.warn("Cannot infer schema for table {} from record, skipping.", tableId);
+ return;
+ }
+ }
+ tableSchemas.put(tableId, tableSchema);
+
+ Schema.Builder schemaBuilder = Schema.newBuilder();
+ for (int i = 0; i < tableSchema.columnNames.size(); i++) {
+ schemaBuilder.physicalColumn(
+ tableSchema.columnNames.get(i), tableSchema.columnTypes.get(i));
+ }
+ if (!tableSchema.primaryKeys.isEmpty()) {
+ schemaBuilder.primaryKey(tableSchema.primaryKeys);
+ }
+ out.collect(new CreateTableEvent(tableId, schemaBuilder.build()));
+ }
+
+ JsonNode dataNode = root.get(FIELD_DATA);
+ JsonNode oldNode = root.get(FIELD_OLD);
+
+ switch (type.toUpperCase()) {
+ case OP_INSERT:
+ emitInsertEvents(tableId, tableSchema, dataNode, out);
+ break;
+ case OP_UPDATE:
+ emitUpdateEvents(tableId, tableSchema, dataNode, oldNode, out);
+ break;
+ case OP_DELETE:
+ emitDeleteEvents(tableId, tableSchema, dataNode, oldNode, out);
+ break;
+ default:
+ LOG.warn("Unsupported Canal operation type '{}', skipping record.", type);
+ }
+ }
+
+ private TableId extractTableId(JsonNode root) {
+ String database =
+ root.has(FIELD_DATABASE) && !root.get(FIELD_DATABASE).isNull()
+ ? root.get(FIELD_DATABASE).asText()
+ : (defaultDatabaseName != null ? defaultDatabaseName : "");
+ String table =
+ root.has(FIELD_TABLE) && !root.get(FIELD_TABLE).isNull()
+ ? root.get(FIELD_TABLE).asText()
+ : (defaultTableName != null ? defaultTableName : "unknown");
+
+ if (defaultSchemaName != null) {
+ return TableId.tableId(database, defaultSchemaName, table);
+ } else if (!database.isEmpty()) {
+ return TableId.tableId(database, table);
+ } else {
+ return TableId.tableId(table);
+ }
+ }
+
+ private List extractPrimaryKeys(JsonNode root) {
+ if (!root.has(FIELD_PK_NAMES) || root.get(FIELD_PK_NAMES).isNull()) {
+ return Collections.emptyList();
+ }
+ JsonNode pkNode = root.get(FIELD_PK_NAMES);
+ if (!pkNode.isArray()) {
+ return Collections.emptyList();
+ }
+ List primaryKeys = new ArrayList<>();
+ for (JsonNode pk : pkNode) {
+ primaryKeys.add(pk.asText());
+ }
+ return primaryKeys;
+ }
+
+ private TableSchema inferSchema(JsonNode rowNode, List primaryKeys) {
+ List columnNames = new ArrayList<>();
+ List columnTypes = new ArrayList<>();
+
+ Iterator> fields = rowNode.fields();
+ while (fields.hasNext()) {
+ Map.Entry field = fields.next();
+ columnNames.add(field.getKey());
+ if (schemaInferEnabled) {
+ columnTypes.add(inferDataType(field.getValue()));
+ } else {
+ columnTypes.add(DataTypes.STRING());
+ }
+ }
+
+ return new TableSchema(columnNames, columnTypes, new ArrayList<>(primaryKeys));
+ }
+
+ private DataType inferDataType(JsonNode value) {
+ if (value == null || value.isNull()) {
+ return DataTypes.STRING();
+ }
+ if (value.isBoolean()) {
+ return DataTypes.BOOLEAN();
+ }
+ if (value.isInt()) {
+ return DataTypes.INT();
+ }
+ if (value.isLong()) {
+ return DataTypes.BIGINT();
+ }
+ if (value.isDouble() || value.isFloat()) {
+ return DataTypes.DOUBLE();
+ }
+ if (value.isBigDecimal()) {
+ return DataTypes.DECIMAL(
+ value.decimalValue().precision(), value.decimalValue().scale());
+ }
+ // Default to STRING for text, objects, arrays
+ return DataTypes.STRING();
+ }
+
+ private void emitInsertEvents(
+ TableId tableId, TableSchema tableSchema, JsonNode dataNode, Collector out) {
+ if (dataNode == null || !dataNode.isArray()) {
+ return;
+ }
+ for (JsonNode row : dataNode) {
+ GenericRecordData recordData = convertToRecordData(row, tableSchema);
+ out.collect(DataChangeEvent.insertEvent(tableId, recordData));
+ }
+ }
+
+ private void emitUpdateEvents(
+ TableId tableId,
+ TableSchema tableSchema,
+ JsonNode dataNode,
+ JsonNode oldNode,
+ Collector out) {
+ if (dataNode == null || !dataNode.isArray()) {
+ return;
+ }
+ int size = dataNode.size();
+ for (int i = 0; i < size; i++) {
+ GenericRecordData after = convertToRecordData(dataNode.get(i), tableSchema);
+ GenericRecordData before = null;
+ if (oldNode != null && oldNode.isArray() && i < oldNode.size()) {
+ before = convertToRecordData(oldNode.get(i), tableSchema);
+ }
+ out.collect(DataChangeEvent.updateEvent(tableId, before, after));
+ }
+ }
+
+ private void emitDeleteEvents(
+ TableId tableId,
+ TableSchema tableSchema,
+ JsonNode dataNode,
+ JsonNode oldNode,
+ Collector out) {
+ // For DELETE, Canal puts the deleted data in 'data' field
+ JsonNode sourceNode = (dataNode != null && dataNode.isArray()) ? dataNode : oldNode;
+ if (sourceNode == null || !sourceNode.isArray()) {
+ return;
+ }
+ for (JsonNode row : sourceNode) {
+ GenericRecordData recordData = convertToRecordData(row, tableSchema);
+ out.collect(DataChangeEvent.deleteEvent(tableId, recordData));
+ }
+ }
+
+ private GenericRecordData convertToRecordData(JsonNode rowNode, TableSchema tableSchema) {
+ int arity = tableSchema.columnNames.size();
+ GenericRecordData recordData = new GenericRecordData(arity);
+ for (int i = 0; i < arity; i++) {
+ String columnName = tableSchema.columnNames.get(i);
+ JsonNode value = rowNode.get(columnName);
+ recordData.setField(i, convertValue(value, tableSchema.columnTypes.get(i)));
+ }
+ return recordData;
+ }
+
+ private Object convertValue(JsonNode value, DataType dataType) {
+ if (value == null || value.isNull()) {
+ return null;
+ }
+ switch (dataType.getTypeRoot()) {
+ case CHAR:
+ case VARCHAR:
+ return BinaryStringData.fromString(value.asText());
+ case BOOLEAN:
+ return value.asBoolean();
+ case TINYINT:
+ return (byte) value.asInt();
+ case SMALLINT:
+ return (short) value.asInt();
+ case INTEGER:
+ return value.asInt();
+ case BIGINT:
+ return value.asLong();
+ case FLOAT:
+ return (float) value.asDouble();
+ case DOUBLE:
+ return value.asDouble();
+ case DECIMAL:
+ return DecimalData.fromBigDecimal(
+ value.decimalValue(),
+ ((DecimalType) dataType).getPrecision(),
+ ((DecimalType) dataType).getScale());
+ default:
+ // Fallback: store as string
+ return BinaryStringData.fromString(value.asText());
+ }
+ }
+
+ @Override
+ public TypeInformation getProducedType() {
+ return new EventTypeInfo();
+ }
+
+ @VisibleForTesting
+ Map getTableSchemas() {
+ return tableSchemas;
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSource.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSource.java
new file mode 100644
index 00000000000..d268d9e6703
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSource.java
@@ -0,0 +1,166 @@
+/*
+ * 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.flink.cdc.connectors.kafka.source;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.source.DataSource;
+import org.apache.flink.cdc.common.source.EventSourceProvider;
+import org.apache.flink.cdc.common.source.FlinkSourceProvider;
+import org.apache.flink.cdc.common.source.MetadataAccessor;
+import org.apache.flink.cdc.connectors.kafka.json.canal.CanalJsonDeserializationSchema;
+import org.apache.flink.connector.kafka.source.KafkaSource;
+import org.apache.flink.connector.kafka.source.KafkaSourceBuilder;
+import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** A {@link DataSource} for reading from Kafka topics with Canal JSON format. */
+@Internal
+public class KafkaDataSource implements DataSource {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String topic;
+ private final Properties kafkaProperties;
+ private final String groupId;
+ private final StartupMode startupMode;
+ @Nullable private final String specificOffsets;
+ @Nullable private final Long startupTimestampMillis;
+ private final CanalJsonDeserializationSchema deserializationSchema;
+
+ public KafkaDataSource(
+ String topic,
+ Properties kafkaProperties,
+ String groupId,
+ StartupMode startupMode,
+ @Nullable String specificOffsets,
+ @Nullable Long startupTimestampMillis,
+ CanalJsonDeserializationSchema deserializationSchema) {
+ this.topic = checkNotNull(topic, "topic must not be null");
+ this.kafkaProperties = checkNotNull(kafkaProperties, "kafkaProperties must not be null");
+ this.groupId = checkNotNull(groupId, "groupId must not be null");
+ this.startupMode = checkNotNull(startupMode, "startupMode must not be null");
+ this.specificOffsets = specificOffsets;
+ this.startupTimestampMillis = startupTimestampMillis;
+ this.deserializationSchema =
+ checkNotNull(deserializationSchema, "deserializer must not be null");
+ }
+
+ @Override
+ public EventSourceProvider getEventSourceProvider() {
+ KafkaSourceBuilder sourceBuilder = KafkaSource.builder();
+
+ // Set bootstrap servers from kafka properties
+ if (kafkaProperties.containsKey("bootstrap.servers")) {
+ sourceBuilder.setBootstrapServers(kafkaProperties.get("bootstrap.servers").toString());
+ }
+
+ // Set topics (supports comma-separated list)
+ List topics = Arrays.asList(topic.split(","));
+ sourceBuilder.setTopics(topics.toArray(new String[0]));
+
+ // Set group id
+ sourceBuilder.setGroupId(groupId);
+
+ // Set kafka properties
+ sourceBuilder.setProperties(kafkaProperties);
+
+ // Set starting offsets based on startup mode
+ sourceBuilder.setStartingOffsets(getOffsetsInitializer());
+
+ // Set deserializer
+ sourceBuilder.setDeserializer(deserializationSchema);
+
+ KafkaSource source = sourceBuilder.build();
+ return FlinkSourceProvider.of(source);
+ }
+
+ private OffsetsInitializer getOffsetsInitializer() {
+ switch (startupMode) {
+ case EARLIEST_OFFSET:
+ return OffsetsInitializer.earliest();
+ case LATEST_OFFSET:
+ return OffsetsInitializer.latest();
+ case TIMESTAMP:
+ if (startupTimestampMillis == null) {
+ throw new IllegalArgumentException(
+ "scan.startup.timestamp-millis must be set when startup mode is 'timestamp'");
+ }
+ return OffsetsInitializer.timestamp(startupTimestampMillis);
+ case SPECIFIC_OFFSETS:
+ if (specificOffsets == null) {
+ throw new IllegalArgumentException(
+ "scan.startup.specific-offsets must be set when startup mode is 'specific-offsets'");
+ }
+ return parseSpecificOffsets(specificOffsets);
+ case GROUP_OFFSETS:
+ default:
+ return OffsetsInitializer.committedOffsets();
+ }
+ }
+
+ private OffsetsInitializer parseSpecificOffsets(String offsetsStr) {
+ Map offsets = new HashMap<>();
+ String[] parts = offsetsStr.split(";");
+ for (String part : parts) {
+ String trimmed = part.trim();
+ if (trimmed.isEmpty()) {
+ continue;
+ }
+ // Format: partition:0,offset:42
+ String[] keyValuePairs = trimmed.split(",");
+ int partition = -1;
+ long offset = -1;
+ for (String kv : keyValuePairs) {
+ String[] pair = kv.trim().split(":");
+ if (pair.length == 2) {
+ if (pair[0].trim().equals("partition")) {
+ partition = Integer.parseInt(pair[1].trim());
+ } else if (pair[0].trim().equals("offset")) {
+ offset = Long.parseLong(pair[1].trim());
+ }
+ }
+ }
+ if (partition >= 0 && offset >= 0) {
+ offsets.put(new org.apache.kafka.common.TopicPartition(topic, partition), offset);
+ }
+ }
+ return OffsetsInitializer.offsets(offsets);
+ }
+
+ @Override
+ public MetadataAccessor getMetadataAccessor() {
+ // Kafka does not maintain schemas, return a no-op metadata accessor
+ return new KafkaMetadataAccessor(topic);
+ }
+
+ @Override
+ public boolean isParallelMetadataSource() {
+ // Kafka can have schema changes on different partitions/tables in parallel
+ return true;
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSourceFactory.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSourceFactory.java
new file mode 100644
index 00000000000..4e284004998
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSourceFactory.java
@@ -0,0 +1,183 @@
+/*
+ * 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.flink.cdc.connectors.kafka.source;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.factories.DataSourceFactory;
+import org.apache.flink.cdc.common.factories.Factory;
+import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.source.DataSource;
+import org.apache.flink.cdc.connectors.kafka.json.JsonDeserializationType;
+import org.apache.flink.cdc.connectors.kafka.json.canal.CanalJsonDeserializationSchema;
+import org.apache.flink.table.api.ValidationException;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+import static org.apache.flink.cdc.connectors.kafka.source.KafkaDataSourceOptions.DATABASE_NAME;
+import static org.apache.flink.cdc.connectors.kafka.source.KafkaDataSourceOptions.GROUP_ID;
+import static org.apache.flink.cdc.connectors.kafka.source.KafkaDataSourceOptions.PROPERTIES_PREFIX;
+import static org.apache.flink.cdc.connectors.kafka.source.KafkaDataSourceOptions.SCAN_STARTUP_MODE;
+import static org.apache.flink.cdc.connectors.kafka.source.KafkaDataSourceOptions.SCAN_STARTUP_SPECIFIC_OFFSETS;
+import static org.apache.flink.cdc.connectors.kafka.source.KafkaDataSourceOptions.SCAN_STARTUP_TIMESTAMP_MILLIS;
+import static org.apache.flink.cdc.connectors.kafka.source.KafkaDataSourceOptions.SCHEMA_INFER_ENABLED;
+import static org.apache.flink.cdc.connectors.kafka.source.KafkaDataSourceOptions.SCHEMA_NAME;
+import static org.apache.flink.cdc.connectors.kafka.source.KafkaDataSourceOptions.TABLE_NAME;
+import static org.apache.flink.cdc.connectors.kafka.source.KafkaDataSourceOptions.TOPIC;
+import static org.apache.flink.cdc.connectors.kafka.source.KafkaDataSourceOptions.VALUE_FORMAT;
+
+/** A {@link Factory} to create {@link KafkaDataSource} for reading from Kafka topics. */
+@Internal
+public class KafkaDataSourceFactory implements DataSourceFactory {
+
+ public static final String IDENTIFIER = "kafka";
+
+ @Override
+ public DataSource createDataSource(Context context) {
+ FactoryHelper helper = FactoryHelper.createFactoryHelper(this, context);
+ helper.validateExcept(PROPERTIES_PREFIX);
+
+ org.apache.flink.cdc.common.configuration.Configuration config =
+ context.getFactoryConfiguration();
+
+ // Validate required options
+ String topic = config.get(TOPIC);
+ if (topic == null) {
+ throw new ValidationException("'topic' is required for Kafka source.");
+ }
+
+ // Extract Kafka properties from the configuration
+ Properties kafkaProperties = new Properties();
+ Map allOptions = config.toMap();
+ allOptions.keySet().stream()
+ .filter(key -> key.startsWith(PROPERTIES_PREFIX))
+ .forEach(
+ key -> {
+ final String value = allOptions.get(key);
+ final String subKey = key.substring(PROPERTIES_PREFIX.length());
+ kafkaProperties.put(subKey, value);
+ });
+
+ if (!kafkaProperties.containsKey("bootstrap.servers")) {
+ throw new ValidationException(
+ "'properties.bootstrap.servers' is required for Kafka source.");
+ }
+
+ String groupId = config.get(GROUP_ID);
+ if (groupId == null) {
+ throw new ValidationException("'properties.group.id' is required for Kafka source.");
+ }
+
+ // Validate value format
+ JsonDeserializationType valueFormat = config.get(VALUE_FORMAT);
+ if (valueFormat != JsonDeserializationType.CANAL_JSON) {
+ throw new ValidationException(
+ "Kafka source currently only supports 'canal-json' as value.format, but was: "
+ + valueFormat);
+ }
+
+ // Get startup mode and related options
+ StartupMode startupMode = config.get(SCAN_STARTUP_MODE);
+ validateStartupMode(startupMode, config);
+
+ String specificOffsets = config.get(SCAN_STARTUP_SPECIFIC_OFFSETS);
+ Long startupTimestampMillis = config.get(SCAN_STARTUP_TIMESTAMP_MILLIS);
+
+ // Get schema inference and routing options
+ boolean schemaInferEnabled = config.get(SCHEMA_INFER_ENABLED);
+ String defaultDatabaseName = config.get(DATABASE_NAME);
+ String defaultSchemaName = config.get(SCHEMA_NAME);
+ String defaultTableName = config.get(TABLE_NAME);
+
+ // Create deserialization schema
+ CanalJsonDeserializationSchema deserializationSchema =
+ new CanalJsonDeserializationSchema(
+ schemaInferEnabled,
+ defaultDatabaseName,
+ defaultSchemaName,
+ defaultTableName);
+
+ return new KafkaDataSource(
+ topic,
+ kafkaProperties,
+ groupId,
+ startupMode,
+ specificOffsets,
+ startupTimestampMillis,
+ deserializationSchema);
+ }
+
+ private void validateStartupMode(
+ StartupMode startupMode,
+ org.apache.flink.cdc.common.configuration.Configuration config) {
+ switch (startupMode) {
+ case EARLIEST_OFFSET:
+ case LATEST_OFFSET:
+ case GROUP_OFFSETS:
+ break;
+ case TIMESTAMP:
+ if (config.get(SCAN_STARTUP_TIMESTAMP_MILLIS) == null) {
+ throw new ValidationException(
+ "'scan.startup.timestamp-millis' must be set when 'scan.startup.mode' is 'timestamp'.");
+ }
+ break;
+ case SPECIFIC_OFFSETS:
+ if (config.get(SCAN_STARTUP_SPECIFIC_OFFSETS) == null) {
+ throw new ValidationException(
+ "'scan.startup.specific-offsets' must be set when 'scan.startup.mode' is 'specific-offsets'.");
+ }
+ break;
+ default:
+ throw new ValidationException(
+ String.format(
+ "Invalid value for option '%s'. Supported values are "
+ + "[earliest-offset, latest-offset, group-offsets, timestamp, specific-offsets], "
+ + "but was: %s",
+ SCAN_STARTUP_MODE.key(), startupMode));
+ }
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set> requiredOptions() {
+ return new HashSet<>();
+ }
+
+ @Override
+ public Set> optionalOptions() {
+ Set> options = new HashSet<>();
+ options.add(TOPIC);
+ options.add(GROUP_ID);
+ options.add(VALUE_FORMAT);
+ options.add(SCAN_STARTUP_MODE);
+ options.add(SCAN_STARTUP_SPECIFIC_OFFSETS);
+ options.add(SCAN_STARTUP_TIMESTAMP_MILLIS);
+ options.add(DATABASE_NAME);
+ options.add(SCHEMA_NAME);
+ options.add(TABLE_NAME);
+ options.add(SCHEMA_INFER_ENABLED);
+ return options;
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSourceOptions.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSourceOptions.java
new file mode 100644
index 00000000000..9220c33c876
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSourceOptions.java
@@ -0,0 +1,150 @@
+/*
+ * 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.flink.cdc.connectors.kafka.source;
+
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.configuration.description.Description;
+import org.apache.flink.cdc.connectors.kafka.json.JsonDeserializationType;
+
+import static org.apache.flink.cdc.common.configuration.ConfigOptions.key;
+
+/** Options for {@link KafkaDataSource}. */
+public class KafkaDataSourceOptions {
+
+ // Prefix for Kafka specific properties.
+ public static final String PROPERTIES_PREFIX = "properties.";
+
+ /** The topic to consume from. */
+ public static final ConfigOption TOPIC =
+ key("topic")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Required. Topic name(s) from which the data is read. "
+ + "Multiple topics can be specified as a comma-separated list.");
+
+ /** The bootstrap servers for the Kafka cluster. */
+ public static final ConfigOption BOOTSTRAP_SERVERS =
+ key("properties.bootstrap.servers")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Required. The Kafka bootstrap server list used to establish "
+ + "the initial connection to the Kafka cluster.");
+
+ /** The consumer group id for the Kafka consumer. */
+ public static final ConfigOption GROUP_ID =
+ key("properties.group.id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Required. The consumer group id used to identify which "
+ + "consumer group this source belongs to.");
+
+ /** The format used to deserialize the value from Kafka. */
+ public static final ConfigOption VALUE_FORMAT =
+ key("value.format")
+ .enumType(JsonDeserializationType.class)
+ .defaultValue(JsonDeserializationType.CANAL_JSON)
+ .withDescription(
+ "Defines the format identifier for decoding data from Kafka. "
+ + "Currently only supports 'canal-json'.");
+
+ /** The startup mode for the source. */
+ public static final ConfigOption SCAN_STARTUP_MODE =
+ key("scan.startup.mode")
+ .enumType(StartupMode.class)
+ .defaultValue(StartupMode.LATEST_OFFSET)
+ .withDescription(
+ Description.builder()
+ .text(
+ "Startup mode for Kafka consumer, valid values are "
+ + "'earliest-offset', 'latest-offset', "
+ + "'group-offsets', 'timestamp' and "
+ + "'specific-offsets'.")
+ .build());
+
+ /**
+ * The specific offset to start consuming from. Only used when scan.startup.mode is
+ * specific-offsets.
+ */
+ public static final ConfigOption SCAN_STARTUP_SPECIFIC_OFFSETS =
+ key("scan.startup.specific-offsets")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Specifies offsets for each partition in case the startup mode is "
+ + "'specific-offsets'. "
+ + "Example: 'partition:0,offset:42;partition:1,offset:300'.");
+
+ /**
+ * The timestamp in milliseconds to start consuming from. Only used when scan.startup.mode is
+ * timestamp.
+ */
+ public static final ConfigOption SCAN_STARTUP_TIMESTAMP_MILLIS =
+ key("scan.startup.timestamp-millis")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "Starts reading from the earliest record whose timestamp is greater "
+ + "than or equal to the specified timestamp. "
+ + "Only used when scan.startup.mode is 'timestamp'.");
+
+ /**
+ * The database name to use for table routing. If not set, will use the database from the
+ * message.
+ */
+ public static final ConfigOption DATABASE_NAME =
+ key("database.name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional. The default database name to use when the message does not "
+ + "contain database information.");
+
+ /**
+ * The schema name to use for table routing. If not set, will use the schema from the message.
+ */
+ public static final ConfigOption SCHEMA_NAME =
+ key("schema.name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional. The default schema name to use when the message does not "
+ + "contain schema information.");
+
+ /** The table name to use for table routing. If not set, will use the table from the message. */
+ public static final ConfigOption TABLE_NAME =
+ key("table.name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Optional. The default table name to use when the message does not "
+ + "contain table information. When set, all records will be "
+ + "routed to this table.");
+
+ /** Whether to infer schema from the first message or use STRING for all columns. */
+ public static final ConfigOption SCHEMA_INFER_ENABLED =
+ key("schema.infer.enabled")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Optional. If enabled, the schema will be inferred from the first "
+ + "message. Otherwise, all columns will be treated as STRING. "
+ + "Default is true.");
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/KafkaMetadataAccessor.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/KafkaMetadataAccessor.java
new file mode 100644
index 00000000000..c2016820341
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/KafkaMetadataAccessor.java
@@ -0,0 +1,65 @@
+/*
+ * 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.flink.cdc.connectors.kafka.source;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.source.MetadataAccessor;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * A {@link MetadataAccessor} for Kafka source. Kafka does not maintain table schemas, so all
+ * methods return empty results or throw {@link UnsupportedOperationException} for schema access.
+ */
+@Internal
+public class KafkaMetadataAccessor implements MetadataAccessor {
+
+ private final String topic;
+
+ public KafkaMetadataAccessor(String topic) {
+ this.topic = topic;
+ }
+
+ @Override
+ public List listNamespaces() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public List listSchemas(@Nullable String namespace) {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public List listTables(@Nullable String namespace, @Nullable String schemaName) {
+ // Return the topic as a table
+ return Collections.singletonList(TableId.tableId(topic));
+ }
+
+ @Override
+ public Schema getTableSchema(TableId tableId) {
+ throw new UnsupportedOperationException(
+ "Kafka source does not support accessing table schema from external system. "
+ + "The schema is inferred from the Canal JSON messages.");
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/StartupMode.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/StartupMode.java
new file mode 100644
index 00000000000..93f60eaec36
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/source/StartupMode.java
@@ -0,0 +1,55 @@
+/*
+ * 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.flink.cdc.connectors.kafka.source;
+
+/** Startup mode for Kafka consumer. */
+public enum StartupMode {
+ EARLIEST_OFFSET("earliest-offset"),
+
+ LATEST_OFFSET("latest-offset"),
+
+ GROUP_OFFSETS("group-offsets"),
+
+ TIMESTAMP("timestamp"),
+
+ SPECIFIC_OFFSETS("specific-offsets");
+
+ private final String value;
+
+ StartupMode(String value) {
+ this.value = value;
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ public static StartupMode fromValue(String value) {
+ for (StartupMode mode : values()) {
+ if (mode.value.equalsIgnoreCase(value)) {
+ return mode;
+ }
+ }
+ throw new IllegalArgumentException(
+ "Unsupported startup mode: "
+ + value
+ + ". "
+ + "Supported modes are: earliest-offset, latest-offset, "
+ + "group-offsets, timestamp, specific-offsets");
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
index fa13cd60099..232fd44f61e 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
@@ -14,3 +14,4 @@
# limitations under the License.
org.apache.flink.cdc.connectors.kafka.sink.KafkaDataSinkFactory
+org.apache.flink.cdc.connectors.kafka.source.KafkaDataSourceFactory
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonDeserializationSchemaTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonDeserializationSchemaTest.java
new file mode 100644
index 00000000000..b9e6fadec96
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonDeserializationSchemaTest.java
@@ -0,0 +1,477 @@
+/*
+ * 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.flink.cdc.connectors.kafka.json.canal;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.cdc.common.data.RecordData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.DropColumnEvent;
+import org.apache.flink.cdc.common.event.DropTableEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.OperationType;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.event.TruncateTableEvent;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataTypeRoot;
+import org.apache.flink.util.Collector;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link CanalJsonDeserializationSchema}. */
+class CanalJsonDeserializationSchemaTest {
+
+ private CanalJsonDeserializationSchema deserializationSchema;
+ private TestCollector collector;
+ private ObjectMapper objectMapper;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ deserializationSchema = new CanalJsonDeserializationSchema(true, null, null, null);
+ deserializationSchema.open(createMockContext());
+ collector = new TestCollector();
+ objectMapper = new ObjectMapper();
+ }
+
+ @Test
+ void testInsertEventParsing() throws IOException {
+ String json =
+ "{\"old\":null,\"data\":[{\"id\":1,\"name\":\"Alice\",\"age\":30}],"
+ + "\"type\":\"INSERT\",\"database\":\"mydb\",\"table\":\"users\","
+ + "\"pkNames\":[\"id\"]}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ List events = collector.getEvents();
+ assertThat(events).hasSize(2);
+
+ // First event should be CreateTableEvent
+ assertThat(events.get(0)).isInstanceOf(CreateTableEvent.class);
+ CreateTableEvent createTableEvent = (CreateTableEvent) events.get(0);
+ assertThat(createTableEvent.tableId()).isEqualTo(TableId.tableId("mydb", "users"));
+ Schema schema = createTableEvent.getSchema();
+ assertThat(schema.getColumnNames()).containsExactly("id", "name", "age");
+ assertThat(schema.primaryKeys()).containsExactly("id");
+
+ // Second event should be DataChangeEvent with INSERT
+ assertThat(events.get(1)).isInstanceOf(DataChangeEvent.class);
+ DataChangeEvent dataChangeEvent = (DataChangeEvent) events.get(1);
+ assertThat(dataChangeEvent.op()).isEqualTo(OperationType.INSERT);
+ assertThat(dataChangeEvent.before()).isNull();
+ assertThat(dataChangeEvent.after()).isNotNull();
+ }
+
+ @Test
+ void testUpdateEventParsing() throws IOException {
+ String json =
+ "{\"old\":[{\"id\":1,\"name\":\"Alice\",\"age\":25}],"
+ + "\"data\":[{\"id\":1,\"name\":\"Alice\",\"age\":26}],"
+ + "\"type\":\"UPDATE\",\"database\":\"mydb\",\"table\":\"users\","
+ + "\"pkNames\":[\"id\"]}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ List events = collector.getEvents();
+ assertThat(events).hasSize(2);
+
+ DataChangeEvent dataChangeEvent = (DataChangeEvent) events.get(1);
+ assertThat(dataChangeEvent.op()).isEqualTo(OperationType.UPDATE);
+ assertThat(dataChangeEvent.before()).isNotNull();
+ assertThat(dataChangeEvent.after()).isNotNull();
+ }
+
+ @Test
+ void testDeleteEventParsing() throws IOException {
+ String json =
+ "{\"old\":null,\"data\":[{\"id\":1,\"name\":\"Alice\",\"age\":30}],"
+ + "\"type\":\"DELETE\",\"database\":\"mydb\",\"table\":\"users\","
+ + "\"pkNames\":[\"id\"]}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ List events = collector.getEvents();
+ assertThat(events).hasSize(2);
+
+ DataChangeEvent dataChangeEvent = (DataChangeEvent) events.get(1);
+ assertThat(dataChangeEvent.op()).isEqualTo(OperationType.DELETE);
+ assertThat(dataChangeEvent.after()).isNull();
+ assertThat(dataChangeEvent.before()).isNotNull();
+ }
+
+ @Test
+ void testSchemaInferenceDisabled() throws Exception {
+ CanalJsonDeserializationSchema noInferSchema =
+ new CanalJsonDeserializationSchema(false, null, null, null);
+ noInferSchema.open(createMockContext());
+
+ String json =
+ "{\"old\":null,\"data\":[{\"id\":1,\"name\":\"Alice\"}],"
+ + "\"type\":\"INSERT\",\"database\":\"mydb\",\"table\":\"users\","
+ + "\"pkNames\":[\"id\"]}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ TestCollector testCollector = new TestCollector();
+ noInferSchema.deserialize(record, testCollector);
+
+ CreateTableEvent createTableEvent = (CreateTableEvent) testCollector.getEvents().get(0);
+ Schema schema = createTableEvent.getSchema();
+ // All columns should be STRING type when schema inference is disabled
+ assertThat(schema.getColumns())
+ .allMatch(col -> col.getType().getTypeRoot() == DataTypeRoot.VARCHAR);
+ }
+
+ @Test
+ void testNullValueSkipped() throws IOException {
+ ConsumerRecord record =
+ new ConsumerRecord<>("test-topic", 0, 0L, null, null);
+ deserializationSchema.deserialize(record, collector);
+ assertThat(collector.getEvents()).isEmpty();
+ }
+
+ @Test
+ void testDefaultTableName() throws Exception {
+ CanalJsonDeserializationSchema schemaWithDefault =
+ new CanalJsonDeserializationSchema(true, "defaultdb", null, "default_table");
+ schemaWithDefault.open(createMockContext());
+
+ // Canal JSON without table field
+ String json =
+ "{\"old\":null,\"data\":[{\"id\":1,\"name\":\"Alice\"}],"
+ + "\"type\":\"INSERT\",\"database\":\"mydb\","
+ + "\"pkNames\":[\"id\"]}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ TestCollector testCollector = new TestCollector();
+ schemaWithDefault.deserialize(record, testCollector);
+
+ CreateTableEvent createTableEvent = (CreateTableEvent) testCollector.getEvents().get(0);
+ assertThat(createTableEvent.tableId().getTableName()).isEqualTo("default_table");
+ assertThat(createTableEvent.tableId().getSchemaName()).isEqualTo("mydb");
+ }
+
+ @Test
+ void testMultipleInsertsInOneMessage() throws IOException {
+ String json =
+ "{\"old\":null,\"data\":["
+ + "{\"id\":1,\"name\":\"Alice\"},"
+ + "{\"id\":2,\"name\":\"Bob\"}"
+ + "],\"type\":\"INSERT\",\"database\":\"mydb\",\"table\":\"users\","
+ + "\"pkNames\":[\"id\"]}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ // 1 CreateTableEvent + 2 DataChangeEvents
+ assertThat(collector.getEvents()).hasSize(3);
+ assertThat(collector.getEvents().get(1)).isInstanceOf(DataChangeEvent.class);
+ assertThat(collector.getEvents().get(2)).isInstanceOf(DataChangeEvent.class);
+ }
+
+ @Test
+ void testCreateTableEventOnlyEmittedOnce() throws IOException {
+ String json1 =
+ "{\"old\":null,\"data\":[{\"id\":1,\"name\":\"Alice\"}],"
+ + "\"type\":\"INSERT\",\"database\":\"mydb\",\"table\":\"users\","
+ + "\"pkNames\":[\"id\"]}";
+ String json2 =
+ "{\"old\":null,\"data\":[{\"id\":2,\"name\":\"Bob\"}],"
+ + "\"type\":\"INSERT\",\"database\":\"mydb\",\"table\":\"users\","
+ + "\"pkNames\":[\"id\"]}";
+
+ ConsumerRecord record1 =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json1.getBytes(StandardCharsets.UTF_8));
+ ConsumerRecord record2 =
+ new ConsumerRecord<>(
+ "test-topic", 0, 1L, null, json2.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record1, collector);
+ deserializationSchema.deserialize(record2, collector);
+
+ // First message: 1 CreateTableEvent + 1 DataChangeEvent = 2
+ // Second message: 1 DataChangeEvent = 1
+ // Total: 3
+ List events = collector.getEvents();
+ assertThat(events).hasSize(3);
+ assertThat(events.get(0)).isInstanceOf(CreateTableEvent.class);
+ assertThat(events.get(1)).isInstanceOf(DataChangeEvent.class);
+ assertThat(events.get(2)).isInstanceOf(DataChangeEvent.class);
+ }
+
+ @Test
+ void testDataValueExtraction() throws IOException {
+ String json =
+ "{\"old\":null,\"data\":[{\"id\":42,\"name\":\"Charlie\",\"active\":true}],"
+ + "\"type\":\"INSERT\",\"database\":\"mydb\",\"table\":\"users\","
+ + "\"pkNames\":[\"id\"]}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ DataChangeEvent dataChangeEvent = (DataChangeEvent) collector.getEvents().get(1);
+ RecordData after = dataChangeEvent.after();
+ assertThat(after.getInt(0)).isEqualTo(42);
+ assertThat(after.getString(1).toString()).isEqualTo("Charlie");
+ assertThat(after.getBoolean(2)).isTrue();
+ }
+
+ @Test
+ void testCreateTableDdl() throws IOException {
+ String json =
+ "{\"isDdl\":true,\"sql\":\"CREATE TABLE products (id INT PRIMARY KEY, name VARCHAR(100), price DECIMAL(10,2))\","
+ + "\"type\":\"CREATE\",\"database\":\"mydb\",\"table\":\"products\"}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ List events = collector.getEvents();
+ assertThat(events).hasSize(1);
+ assertThat(events.get(0)).isInstanceOf(CreateTableEvent.class);
+ CreateTableEvent createTableEvent = (CreateTableEvent) events.get(0);
+ assertThat(createTableEvent.tableId()).isEqualTo(TableId.tableId("mydb", "products"));
+ Schema schema = createTableEvent.getSchema();
+ assertThat(schema.getColumnNames()).containsExactly("id", "name", "price");
+ assertThat(schema.primaryKeys()).containsExactly("id");
+ }
+
+ @Test
+ void testDropTableDdl() throws IOException {
+ String json =
+ "{\"isDdl\":true,\"sql\":\"DROP TABLE users\",\"type\":\"DROP\","
+ + "\"database\":\"mydb\",\"table\":\"users\"}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ List events = collector.getEvents();
+ assertThat(events).hasSize(1);
+ assertThat(events.get(0)).isInstanceOf(DropTableEvent.class);
+ DropTableEvent dropTableEvent = (DropTableEvent) events.get(0);
+ assertThat(dropTableEvent.tableId()).isEqualTo(TableId.tableId("mydb", "users"));
+ }
+
+ @Test
+ void testTruncateTableDdl() throws IOException {
+ String json =
+ "{\"isDdl\":true,\"sql\":\"TRUNCATE TABLE orders\",\"type\":\"TRUNCATE\","
+ + "\"database\":\"mydb\",\"table\":\"orders\"}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ List events = collector.getEvents();
+ assertThat(events).hasSize(1);
+ assertThat(events.get(0)).isInstanceOf(TruncateTableEvent.class);
+ TruncateTableEvent truncateTableEvent = (TruncateTableEvent) events.get(0);
+ assertThat(truncateTableEvent.tableId()).isEqualTo(TableId.tableId("mydb", "orders"));
+ }
+
+ @Test
+ void testAlterAddColumnDdl() throws IOException {
+ String json =
+ "{\"isDdl\":true,\"sql\":\"ALTER TABLE users ADD COLUMN email VARCHAR(255)\","
+ + "\"type\":\"ALTER\",\"database\":\"mydb\",\"table\":\"users\"}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ List events = collector.getEvents();
+ assertThat(events).hasSize(1);
+ assertThat(events.get(0)).isInstanceOf(AddColumnEvent.class);
+ AddColumnEvent addColumnEvent = (AddColumnEvent) events.get(0);
+ assertThat(addColumnEvent.tableId()).isEqualTo(TableId.tableId("mydb", "users"));
+ }
+
+ @Test
+ void testAlterDropColumnDdl() throws IOException {
+ String json =
+ "{\"isDdl\":true,\"sql\":\"ALTER TABLE users DROP COLUMN email\","
+ + "\"type\":\"ALTER\",\"database\":\"mydb\",\"table\":\"users\"}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ List events = collector.getEvents();
+ assertThat(events).hasSize(1);
+ assertThat(events.get(0)).isInstanceOf(DropColumnEvent.class);
+ DropColumnEvent dropColumnEvent = (DropColumnEvent) events.get(0);
+ assertThat(dropColumnEvent.tableId()).isEqualTo(TableId.tableId("mydb", "users"));
+ }
+
+ @Test
+ void testAlterModifyColumnDdl() throws IOException {
+ String json =
+ "{\"isDdl\":true,\"sql\":\"ALTER TABLE users MODIFY COLUMN age BIGINT\","
+ + "\"type\":\"ALTER\",\"database\":\"mydb\",\"table\":\"users\"}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ List events = collector.getEvents();
+ assertThat(events).hasSize(1);
+ assertThat(events.get(0)).isInstanceOf(AlterColumnTypeEvent.class);
+ AlterColumnTypeEvent alterColumnTypeEvent = (AlterColumnTypeEvent) events.get(0);
+ assertThat(alterColumnTypeEvent.tableId()).isEqualTo(TableId.tableId("mydb", "users"));
+ }
+
+ @Test
+ void testDdlWithoutIsDdlFlag() throws IOException {
+ String json =
+ "{\"isDdl\":true,\"sql\":\"CREATE TABLE test_table (id INT)\","
+ + "\"type\":\"CREATE TABLE\",\"database\":\"mydb\",\"table\":\"test_table\"}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ List events = collector.getEvents();
+ assertThat(events).hasSize(1);
+ assertThat(events.get(0)).isInstanceOf(CreateTableEvent.class);
+ }
+
+ @Test
+ void testDdlWithoutTypeField() throws IOException {
+ String json =
+ "{\"isDdl\":true,\"sql\":\"CREATE TABLE products (id INT PRIMARY KEY, name VARCHAR(100))\","
+ + "\"database\":\"mydb\",\"table\":\"products\"}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ List events = collector.getEvents();
+ assertThat(events).hasSize(1);
+ assertThat(events.get(0)).isInstanceOf(CreateTableEvent.class);
+ CreateTableEvent createTableEvent = (CreateTableEvent) events.get(0);
+ assertThat(createTableEvent.tableId()).isEqualTo(TableId.tableId("mydb", "products"));
+ Schema schema = createTableEvent.getSchema();
+ assertThat(schema.getColumnNames()).containsExactly("id", "name");
+ assertThat(schema.primaryKeys()).containsExactly("id");
+ }
+
+ @Test
+ void testDdlWithQueryType() throws IOException {
+ String json =
+ "{\"isDdl\":true,\"sql\":\"ALTER TABLE users ADD COLUMN email VARCHAR(255)\","
+ + "\"type\":\"QUERY\",\"database\":\"mydb\",\"table\":\"users\"}";
+
+ ConsumerRecord record =
+ new ConsumerRecord<>(
+ "test-topic", 0, 0L, null, json.getBytes(StandardCharsets.UTF_8));
+
+ deserializationSchema.deserialize(record, collector);
+
+ List events = collector.getEvents();
+ assertThat(events).hasSize(1);
+ assertThat(events.get(0)).isInstanceOf(AddColumnEvent.class);
+ AddColumnEvent addColumnEvent = (AddColumnEvent) events.get(0);
+ assertThat(addColumnEvent.tableId()).isEqualTo(TableId.tableId("mydb", "users"));
+ }
+
+ private static DeserializationSchema.InitializationContext createMockContext() {
+ return new DeserializationSchema.InitializationContext() {
+ @Override
+ public org.apache.flink.metrics.MetricGroup getMetricGroup() {
+ return null;
+ }
+
+ @Override
+ public org.apache.flink.util.UserCodeClassLoader getUserCodeClassLoader() {
+ return null;
+ }
+ };
+ }
+
+ /** A test collector that collects emitted events into a list. */
+ private static class TestCollector implements Collector {
+ private final List events = new ArrayList<>();
+
+ @Override
+ public void collect(Event record) {
+ events.add(record);
+ }
+
+ @Override
+ public void close() {
+ // no-op
+ }
+
+ public List getEvents() {
+ return events;
+ }
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSourceFactoryTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSourceFactoryTest.java
new file mode 100644
index 00000000000..06e5a277406
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSourceFactoryTest.java
@@ -0,0 +1,266 @@
+/*
+ * 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.flink.cdc.connectors.kafka.source;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.factories.DataSourceFactory;
+import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.source.DataSource;
+import org.apache.flink.cdc.composer.utils.FactoryDiscoveryUtils;
+import org.apache.flink.table.api.ValidationException;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Tests for {@link KafkaDataSourceFactory}. */
+class KafkaDataSourceFactoryTest {
+
+ private static final String BOOTSTRAP_SERVERS = "localhost:9092";
+ private static final String GROUP_ID = "test-group";
+ private static final String TOPIC = "test-topic";
+
+ @Test
+ void testCreateDataSource() {
+ DataSourceFactory sourceFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("kafka", DataSourceFactory.class);
+ Assertions.assertThat(sourceFactory).isInstanceOf(KafkaDataSourceFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.builder()
+ .put(KafkaDataSourceOptions.TOPIC.key(), TOPIC)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX
+ + "bootstrap.servers",
+ BOOTSTRAP_SERVERS)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX + "group.id",
+ GROUP_ID)
+ .build());
+
+ DataSource dataSource =
+ sourceFactory.createDataSource(
+ new FactoryHelper.DefaultContext(
+ conf, conf, Thread.currentThread().getContextClassLoader()));
+ Assertions.assertThat(dataSource).isInstanceOf(KafkaDataSource.class);
+ }
+
+ @Test
+ void testMissingTopicThrowsException() {
+ DataSourceFactory sourceFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("kafka", DataSourceFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.builder()
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX
+ + "bootstrap.servers",
+ BOOTSTRAP_SERVERS)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX + "group.id",
+ GROUP_ID)
+ .build());
+
+ Assertions.assertThatThrownBy(
+ () ->
+ sourceFactory.createDataSource(
+ new FactoryHelper.DefaultContext(
+ conf,
+ conf,
+ Thread.currentThread().getContextClassLoader())))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("'topic' is required");
+ }
+
+ @Test
+ void testMissingBootstrapServersThrowsException() {
+ DataSourceFactory sourceFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("kafka", DataSourceFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.builder()
+ .put(KafkaDataSourceOptions.TOPIC.key(), TOPIC)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX + "group.id",
+ GROUP_ID)
+ .build());
+
+ Assertions.assertThatThrownBy(
+ () ->
+ sourceFactory.createDataSource(
+ new FactoryHelper.DefaultContext(
+ conf,
+ conf,
+ Thread.currentThread().getContextClassLoader())))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("'properties.bootstrap.servers' is required");
+ }
+
+ @Test
+ void testUnsupportedValueFormat() {
+ DataSourceFactory sourceFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("kafka", DataSourceFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.builder()
+ .put(KafkaDataSourceOptions.TOPIC.key(), TOPIC)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX
+ + "bootstrap.servers",
+ BOOTSTRAP_SERVERS)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX + "group.id",
+ GROUP_ID)
+ .put(KafkaDataSourceOptions.VALUE_FORMAT.key(), "debezium-json")
+ .build());
+
+ Assertions.assertThatThrownBy(
+ () ->
+ sourceFactory.createDataSource(
+ new FactoryHelper.DefaultContext(
+ conf,
+ conf,
+ Thread.currentThread().getContextClassLoader())))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Could not parse value 'debezium-json'");
+ }
+
+ @Test
+ void testInvalidStartupMode() {
+ DataSourceFactory sourceFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("kafka", DataSourceFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.builder()
+ .put(KafkaDataSourceOptions.TOPIC.key(), TOPIC)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX
+ + "bootstrap.servers",
+ BOOTSTRAP_SERVERS)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX + "group.id",
+ GROUP_ID)
+ .put(KafkaDataSourceOptions.SCAN_STARTUP_MODE.key(), "invalid-mode")
+ .build());
+
+ Assertions.assertThatThrownBy(
+ () ->
+ sourceFactory.createDataSource(
+ new FactoryHelper.DefaultContext(
+ conf,
+ conf,
+ Thread.currentThread().getContextClassLoader())))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Could not parse value 'invalid-mode'");
+ }
+
+ @Test
+ void testTimestampStartupModeWithoutTimestamp() {
+ DataSourceFactory sourceFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("kafka", DataSourceFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.builder()
+ .put(KafkaDataSourceOptions.TOPIC.key(), TOPIC)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX
+ + "bootstrap.servers",
+ BOOTSTRAP_SERVERS)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX + "group.id",
+ GROUP_ID)
+ .put(KafkaDataSourceOptions.SCAN_STARTUP_MODE.key(), "timestamp")
+ .build());
+
+ Assertions.assertThatThrownBy(
+ () ->
+ sourceFactory.createDataSource(
+ new FactoryHelper.DefaultContext(
+ conf,
+ conf,
+ Thread.currentThread().getContextClassLoader())))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("scan.startup.timestamp-millis");
+ }
+
+ @Test
+ void testUnsupportedOption() {
+ DataSourceFactory sourceFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("kafka", DataSourceFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.builder()
+ .put(KafkaDataSourceOptions.TOPIC.key(), TOPIC)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX
+ + "bootstrap.servers",
+ BOOTSTRAP_SERVERS)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX + "group.id",
+ GROUP_ID)
+ .put("unsupported_key", "unsupported_value")
+ .build());
+
+ Assertions.assertThatThrownBy(
+ () ->
+ sourceFactory.createDataSource(
+ new FactoryHelper.DefaultContext(
+ conf,
+ conf,
+ Thread.currentThread().getContextClassLoader())))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Unsupported options found for 'kafka'");
+ }
+
+ @Test
+ void testPropertiesPrefixOption() {
+ DataSourceFactory sourceFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("kafka", DataSourceFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.builder()
+ .put(KafkaDataSourceOptions.TOPIC.key(), TOPIC)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX
+ + "bootstrap.servers",
+ BOOTSTRAP_SERVERS)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX + "group.id",
+ GROUP_ID)
+ .put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX
+ + "auto.offset.reset",
+ "earliest")
+ .build());
+
+ DataSource dataSource =
+ sourceFactory.createDataSource(
+ new FactoryHelper.DefaultContext(
+ conf, conf, Thread.currentThread().getContextClassLoader()));
+ Assertions.assertThat(dataSource).isInstanceOf(KafkaDataSource.class);
+ }
+}
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSourceITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSourceITCase.java
new file mode 100644
index 00000000000..a39ec7d1023
--- /dev/null
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/source/KafkaDataSourceITCase.java
@@ -0,0 +1,199 @@
+/*
+ * 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.flink.cdc.connectors.kafka.source;
+
+import org.apache.flink.cdc.common.source.FlinkSourceProvider;
+import org.apache.flink.cdc.connectors.kafka.json.JsonDeserializationType;
+import org.apache.flink.cdc.connectors.kafka.json.canal.CanalJsonDeserializationSchema;
+
+import org.apache.flink.cdc.connectors.kafka.sink.KafkaUtil;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.Network;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.flink.util.DockerImageVersions.KAFKA;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Integration tests for {@link KafkaDataSource} reading Canal JSON from Kafka. */
+@Timeout(value = 120, unit = TimeUnit.SECONDS)
+class KafkaDataSourceITCase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaDataSourceITCase.class);
+ private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
+ private static final Network NETWORK = Network.newNetwork();
+
+ public static final KafkaContainer KAFKA_CONTAINER =
+ KafkaUtil.createKafkaContainer(KAFKA, LOG)
+ .withEmbeddedZookeeper()
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+
+ @BeforeAll
+ static void setup() {
+ KAFKA_CONTAINER.start();
+ }
+
+ @AfterAll
+ static void teardown() {
+ KAFKA_CONTAINER.stop();
+ }
+
+ @Test
+ void testReadCanalJsonFromKafka() throws Exception {
+ String topic = "test-canal-json-" + UUID.randomUUID();
+ String groupId = "test-group-" + UUID.randomUUID();
+ String database = "mydb";
+ String table = "users";
+
+ // Produce test messages to Kafka
+ produceCanalJsonMessages(topic, database, table);
+
+ // Create KafkaDataSource
+ CanalJsonDeserializationSchema deserializationSchema =
+ new CanalJsonDeserializationSchema(true, null, null, null);
+
+ Properties kafkaProperties = new Properties();
+ kafkaProperties.put("bootstrap.servers", KAFKA_CONTAINER.getBootstrapServers());
+
+ KafkaDataSource dataSource =
+ new KafkaDataSource(
+ topic,
+ kafkaProperties,
+ groupId,
+ StartupMode.EARLIEST_OFFSET,
+ null,
+ null,
+ deserializationSchema);
+
+ // Verify the source can be created
+ assertThat(dataSource).isNotNull();
+ assertThat(dataSource.getEventSourceProvider()).isInstanceOf(FlinkSourceProvider.class);
+ assertThat(dataSource.getMetadataAccessor()).isNotNull();
+ assertThat(dataSource.isParallelMetadataSource()).isTrue();
+ }
+
+ @Test
+ void testKafkaDataSourceWithDefaultTableName() {
+ String topic = "test-default-table-" + UUID.randomUUID();
+ String groupId = "test-group-" + UUID.randomUUID();
+
+ CanalJsonDeserializationSchema deserializationSchema =
+ new CanalJsonDeserializationSchema(true, "defaultdb", null, "mytable");
+
+ Properties kafkaProperties = new Properties();
+ kafkaProperties.put("bootstrap.servers", KAFKA_CONTAINER.getBootstrapServers());
+
+ KafkaDataSource dataSource =
+ new KafkaDataSource(
+ topic,
+ kafkaProperties,
+ groupId,
+ StartupMode.LATEST_OFFSET,
+ null,
+ null,
+ deserializationSchema);
+
+ assertThat(dataSource).isNotNull();
+ }
+
+ @Test
+ void testDataSourceFactoryCreation() {
+ String topic = "test-factory-" + UUID.randomUUID();
+ String groupId = "test-group-" + UUID.randomUUID();
+
+ Map config = new HashMap<>();
+ config.put(KafkaDataSourceOptions.TOPIC.key(), topic);
+ config.put(
+ KafkaDataSourceOptions.PROPERTIES_PREFIX + "bootstrap.servers",
+ KAFKA_CONTAINER.getBootstrapServers());
+ config.put(KafkaDataSourceOptions.PROPERTIES_PREFIX + "group.id", groupId);
+ config.put(KafkaDataSourceOptions.VALUE_FORMAT.key(), JsonDeserializationType.CANAL_JSON.getValue());
+ config.put(KafkaDataSourceOptions.SCAN_STARTUP_MODE.key(), "EARLIEST_OFFSET");
+
+ KafkaDataSourceFactory factory = new KafkaDataSourceFactory();
+ KafkaDataSource dataSource =
+ (KafkaDataSource)
+ factory.createDataSource(
+ new org.apache.flink.cdc.common.factories.FactoryHelper
+ .DefaultContext(
+ org.apache.flink.cdc.common.configuration.Configuration
+ .fromMap(config),
+ org.apache.flink.cdc.common.configuration.Configuration
+ .fromMap(new HashMap<>()),
+ Thread.currentThread().getContextClassLoader()));
+
+ assertThat(dataSource).isNotNull();
+ }
+
+ private void produceCanalJsonMessages(String topic, String database, String table) {
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_CONTAINER.getBootstrapServers());
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ props.put(
+ ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+
+ try (KafkaProducer producer = new KafkaProducer<>(props)) {
+ // Insert
+ String insertJson =
+ String.format(
+ "{\"old\":null,\"data\":[{\"id\":1,\"name\":\"Alice\",\"age\":30}],"
+ + "\"type\":\"INSERT\",\"database\":\"%s\",\"table\":\"%s\","
+ + "\"pkNames\":[\"id\"]}",
+ database, table);
+ producer.send(new ProducerRecord<>(topic, insertJson.getBytes(StandardCharsets.UTF_8)));
+
+ // Update
+ String updateJson =
+ String.format(
+ "{\"old\":[{\"id\":1,\"name\":\"Alice\",\"age\":30}],"
+ + "\"data\":[{\"id\":1,\"name\":\"Alice\",\"age\":31}],"
+ + "\"type\":\"UPDATE\",\"database\":\"%s\",\"table\":\"%s\","
+ + "\"pkNames\":[\"id\"]}",
+ database, table);
+ producer.send(new ProducerRecord<>(topic, updateJson.getBytes(StandardCharsets.UTF_8)));
+
+ // Delete
+ String deleteJson =
+ String.format(
+ "{\"old\":null,\"data\":[{\"id\":1,\"name\":\"Alice\",\"age\":31}],"
+ + "\"type\":\"DELETE\",\"database\":\"%s\",\"table\":\"%s\","
+ + "\"pkNames\":[\"id\"]}",
+ database, table);
+ producer.send(new ProducerRecord<>(topic, deleteJson.getBytes(StandardCharsets.UTF_8)));
+
+ producer.flush();
+ }
+ }
+}
diff --git a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/KafkaE2eITCase.java b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/KafkaE2eITCase.java
new file mode 100644
index 00000000000..e525193a271
--- /dev/null
+++ b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/KafkaE2eITCase.java
@@ -0,0 +1,262 @@
+/*
+ * 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.flink.cdc.pipeline.tests;
+
+import org.apache.flink.cdc.common.test.utils.TestUtils;
+import org.apache.flink.cdc.connectors.kafka.sink.KafkaUtil;
+import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.lifecycle.Startables;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Stream;
+
+import static org.apache.flink.util.DockerImageVersions.KAFKA;
+
+/** End-to-end tests for Kafka source to values sink pipeline job. */
+class KafkaE2eITCase extends PipelineTestEnvironment {
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaE2eITCase.class);
+
+ private static AdminClient admin;
+ private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
+ private static final short TOPIC_REPLICATION_FACTOR = 1;
+ private String topic;
+
+ @Container
+ static final KafkaContainer KAFKA_CONTAINER =
+ KafkaUtil.createKafkaContainer(KAFKA, LOG)
+ .withNetworkAliases("kafka")
+ .withEmbeddedZookeeper()
+ .withNetwork(NETWORK)
+ .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+
+ @BeforeAll
+ static void initializeContainers() {
+ LOG.info("Starting containers...");
+ Startables.deepStart(Stream.of(KAFKA_CONTAINER)).join();
+ Map properties = new HashMap<>();
+ properties.put(
+ CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
+ KAFKA_CONTAINER.getBootstrapServers());
+ admin = AdminClient.create(properties);
+ LOG.info("Containers are started.");
+ }
+
+ @BeforeEach
+ public void before() throws Exception {
+ super.before();
+ createTestTopic(1, TOPIC_REPLICATION_FACTOR);
+ }
+
+ @AfterEach
+ public void after() {
+ super.after();
+ admin.deleteTopics(Collections.singletonList(topic));
+ }
+
+ @Test
+ void testSyncCanalJsonFromKafka() throws Exception {
+ String groupId = "kafka-source-test-" + UUID.randomUUID();
+ String pipelineJob = buildPipeline(topic, groupId);
+ Path kafkaCdcJar = TestUtils.getResource("kafka-cdc-pipeline-connector.jar");
+ submitPipelineJob(pipelineJob, kafkaCdcJar);
+ waitUntilJobRunning(Duration.ofSeconds(30));
+ LOG.info("Pipeline job is running");
+
+ produceDmlMessages();
+
+ // Verify the CreateTableEvent is emitted with inferred schema and primary key
+ waitUntilSpecificEvent(
+ "CreateTableEvent{tableId=flink-test.kafka_source_test, "
+ + "schema=columns={`id` INT,`name` STRING,`description` STRING}, "
+ + "primaryKeys=id, options=()}");
+
+ // Verify INSERT events
+ waitUntilSpecificEvent(
+ "DataChangeEvent{tableId=flink-test.kafka_source_test, "
+ + "before=[], after=[1, test1, description1], op=INSERT, meta=()}");
+ waitUntilSpecificEvent(
+ "DataChangeEvent{tableId=flink-test.kafka_source_test, "
+ + "before=[], after=[2, test2, description2], op=INSERT, meta=()}");
+
+ // Verify UPDATE event
+ waitUntilSpecificEvent(
+ "DataChangeEvent{tableId=flink-test.kafka_source_test, "
+ + "before=[1, test1, description1], "
+ + "after=[1, test1_updated, description1_updated], op=UPDATE, meta=()}");
+
+ // Verify DELETE event
+ waitUntilSpecificEvent(
+ "DataChangeEvent{tableId=flink-test.kafka_source_test, "
+ + "before=[2, test2, description2], after=[], op=DELETE, meta=()}");
+ }
+
+ @Test
+ void testSyncCanalDdlFromKafka() throws Exception {
+ String groupId = "kafka-ddl-test-" + UUID.randomUUID();
+ String pipelineJob = buildPipeline(topic, groupId);
+ Path kafkaCdcJar = TestUtils.getResource("kafka-cdc-pipeline-connector.jar");
+ submitPipelineJob(pipelineJob, kafkaCdcJar);
+ waitUntilJobRunning(Duration.ofSeconds(30));
+ LOG.info("Pipeline job is running");
+
+ produceDdlMessages();
+
+ // Verify the CreateTableEvent from DDL message
+ waitUntilSpecificEvent(
+ "CreateTableEvent{tableId=flink-test.users, "
+ + "schema=columns={`id` INT,`name` VARCHAR(100),`age` INT}, "
+ + "primaryKeys=id, options=()}");
+
+ // Verify the AddColumnEvent from ALTER TABLE ADD COLUMN.
+ // Note: Under LENIENT mode, DropColumnEvent would be filtered (or converted to
+ // AlterColumnTypeEvent when the column is NOT NULL), so we don't verify it here.
+ waitUntilSpecificEvent(
+ "AddColumnEvent{tableId=flink-test.users, addedColumns="
+ + "[ColumnWithPosition{column=`email` VARCHAR(255), "
+ + "position=LAST, existedColumnName=null}]}");
+ }
+
+ private String buildPipeline(String topic, String groupId) {
+ return String.format(
+ "source:\n"
+ + " type: kafka\n"
+ + " topic: %s\n"
+ + " properties.bootstrap.servers: kafka:9092\n"
+ + " properties.group.id: %s\n"
+ + " value.format: canal-json\n"
+ + "\n"
+ + "sink:\n"
+ + " type: values\n"
+ + "\n"
+ + "pipeline:\n"
+ + " parallelism: %d\n"
+ + " schema.change.behavior: lenient",
+ topic, groupId, parallelism);
+ }
+
+ /** Produce DML messages (INSERT / UPDATE / DELETE) with primary key. */
+ private void produceDmlMessages() throws Exception {
+ try (KafkaProducer producer = buildProducer()) {
+ // INSERT id=1 (with pkNames)
+ String insertMessage1 =
+ "{\"data\":[{\"id\":1,\"name\":\"test1\",\"description\":\"description1\"}],"
+ + "\"database\":\"flink-test\",\"table\":\"kafka_source_test\","
+ + "\"type\":\"INSERT\",\"pkNames\":[\"id\"],\"isDdl\":false}";
+ producer.send(record(insertMessage1)).get();
+
+ // INSERT id=2 (with pkNames)
+ String insertMessage2 =
+ "{\"data\":[{\"id\":2,\"name\":\"test2\",\"description\":\"description2\"}],"
+ + "\"database\":\"flink-test\",\"table\":\"kafka_source_test\","
+ + "\"type\":\"INSERT\",\"pkNames\":[\"id\"],\"isDdl\":false}";
+ producer.send(record(insertMessage2)).get();
+
+ // UPDATE id=1 (with pkNames)
+ String updateMessage =
+ "{\"data\":[{\"id\":1,\"name\":\"test1_updated\",\"description\":\"description1_updated\"}],"
+ + "\"old\":[{\"id\":1,\"name\":\"test1\",\"description\":\"description1\"}],"
+ + "\"database\":\"flink-test\",\"table\":\"kafka_source_test\","
+ + "\"type\":\"UPDATE\",\"pkNames\":[\"id\"],\"isDdl\":false}";
+ producer.send(record(updateMessage)).get();
+
+ // DELETE id=2 (with pkNames)
+ String deleteMessage =
+ "{\"data\":[{\"id\":2,\"name\":\"test2\",\"description\":\"description2\"}],"
+ + "\"database\":\"flink-test\",\"table\":\"kafka_source_test\","
+ + "\"type\":\"DELETE\",\"pkNames\":[\"id\"],\"isDdl\":false}";
+ producer.send(record(deleteMessage)).get();
+ }
+ LOG.info("DML Canal JSON messages produced to topic: {}", topic);
+ }
+
+ /** Produce DDL messages (CREATE TABLE / ALTER TABLE ADD COLUMN / ALTER TABLE DROP COLUMN). */
+ private void produceDdlMessages() throws Exception {
+ try (KafkaProducer producer = buildProducer()) {
+ // CREATE TABLE with primary key
+ String createTableDdl =
+ "{\"isDdl\":true,"
+ + "\"sql\":\"CREATE TABLE users (id INT PRIMARY KEY, name VARCHAR(100) NOT NULL, age INT)\","
+ + "\"type\":\"CREATE\","
+ + "\"database\":\"flink-test\",\"table\":\"users\"}";
+ producer.send(record(createTableDdl)).get();
+
+ // ALTER TABLE ADD COLUMN
+ String addColumnDdl =
+ "{\"isDdl\":true,"
+ + "\"sql\":\"ALTER TABLE users ADD COLUMN email VARCHAR(255) AFTER age\","
+ + "\"type\":\"ALTER\","
+ + "\"database\":\"flink-test\",\"table\":\"users\"}";
+ producer.send(record(addColumnDdl)).get();
+
+ // ALTER TABLE DROP COLUMN
+ String dropColumnDdl =
+ "{\"isDdl\":true,"
+ + "\"sql\":\"ALTER TABLE users DROP COLUMN age\","
+ + "\"type\":\"ALTER\","
+ + "\"database\":\"flink-test\",\"table\":\"users\"}";
+ producer.send(record(dropColumnDdl)).get();
+ }
+ LOG.info("DDL Canal JSON messages produced to topic: {}", topic);
+ }
+
+ private KafkaProducer buildProducer() {
+ Properties producerProps = new Properties();
+ producerProps.put("bootstrap.servers", KAFKA_CONTAINER.getBootstrapServers());
+ producerProps.put("key.serializer", ByteArraySerializer.class.getName());
+ producerProps.put("value.serializer", ByteArraySerializer.class.getName());
+ return new KafkaProducer<>(producerProps);
+ }
+
+ private ProducerRecord record(String message) {
+ return new ProducerRecord<>(topic, message.getBytes(StandardCharsets.UTF_8));
+ }
+
+ private void createTestTopic(int numPartitions, short replicationFactor)
+ throws ExecutionException, InterruptedException {
+ topic = "kafka-source-test-" + UUID.randomUUID();
+ final CreateTopicsResult result =
+ admin.createTopics(
+ Collections.singletonList(
+ new NewTopic(topic, numPartitions, replicationFactor)));
+ result.all().get();
+ }
+}