[FLINK-35104] [pipeline-connector/kafka]support pipeline kafka source

.github/workflows/flink_cdc_base.yml

@@ -90,7 +90,7 @@ jobs:
           maven-version: 3.8.6
 
       - name: Compile and test
-        timeout-minutes: 90
+        timeout-minutes: 120
         run: |
           . .github/workflows/utils.sh
           jvm_timezone=$(random_timezone)

docs/content.zh/docs/connectors/pipeline-connectors/kafka.md

@@ -26,12 +26,74 @@ under the License.
 
 # Kafka Pipeline 连接器
 
-Kafka Pipeline 连接器可以用作 Pipeline 的 *Data Sink*，将数据写入[Kafka](https://kafka.apache.org)。 本文档介绍如何设置 Kafka Pipeline 连接器。
+Kafka Pipeline 连接器可以用作 Pipeline 的 *Data Source* 或 *Data Sink*。作为 Source 时，它消费 Debezium JSON 或 Canal JSON Changelog，并将推断出的表结构变化转换为 Pipeline Schema 事件。
 
 ## 连接器的功能
 * 自动建表
 * 表结构变更同步
 * 数据实时同步
+* 消费 Debezium JSON 或 Canal JSON Changelog
+
+Kafka Source
+----------------
+
+下面的 Pipeline 从 Kafka 多分区消费 Debezium JSON，并写入 StarRocks：
+
+```yaml
+source:
+  type: kafka
+  name: Kafka Debezium Source
+  topic: inventory.customers
+  group-id: flink-cdc-kafka-source
+  scan.startup.mode: group-offsets
+  properties.bootstrap.servers: localhost:9092
+
+transform:
+  - source-table: inventory.\.*
+    primary-keys: id
+
+sink:
+  type: starrocks
+  name: StarRocks Sink
+  jdbc-url: jdbc:mysql://localhost:9030
+  load-url: localhost:8030
+  username: root
+  password: ""
+
+pipeline:
+  name: Kafka to StarRocks Pipeline
+  parallelism: 4
+  schema.change.behavior: lenient
+```
+
+Kafka Source 配置项：
+
+* `topic`：单个 Topic 或逗号分隔的 Topic 列表。
+* `topic-pattern`：用于动态发现 Topic 的正则表达式；`topic` 与 `topic-pattern` 必须且只能配置一个。
+* `group-id`（未配置 `properties.group.id` 时必填）：Kafka Consumer Group。
+* `scan.startup.mode`：`group-offsets`（默认）、`earliest-offset`、`latest-offset`、`timestamp` 或 `specific-offsets`。
+* `scan.startup.timestamp-millis`：当 `scan.startup.mode` 为 `timestamp` 时必填。
+* `scan.startup.specific-offsets`：当 `scan.startup.mode` 为 `specific-offsets` 时必填。仅配置一个 `topic` 时可写 `partition:0,offset:42;partition:1,offset:300`；多 Topic 或 `topic-pattern` 时每条必须带 topic，例如 `topic:dbz.customers,partition:0,offset:42`。未列出的分区从 earliest offset 开始。
+* `tables`：可选，按 Debezium `source.db`/`source.table` 或 Canal `database`/`table` 做包含过滤，语法与 MySQL source 的 `tables` 相同，例如 `inventory.customers` 或 `inventory.\\.*`。
+* `tables.exclude`：可选，排除匹配的表，可单独使用，也可与 `tables` 同时使用。
+* `value.format`：`debezium-json`（默认）或 `canal-json`。
+* `properties.bootstrap.servers`（必填）及 `properties.*`：Kafka Consumer 参数。
+
+Debezium JSON 不推断主键，请在 Pipeline 的 `transform` 中通过 `primary-keys` 指定。Canal JSON 会使用消息里的 `pkNames` 作为主键，仍可用 transform 覆盖。写入 StarRocks 前表必须具备主键。
+
+Debezium JSON 的 value 必须同时包含 `schema` 与 `payload`。不包含 schema 的 value 无法可靠识别字段类型变化，因此会被拒绝。
+
+Canal JSON 用 `mysqlType` 推断列类型、用 `pkNames` 作为主键。只消费 `INSERT`/`UPDATE`/`DELETE`；`isDdl=true` 以及其他 `type` 会被跳过。Canal 的 `UPDATE` 往往只在 `old` 里放变更列，Source 会用 `old` 覆盖 `data` 拼出完整 before。
+
+Transform 中的主键用于下游分区和 upsert 语义，但无法恢复 Kafka 中已经丢失的顺序；生产端仍需保证相同逻辑主键的变更进入同一个 Kafka 分区。
+
+### Schema Evolution 与多分区
+
+Kafka 只保证分区内有序。Source 会先在每个 Source subtask 内对其负责分区的 schema 做单调扩宽，再由 distributed schema coordinator 跨 subtask 合并。新 schema 出现后到达的旧格式消息会被转换到当前最宽 schema，不会触发类型回退。
+
+Source 支持首次发现表时建表、增加 nullable 字段，以及把 schema 做成单调超集：源端删列或改名时会保留旧列（NOT NULL 会改为 nullable）并加入新列名，不会发出 Drop/Rename。历史行的新列为 null，之后行的旧列为 null。另外支持 `INT → BIGINT`、`INT → STRING`、Decimal 精度扩大等兼容扩宽。Kafka Connect 的 `string` 一律映射为 `STRING`（MySQL 的 `CHAR`/`VARCHAR`/`TEXT` 在 Debezium JSON 中都是 `string`）。从旧 offset 重放并跨过 `INT → STRING` 时，历史整型值会被转成字符串，而不会失败。类型缩窄、不兼容类型变化会明确失败。并行 Kafka Source 应配置 `schema.change.behavior: lenient`。
+
+从旧 offset 重刷时，空目标表会按历史顺序执行 `CREATE → ADD/ALTER`。已有 StarRocks 表必须是历史 schema 的兼容超集；重复的 Create/Add/Alter 会按幂等方式处理，目标表额外字段必须 nullable 或有默认值。StarRocks 主键表可以通过 upsert 覆盖旧记录；duplicate-key 表全量重刷前应清表或改写新表。
 
 如何创建 Pipeline
 ----------------

docs/content/docs/connectors/pipeline-connectors/kafka.md

@@ -26,10 +26,80 @@ under the License.
 
 # Kafka Pipeline Connector
 
-The Kafka Pipeline connector can be used as the *Data Sink* of the pipeline, and write data to [Kafka](https://kafka.apache.org). This document describes how to set up the Kafka Pipeline connector.
+The Kafka Pipeline connector can be used as a *Data Source* or *Data Sink* of the pipeline. As a source, it consumes Debezium JSON or Canal JSON changelog records and converts inferred schema changes into pipeline schema events.
 
 ## What can the connector do?
 * Data synchronization
+* Consume Debezium JSON or Canal JSON changelog records
+* Infer create table, add column, and compatible column type widening events
+
+Kafka Source
+----------------
+
+The following pipeline consumes Debezium JSON from multiple Kafka partitions and writes it to StarRocks:
+
+```yaml
+source:
+  type: kafka
+  name: Kafka Debezium Source
+  topic: inventory.customers
+  group-id: flink-cdc-kafka-source
+  scan.startup.mode: group-offsets
+  properties.bootstrap.servers: localhost:9092
+
+transform:
+  - source-table: inventory.\.*
+    primary-keys: id
+
+sink:
+  type: starrocks
+  name: StarRocks Sink
+  jdbc-url: jdbc:mysql://localhost:9030
+  load-url: localhost:8030
+  username: root
+  password: ""
+
+pipeline:
+  name: Kafka to StarRocks Pipeline
+  parallelism: 4
+  schema.change.behavior: lenient
+```
+
+Kafka source options:
+
+* `topic`: one topic or a comma-separated topic list.
+* `topic-pattern`: a regular expression for discovering topics. Configure exactly one of `topic` and `topic-pattern`.
+* `group-id` (required unless `properties.group.id` is set): Kafka consumer group.
+* `scan.startup.mode`: `group-offsets` (default), `earliest-offset`, `latest-offset`, `timestamp`, or `specific-offsets`.
+* `scan.startup.timestamp-millis`: required when `scan.startup.mode` is `timestamp`.
+* `scan.startup.specific-offsets`: required when `scan.startup.mode` is `specific-offsets`. Use `partition:0,offset:42;partition:1,offset:300` when exactly one `topic` is configured, or include a topic in each entry such as `topic:dbz.customers,partition:0,offset:42`. Partitions that are not listed start from the earliest offset.
+* `tables`: optional inclusion patterns matched against Debezium `source.db`/`source.table` or Canal `database`/`table` (same selector syntax as the MySQL source, for example `inventory.customers` or `inventory.\\.*`).
+* `tables.exclude`: optional exclusion patterns. Can be used alone or together with `tables`.
+* `value.format`: `debezium-json` (default) or `canal-json`.
+* `properties.bootstrap.servers` (required) and `properties.*`: Kafka consumer properties.
+
+Primary keys are not inferred from Debezium JSON. Assign them with a pipeline `transform` `primary-keys` option. Canal JSON uses `pkNames` as the table primary key when present; transform can still override them. A StarRocks sink still requires a primary key before it can create a table.
+
+Debezium JSON values must include the `schema` and `payload` fields. Values without an embedded schema cannot reliably describe column type changes and are rejected.
+
+Canal JSON values use `mysqlType` for column types and `pkNames` for primary keys. `INSERT`/`UPDATE`/`DELETE` are consumed; `isDdl=true` and other `type` values are skipped. Canal `UPDATE` records often put only changed columns in `old`: the source builds a full before image by overlaying `old` onto `data`.
+
+The transform primary key determines downstream partitioning and upsert semantics, but it cannot restore ordering already lost in Kafka. Producers must still send changes for the same logical primary key to the same Kafka partition.
+
+### Schema evolution and multiple partitions
+
+Kafka only guarantees ordering within a partition. The source therefore merges schemas monotonically across the partitions assigned to each source subtask and the distributed schema coordinator merges them again across subtasks. Old records arriving after a newer schema are converted to the widest known schema instead of reverting it.
+
+The source supports:
+
+* creating a table from the first record seen for a table;
+* adding nullable columns;
+* keeping a monotonic column superset: dropped or renamed source columns are retained (NOT NULL columns become nullable) and new names are added. Historical rows have nulls in new columns; later rows have nulls in old columns. The source does not emit drop or rename events;
+* compatible type widening, such as `INT` to `BIGINT`, `INT` to `STRING`, or increasing decimal precision. Kafka Connect `string` is always mapped to `STRING` (MySQL `CHAR`/`VARCHAR`/`TEXT` all become `string` in Debezium JSON). Replaying from an old offset that spans an `INT` → `STRING` change converts historical integer values to strings instead of failing.
+
+Narrowing types and incompatible type changes fail explicitly. Use `schema.change.behavior: lenient` for a parallel Kafka source.
+
+When replaying from an old offset, an empty target table follows the historical `CREATE → ADD/ALTER` sequence. An existing StarRocks table must be a compatible superset. Replayed create/add/alter operations are idempotent; extra target columns must be nullable or have defaults. Replaying rows is safe for primary-key tables through upserts. Duplicate-key tables should be cleared or replaced before a full replay.
 
 How to create Pipeline
 ----------------

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/sink/SupportsParallelMetadataSource.java

@@ -0,0 +1,30 @@
+/*
+ * 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.common.sink;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+
+/**
+ * Internal capability for sinks whose topology must adapt to sources that emit metadata from
+ * parallel subtasks.
+ */
+@Internal
+public interface SupportsParallelMetadataSource {
+
+    void setParallelMetadataSource(boolean parallelMetadataSource);
+}

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java

@@ -29,6 +29,7 @@
 import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
 import org.apache.flink.cdc.common.sink.DataSink;
 import org.apache.flink.cdc.common.sink.DefaultDataChangeEventHashFunctionProvider;
+import org.apache.flink.cdc.common.sink.SupportsParallelMetadataSource;
 import org.apache.flink.cdc.common.sink.TableIdHashFunctionProvider;
 import org.apache.flink.cdc.common.source.DataSource;
 import org.apache.flink.cdc.composer.PipelineComposer;
@@ -172,6 +173,10 @@ private void translate(StreamExecutionEnvironment env, PipelineDef pipelineDef)
                 resolveHashFunctionProvider(pipelineDefConfig, sinkDefinedHashFunctionProvider);
 
         boolean isParallelMetadataSource = dataSource.isParallelMetadataSource();
+        if (dataSink instanceof SupportsParallelMetadataSource) {
+            ((SupportsParallelMetadataSource) dataSink)
+                    .setParallelMetadataSource(isParallelMetadataSource);
+        }
 
         // O ---> Source
         DataStream<Event> stream =