apache · lifulong · Aug 17, 2026
diff --git a/docs/content.zh/docs/connectors/flink-sources/tidb-cdc.md b/docs/content.zh/docs/connectors/flink-sources/tidb-cdc.md
@@ -237,15 +237,14 @@ The TiDB CDC source can work in parallel reading, because there is multiple task
 
 ### DataStream Source
 
-The TiDB CDC connector can also be a DataStream source. You can create a SourceFunction as the following shows:
-
-### DataStream Source
+The TiDB CDC connector can also be a DataStream source. Region-based splits are computed once by the source enumerator; restore reuses the checkpointed key ranges instead of querying PD again.
 
 ```java
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Source;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
 import org.apache.flink.util.Collector;
 
 import org.apache.flink.cdc.connectors.tidb.TDBSourceOptions;
@@ -261,13 +260,13 @@ public class TiDBSourceExample {
 
     public static void main(String[] args) throws Exception {
 
-        SourceFunction<String> tidbSource =
+        Source<String, ?, ?> tidbSource =
             TiDBSource.<String>builder()
                 .database("mydb") // set captured database
                 .tableName("products") // set captured table
                 .tiConf(
                     TDBSourceOptions.getTiConfiguration(
-                        "localhost:2399", new HashMap<>()))
+                        "localhost:2399", null, new HashMap<>()))
                 .snapshotEventDeserializer(
                     new TiKVSnapshotEventDeserializationSchema<String>() {
                         @Override
@@ -302,7 +301,9 @@ public class TiDBSourceExample {
 
         // enable checkpoint
         env.enableCheckpointing(3000);
-        env.addSource(tidbSource).print().setParallelism(1);
+        env.fromSource(tidbSource, WatermarkStrategy.noWatermarks(), "TiDB Source")
+                .print()
+                .setParallelism(1);
 
         env.execute("Print TiDB Snapshot + Binlog");
     }

diff --git a/docs/content/docs/connectors/flink-sources/tidb-cdc.md b/docs/content/docs/connectors/flink-sources/tidb-cdc.md
@@ -237,15 +237,14 @@ The TiDB CDC source can work in parallel reading, because there is multiple task
 
 ### DataStream Source
 
-The TiDB CDC connector can also be a DataStream source. You can create a SourceFunction as the following shows:
-
-### DataStream Source
+The TiDB CDC connector can also be a DataStream source. Region-based splits are computed once by the source enumerator; restore reuses the checkpointed key ranges instead of querying PD again.
 
 ```java
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Source;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
 import org.apache.flink.util.Collector;
 
 import org.apache.flink.cdc.connectors.tidb.TDBSourceOptions;
@@ -261,13 +260,13 @@ public class TiDBSourceExample {
 
     public static void main(String[] args) throws Exception {
 
-        SourceFunction<String> tidbSource =
+        Source<String, ?, ?> tidbSource =
             TiDBSource.<String>builder()
                 .database("mydb") // set captured database
                 .tableName("products") // set captured table
                 .tiConf(
                     TDBSourceOptions.getTiConfiguration(
-                        "localhost:2399", new HashMap<>()))
+                        "localhost:2399", null, new HashMap<>()))
                 .snapshotEventDeserializer(
                     new TiKVSnapshotEventDeserializationSchema<String>() {
                         @Override
@@ -302,7 +301,9 @@ public class TiDBSourceExample {
 
         // enable checkpoint
         env.enableCheckpointing(3000);
-        env.addSource(tidbSource).print().setParallelism(1);
+        env.fromSource(tidbSource, WatermarkStrategy.noWatermarks(), "TiDB Source")
+                .print()
+                .setParallelism(1);
 
         env.execute("Print TiDB Snapshot + Binlog");
     }

diff --git a/...ink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/TiDBSource.java b/...ink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/TiDBSource.java
@@ -17,25 +17,128 @@
 
 package org.apache.flink.cdc.connectors.tidb;
 
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.cdc.common.annotation.PublicEvolving;
+import org.apache.flink.cdc.connectors.tidb.source.enumerator.TiKVEnumeratorState;
+import org.apache.flink.cdc.connectors.tidb.source.enumerator.TiKVEnumeratorStateSerializer;
+import org.apache.flink.cdc.connectors.tidb.source.enumerator.TiKVSourceEnumerator;
+import org.apache.flink.cdc.connectors.tidb.source.reader.TiKVSourceReader;
+import org.apache.flink.cdc.connectors.tidb.source.split.TiKVKeyRangeSplit;
+import org.apache.flink.cdc.connectors.tidb.source.split.TiKVKeyRangeSplitSerializer;
+import org.apache.flink.cdc.connectors.tidb.table.StartupMode;
 import org.apache.flink.cdc.connectors.tidb.table.StartupOptions;
-import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
 
 import org.tikv.common.TiConfiguration;
 
-/** A builder to build a SourceFunction which can read snapshot and continue to read CDC events. */
-public class TiDBSource {
+/**
+ * The TiDB CDC {@link Source} based on FLIP-27.
+ *
+ * <p>The enumerator splits the captured table by TiKV regions once and assigns a contiguous
+ * key-range to each reader. Restore reuses checkpointed splits and does not re-query PD for region
+ * topology.
+ *
+ * <pre>{@code
+ * Source<String, ?, ?> source =
+ *     TiDBSource.<String>builder()
+ *         .database("mydb")
+ *         .tableName("products")
+ *         .tiConf(tiConf)
+ *         .snapshotEventDeserializer(...)
+ *         .changeEventDeserializer(...)
+ *         .build();
+ * env.fromSource(source, WatermarkStrategy.noWatermarks(), "TiDB Source");
+ * }</pre>
+ *
+ * @param <T> the output type of the source.
+ */
+@PublicEvolving
+public class TiDBSource<T>
+        implements Source<T, TiKVKeyRangeSplit, TiKVEnumeratorState>, ResultTypeQueryable<T> {
+
+    private static final long serialVersionUID = 1L;
+
+    private final TiKVSnapshotEventDeserializationSchema<T> snapshotEventDeserializationSchema;
+    private final TiKVChangeEventDeserializationSchema<T> changeEventDeserializationSchema;
+    private final TiConfiguration tiConf;
+    private final StartupMode startupMode;
+    private final String database;
+    private final String tableName;
+
+    TiDBSource(
+            TiKVSnapshotEventDeserializationSchema<T> snapshotEventDeserializationSchema,
+            TiKVChangeEventDeserializationSchema<T> changeEventDeserializationSchema,
+            TiConfiguration tiConf,
+            StartupMode startupMode,
+            String database,
+            String tableName) {
+        this.snapshotEventDeserializationSchema = snapshotEventDeserializationSchema;
+        this.changeEventDeserializationSchema = changeEventDeserializationSchema;
+        this.tiConf = tiConf;
+        this.startupMode = startupMode;
+        this.database = database;
+        this.tableName = tableName;
+    }
 
     public static <T> Builder<T> builder() {
         return new Builder<>();
     }
 
+    @Override
+    public Boundedness getBoundedness() {
+        return Boundedness.CONTINUOUS_UNBOUNDED;
+    }
+
+    @Override
+    public SourceReader<T, TiKVKeyRangeSplit> createReader(SourceReaderContext readerContext) {
+        return new TiKVSourceReader<>(
+                readerContext,
+                snapshotEventDeserializationSchema,
+                changeEventDeserializationSchema,
+                tiConf,
+                startupMode);
+    }
+
+    @Override
+    public SplitEnumerator<TiKVKeyRangeSplit, TiKVEnumeratorState> createEnumerator(
+            SplitEnumeratorContext<TiKVKeyRangeSplit> enumContext) {
+        return new TiKVSourceEnumerator(enumContext, tiConf, database, tableName);
+    }
+
+    @Override
+    public SplitEnumerator<TiKVKeyRangeSplit, TiKVEnumeratorState> restoreEnumerator(
+            SplitEnumeratorContext<TiKVKeyRangeSplit> enumContext, TiKVEnumeratorState checkpoint) {
+        return new TiKVSourceEnumerator(enumContext, tiConf, database, tableName, checkpoint);
+    }
+
+    @Override
+    public SimpleVersionedSerializer<TiKVKeyRangeSplit> getSplitSerializer() {
+        return TiKVKeyRangeSplitSerializer.INSTANCE;
+    }
+
+    @Override
+    public SimpleVersionedSerializer<TiKVEnumeratorState> getEnumeratorCheckpointSerializer() {
+        return TiKVEnumeratorStateSerializer.INSTANCE;
+    }
+
+    @Override
+    public TypeInformation<T> getProducedType() {
+        return snapshotEventDeserializationSchema.getProducedType();
+    }
+
     /** Builder class of {@link TiDBSource}. */
     public static class Builder<T> {
         private String database;
         private String tableName;
         private StartupOptions startupOptions = StartupOptions.initial();
         private TiConfiguration tiConf;
-
         private TiKVSnapshotEventDeserializationSchema<T> snapshotEventDeserializationSchema;
         private TiKVChangeEventDeserializationSchema<T> changeEventDeserializationSchema;
 
@@ -77,9 +180,8 @@ public Builder<T> tiConf(TiConfiguration tiConf) {
             return this;
         }
 
-        public RichParallelSourceFunction<T> build() {
-
-            return new TiKVRichParallelSourceFunction<>(
+        public TiDBSource<T> build() {
+            return new TiDBSource<>(
                     snapshotEventDeserializationSchema,
                     changeEventDeserializationSchema,
                     tiConf,

diff --git a/...dc/src/main/java/org/apache/flink/cdc/connectors/tidb/TiKVRichParallelSourceFunction.java b/...dc/src/main/java/org/apache/flink/cdc/connectors/tidb/TiKVRichParallelSourceFunction.java
@@ -63,9 +63,14 @@
 import java.util.concurrent.TimeUnit;
 
 /**
- * The source implementation for TiKV that read snapshot events first and then read the change
- * event.
+ * Deprecated SourceFunction implementation. Use {@link
+ * org.apache.flink.cdc.connectors.tidb.TiDBSource} which splits the table once on the enumerator.
+ *
+ * @deprecated Use {@link org.apache.flink.cdc.connectors.tidb.TiDBSource} with {@code
+ *     env.fromSource(...)}. Per-subtask region discovery can assign overlapping or gapped key
+ *     ranges when tasks start at different times.
  */
+@Deprecated
 public class TiKVRichParallelSourceFunction<T> extends RichParallelSourceFunction<T>
         implements CheckpointListener, CheckpointedFunction, ResultTypeQueryable<T> {
 

diff --git a/...idb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/metrics/TiDBSourceMetrics.java b/...idb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/metrics/TiDBSourceMetrics.java
@@ -17,23 +17,22 @@
 
 package org.apache.flink.cdc.connectors.tidb.metrics;
 
-import org.apache.flink.cdc.connectors.tidb.TiKVRichParallelSourceFunction;
 import org.apache.flink.metrics.Gauge;
 import org.apache.flink.metrics.MetricGroup;
 
 import static org.apache.flink.runtime.metrics.MetricNames.CURRENT_EMIT_EVENT_TIME_LAG;
 import static org.apache.flink.runtime.metrics.MetricNames.CURRENT_FETCH_EVENT_TIME_LAG;
 import static org.apache.flink.runtime.metrics.MetricNames.SOURCE_IDLE_TIME;
 
-/** A collection class for handling metrics in {@link TiKVRichParallelSourceFunction}. */
+/** A collection class for handling metrics in the TiDB CDC source. */
 public class TiDBSourceMetrics {
 
     private final MetricGroup metricGroup;
 
     /**
-     * The last record processing time, which is updated after {@link
-     * TiKVRichParallelSourceFunction} fetches a batch of data. It's mainly used to report metrics
-     * sourceIdleTime for sourceIdleTime = System.currentTimeMillis() - processTime.
+     * The last record processing time, which is updated after the source fetches a batch of data.
+     * It's mainly used to report metrics sourceIdleTime for sourceIdleTime =
+     * System.currentTimeMillis() - processTime.
      */
     private long processTime = 0L;
 

diff --git a/...main/java/org/apache/flink/cdc/connectors/tidb/source/enumerator/TiKVEnumeratorState.java b/...main/java/org/apache/flink/cdc/connectors/tidb/source/enumerator/TiKVEnumeratorState.java
@@ -0,0 +1,95 @@
+/*
+ * 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.tidb.source.enumerator;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.connectors.tidb.source.split.TiKVKeyRangeSplit;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Checkpoint state of {@link TiKVSourceEnumerator}.
+ *
+ * <p>{@code enumerated} must stay true after the first successful PD split so restore never
+ * re-fetches region topology.
+ */
+@Internal
+public class TiKVEnumeratorState {
+
+    private final List<TiKVKeyRangeSplit> unassignedSplits;
+    private final boolean enumerated;
+    private final int parallelism;
+
+    public TiKVEnumeratorState(List<TiKVKeyRangeSplit> unassignedSplits, boolean enumerated) {
+        this(unassignedSplits, enumerated, -1);
+    }
+
+    public TiKVEnumeratorState(
+            List<TiKVKeyRangeSplit> unassignedSplits, boolean enumerated, int parallelism) {
+        this.unassignedSplits =
+                Collections.unmodifiableList(
+                        new ArrayList<>(Objects.requireNonNull(unassignedSplits)));
+        this.enumerated = enumerated;
+        this.parallelism = parallelism;
+    }
+
+    public List<TiKVKeyRangeSplit> getUnassignedSplits() {
+        return unassignedSplits;
+    }
+
+    public boolean isEnumerated() {
+        return enumerated;
+    }
+
+    public int getParallelism() {
+        return parallelism;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        TiKVEnumeratorState that = (TiKVEnumeratorState) o;
+        return enumerated == that.enumerated
+                && parallelism == that.parallelism
+                && Objects.equals(unassignedSplits, that.unassignedSplits);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(unassignedSplits, enumerated, parallelism);
+    }
+
+    @Override
+    public String toString() {
+        return "TiKVEnumeratorState{enumerated="
+                + enumerated
+                + ", parallelism="
+                + parallelism
+                + ", unassigned="
+                + unassignedSplits.size()
+                + '}';
+    }
+}