leonardBang commented on code in PR #3952: URL: https://github.com/apache/flink-cdc/pull/3952#discussion_r4061297157
########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/source/fetch/TiDBScanFetchTask.java: ########## @@ -0,0 +1,318 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.tidb.source.fetch; + +import org.apache.flink.cdc.connectors.base.relational.JdbcSourceEventDispatcher; +import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit; +import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit; +import org.apache.flink.cdc.connectors.base.source.meta.wartermark.WatermarkKind; +import org.apache.flink.cdc.connectors.base.source.reader.external.AbstractScanFetchTask; +import org.apache.flink.cdc.connectors.tidb.source.config.TiDBConnectorConfig; +import org.apache.flink.cdc.connectors.tidb.source.connection.TiDBConnection; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffset; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetContext; +import org.apache.flink.cdc.connectors.tidb.source.schema.TiDBDatabaseSchema; +import org.apache.flink.cdc.connectors.tidb.utils.TiDBUtils; + +import io.debezium.connector.tidb.TiDBPartition; +import io.debezium.pipeline.EventDispatcher; +import io.debezium.pipeline.source.AbstractSnapshotChangeEventSource; +import io.debezium.pipeline.source.spi.ChangeEventSource; +import io.debezium.pipeline.source.spi.SnapshotProgressListener; +import io.debezium.pipeline.spi.ChangeRecordEmitter; +import io.debezium.pipeline.spi.SnapshotResult; +import io.debezium.relational.RelationalSnapshotChangeEventSource; +import io.debezium.relational.SnapshotChangeRecordEmitter; +import io.debezium.relational.Table; +import io.debezium.relational.TableId; +import io.debezium.util.Clock; +import io.debezium.util.ColumnUtils; +import io.debezium.util.Strings; +import io.debezium.util.Threads; +import org.apache.kafka.connect.errors.ConnectException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.time.Duration; + +/** A wrapped task to fetch snapshot split of table. */ +public class TiDBScanFetchTask extends AbstractScanFetchTask { + private static final Logger LOG = LoggerFactory.getLogger(TiDBScanFetchTask.class); + + public TiDBScanFetchTask(SnapshotSplit split) { + super(split); + } + + @Override + protected void executeBackfillTask(Context context, StreamSplit backfillStreamSplit) + throws Exception { + + // just for test + TiDBSourceFetchTaskContext ctx = (TiDBSourceFetchTaskContext) context; + final EventOffset currentOffset = + EventOffset.of( + ((TiDBSourceFetchTaskContext) context).getOffsetContext().getOffset()); + JdbcSourceEventDispatcher dispatcher = ctx.getEventDispatcher(); + dispatcher.dispatchWatermarkEvent( + ctx.getPartition().getSourcePartition(), + backfillStreamSplit, + currentOffset, + WatermarkKind.END); Review Comment: **Blocking: the backfill read is skipped entirely.** `executeBackfillTask()` never consumes `backfillStreamSplit` — it just takes the current offset and dispatches the END watermark, which discards the whole [LOW, HIGH) interval that the incremental framework relies on for snapshot/stream consistency. Concretely: a row read as `v1` during the snapshot and updated to `v2` before HIGH will stay `v1` forever, since streaming starts after the boundary. Unless the snapshot reads are proven TSO-consistent at HIGH (which would need to be made explicit), this breaks exactly-once semantics — and the `// just for test` comment suggests this is a known placeholder. Could you either implement the backfill consumption (via the TiKV CDC reader over the same key range), or document + prove why it's unnecessary for TiDB? Also worth removing the raw `JdbcSourceEventDispatcher` usage in favor of the typed one while here. ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/source/config/TiDBSourceConfig.java: ########## @@ -0,0 +1,141 @@ +/* + * 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.config; + +import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfig; +import org.apache.flink.cdc.connectors.base.options.StartupOptions; +import org.apache.flink.table.catalog.ObjectPath; + +import io.debezium.config.Configuration; +import org.tikv.common.TiConfiguration; + +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +/** The configuration for TiDB source. */ +public class TiDBSourceConfig extends JdbcSourceConfig { + private static final long serialVersionUID = 1L; + private final String compatibleMode; + private final String pdAddresses; + + private final String hostMapping; + private TiConfiguration tiConfiguration; Review Comment: **Blocking: non-serializable field on a serializable config.** `org.tikv.common.TiConfiguration` is not `Serializable` (and not marked `transient` here), but this config crosses the source serialization boundary — it's embedded in the source/fetch-task objects that go JM → TM and into checkpoints. This will blow up with `NotSerializableException` at submission or restore. Better to store only serializable primitives here (pd addresses, host mapping, TiKV properties, timeouts) and construct `TiConfiguration` lazily on the TM side. A small `InstantiationUtil.serializeObject(config)` unit test would lock this down. ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/source/fetch/EventSourceReader.java: ########## @@ -0,0 +1,499 @@ +/* + * 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.fetch; + +import org.apache.flink.cdc.connectors.base.relational.JdbcSourceEventDispatcher; +import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit; +import org.apache.flink.cdc.connectors.tidb.source.config.TiDBConnectorConfig; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffset; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetContext; +import org.apache.flink.cdc.connectors.tidb.utils.TableKeyRangeUtils; +import org.apache.flink.util.Preconditions; + +import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder; + +import io.debezium.connector.tidb.TiDBPartition; +import io.debezium.data.Envelope; +import io.debezium.pipeline.ErrorHandler; +import io.debezium.pipeline.source.spi.StreamingChangeEventSource; +import io.debezium.relational.TableId; +import io.debezium.relational.TableSchema; +import io.debezium.util.Clock; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.tikv.cdc.CDCClient; +import org.tikv.common.TiConfiguration; +import org.tikv.common.TiSession; +import org.tikv.common.key.RowKey; +import org.tikv.common.meta.TiColumnInfo; +import org.tikv.common.meta.TiTableInfo; +import org.tikv.kvproto.Cdcpb; +import org.tikv.kvproto.Coprocessor; +import org.tikv.shade.com.google.protobuf.ByteString; + +import java.io.Serializable; +import java.time.Instant; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.tikv.common.codec.TableCodec.decodeObjects; +import static org.tikv.common.codec.TiDBRowV2Decoder.decodeObjectsPreservingBinary; + +/** TiDB streaming change event source reader. */ +public class EventSourceReader + implements StreamingChangeEventSource<TiDBPartition, EventOffsetContext> { + private static final Logger LOG = LoggerFactory.getLogger(EventSourceReader.class); + private final StreamSplit split; + private final TiDBConnectorConfig connectorConfig; + private final TiConfiguration ticonf; + private final JdbcSourceEventDispatcher<TiDBPartition> eventDispatcher; + private final ErrorHandler errorHandler; + private final TiDBSourceFetchTaskContext taskContext; + private final Map<TableSchema, Map<String, Integer>> fieldIndexMap = new HashMap<>(); + public volatile ChangeEventSourceContext context; + + private static final long STREAMING_VERSION_START_EPOCH = 0L; + + /** Task local variables. */ + private transient TiSession session = null; + + private transient Coprocessor.KeyRange keyRange = null; + private transient CDCClient cdcClient = null; + private transient volatile long resolvedTs = -1L; + private transient TreeMap<RowKeyWithTs, Cdcpb.Event.Row> prewrites = null; + private transient TreeMap<RowKeyWithTs, Cdcpb.Event.Row> commits = null; + private transient BlockingQueue<Cdcpb.Event.Row> committedEvents = null; + private transient TableId tableId; + private transient TiTableInfo tableInfo; + + private transient volatile boolean running; + private transient volatile Thread executionThread; + private transient ExecutorService executorService; + private final AtomicBoolean closed = new AtomicBoolean(false); + + public EventSourceReader( + TiDBConnectorConfig connectorConfig, + JdbcSourceEventDispatcher eventDispatcher, + ErrorHandler errorHandler, + TiDBSourceFetchTaskContext taskContext, + StreamSplit split) { + this.connectorConfig = connectorConfig; + this.ticonf = connectorConfig.getSourceConfig().getTiConfiguration(); + this.eventDispatcher = eventDispatcher; + this.errorHandler = errorHandler; + this.taskContext = taskContext; + this.split = split; + } + + @Override + public synchronized void init() throws InterruptedException { + if (closed.get()) { + return; + } + StreamingChangeEventSource.super.init(); + try { + session = TiSession.create(ticonf); + Set<TableId> tableIds = this.split.getTableSchemas().keySet(); + if (tableIds.size() != 1) { + throw new IllegalStateException( + "Currently only single table ingest is supported, but found " + + tableIds.size() + + " tables."); + } + this.tableId = tableIds.stream().findFirst().get(); + this.tableInfo = session.getCatalog().getTable(tableId.catalog(), tableId.table()); + if (tableInfo == null) { + throw new RuntimeException( + String.format( + "Table %s.%s does not exist.", tableId.catalog(), tableId.table())); + } + keyRange = TableKeyRangeUtils.getTableKeyRange(tableInfo.getId(), 1, 0); + cdcClient = new CDCClient(session, keyRange); + prewrites = new TreeMap<>(); + commits = new TreeMap<>(); + // cdc event will lose if pull cdc event block when region split + // use queue to separate read and write to ensure pull event unblock. + // since sink jdbc is slow, 5000W queue size may be safe size. + committedEvents = new LinkedBlockingQueue<>(); + resolvedTs = EventOffset.getStartTs(this.split.getStartingOffset()); + ThreadFactory threadFactory = + new ThreadFactoryBuilder().setNameFormat("tidb-source-function-0").build(); + executorService = Executors.newSingleThreadExecutor(threadFactory); + } catch (RuntimeException e) { + close(); + throw e; + } + } + + @Override + public void execute( + ChangeEventSourceContext context, + TiDBPartition partition, + EventOffsetContext offsetContext) + throws InterruptedException { + if (closed.get()) { + return; + } + this.context = context; + this.executionThread = Thread.currentThread(); + running = true; + try { + if (connectorConfig.getSourceConfig().getStartupOptions().isSnapshotOnly()) { + LOG.info("Streaming is not enabled in current configuration"); + return; + } + this.taskContext.getDatabaseSchema().assureNonEmptySchema(); + cdcClient.start(resolvedTs); + EventOffsetContext effectiveOffsetContext = + offsetContext != null + ? offsetContext + : EventOffsetContext.initial(this.connectorConfig); + readChangeEvents(partition, effectiveOffsetContext); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + if (!closed.get()) { + throw e; + } + } catch (Exception e) { + if (!closed.get()) { + this.errorHandler.setProducerThrowable(e); + } + } finally { + running = false; + executionThread = null; + } + } + + protected void readChangeEvents(TiDBPartition partition, EventOffsetContext offsetContext) + throws Exception { + LOG.info("read change event from resolvedTs:{}", resolvedTs); + // child thread to sink committed rows. + executorService.execute( + () -> { + while (running && context.isRunning()) { + try { + Cdcpb.Event.Row committedRow = committedEvents.take(); + emitChangeEvent(partition, offsetContext, committedRow); + // use startTs of row as messageTs, use commitTs of row as fetchTs + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } catch (Exception e) { + if (running && context.isRunning()) { + LOG.error("Read change events error.", e); + } + } Review Comment: **Blocking: swallowed emission exceptions = silent data loss.** Once `committedEvents.take()` succeeds, the record is dequeued; if `emitChangeEvent` then throws, the catch-all logs and continues — that CDC event is gone permanently while the job stays RUNNING. This PR adds several new conversion paths (RowV2 decoding, value converters, schema lookups) where such failures are realistic. An emission failure should be fatal: propagate it via the error handler / fail the source so Flink restores from the last successful checkpoint, instead of converting data loss into a log line. ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/source/fetch/EventSourceReader.java: ########## @@ -0,0 +1,499 @@ +/* + * 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.fetch; + +import org.apache.flink.cdc.connectors.base.relational.JdbcSourceEventDispatcher; +import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit; +import org.apache.flink.cdc.connectors.tidb.source.config.TiDBConnectorConfig; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffset; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetContext; +import org.apache.flink.cdc.connectors.tidb.utils.TableKeyRangeUtils; +import org.apache.flink.util.Preconditions; + +import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder; + +import io.debezium.connector.tidb.TiDBPartition; +import io.debezium.data.Envelope; +import io.debezium.pipeline.ErrorHandler; +import io.debezium.pipeline.source.spi.StreamingChangeEventSource; +import io.debezium.relational.TableId; +import io.debezium.relational.TableSchema; +import io.debezium.util.Clock; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.tikv.cdc.CDCClient; +import org.tikv.common.TiConfiguration; +import org.tikv.common.TiSession; +import org.tikv.common.key.RowKey; +import org.tikv.common.meta.TiColumnInfo; +import org.tikv.common.meta.TiTableInfo; +import org.tikv.kvproto.Cdcpb; +import org.tikv.kvproto.Coprocessor; +import org.tikv.shade.com.google.protobuf.ByteString; + +import java.io.Serializable; +import java.time.Instant; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.tikv.common.codec.TableCodec.decodeObjects; +import static org.tikv.common.codec.TiDBRowV2Decoder.decodeObjectsPreservingBinary; + +/** TiDB streaming change event source reader. */ +public class EventSourceReader + implements StreamingChangeEventSource<TiDBPartition, EventOffsetContext> { + private static final Logger LOG = LoggerFactory.getLogger(EventSourceReader.class); + private final StreamSplit split; + private final TiDBConnectorConfig connectorConfig; + private final TiConfiguration ticonf; + private final JdbcSourceEventDispatcher<TiDBPartition> eventDispatcher; + private final ErrorHandler errorHandler; + private final TiDBSourceFetchTaskContext taskContext; + private final Map<TableSchema, Map<String, Integer>> fieldIndexMap = new HashMap<>(); + public volatile ChangeEventSourceContext context; + + private static final long STREAMING_VERSION_START_EPOCH = 0L; + + /** Task local variables. */ + private transient TiSession session = null; + + private transient Coprocessor.KeyRange keyRange = null; + private transient CDCClient cdcClient = null; + private transient volatile long resolvedTs = -1L; + private transient TreeMap<RowKeyWithTs, Cdcpb.Event.Row> prewrites = null; + private transient TreeMap<RowKeyWithTs, Cdcpb.Event.Row> commits = null; + private transient BlockingQueue<Cdcpb.Event.Row> committedEvents = null; + private transient TableId tableId; + private transient TiTableInfo tableInfo; + + private transient volatile boolean running; + private transient volatile Thread executionThread; + private transient ExecutorService executorService; + private final AtomicBoolean closed = new AtomicBoolean(false); + + public EventSourceReader( + TiDBConnectorConfig connectorConfig, + JdbcSourceEventDispatcher eventDispatcher, + ErrorHandler errorHandler, + TiDBSourceFetchTaskContext taskContext, + StreamSplit split) { + this.connectorConfig = connectorConfig; + this.ticonf = connectorConfig.getSourceConfig().getTiConfiguration(); + this.eventDispatcher = eventDispatcher; + this.errorHandler = errorHandler; + this.taskContext = taskContext; + this.split = split; + } + + @Override + public synchronized void init() throws InterruptedException { + if (closed.get()) { + return; + } + StreamingChangeEventSource.super.init(); + try { + session = TiSession.create(ticonf); + Set<TableId> tableIds = this.split.getTableSchemas().keySet(); + if (tableIds.size() != 1) { + throw new IllegalStateException( + "Currently only single table ingest is supported, but found " + + tableIds.size() + + " tables."); + } + this.tableId = tableIds.stream().findFirst().get(); + this.tableInfo = session.getCatalog().getTable(tableId.catalog(), tableId.table()); + if (tableInfo == null) { + throw new RuntimeException( + String.format( + "Table %s.%s does not exist.", tableId.catalog(), tableId.table())); + } + keyRange = TableKeyRangeUtils.getTableKeyRange(tableInfo.getId(), 1, 0); + cdcClient = new CDCClient(session, keyRange); + prewrites = new TreeMap<>(); + commits = new TreeMap<>(); + // cdc event will lose if pull cdc event block when region split + // use queue to separate read and write to ensure pull event unblock. + // since sink jdbc is slow, 5000W queue size may be safe size. + committedEvents = new LinkedBlockingQueue<>(); Review Comment: **Unbounded queue defeats backpressure.** The comment above says "5000W queue size may be safe size", but the `LinkedBlockingQueue` is constructed without a capacity, so nothing enforces it. Under sink backpressure the CDC pull thread keeps ingesting from TiKV and the heap grows without bound → eventual OOM. A bounded capacity (config-driven) would turn this into normal Flink backpressure instead. ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/source/fetch/EventSourceReader.java: ########## @@ -0,0 +1,499 @@ +/* + * 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.fetch; + +import org.apache.flink.cdc.connectors.base.relational.JdbcSourceEventDispatcher; +import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit; +import org.apache.flink.cdc.connectors.tidb.source.config.TiDBConnectorConfig; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffset; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetContext; +import org.apache.flink.cdc.connectors.tidb.utils.TableKeyRangeUtils; +import org.apache.flink.util.Preconditions; + +import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder; + +import io.debezium.connector.tidb.TiDBPartition; +import io.debezium.data.Envelope; +import io.debezium.pipeline.ErrorHandler; +import io.debezium.pipeline.source.spi.StreamingChangeEventSource; +import io.debezium.relational.TableId; +import io.debezium.relational.TableSchema; +import io.debezium.util.Clock; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.tikv.cdc.CDCClient; +import org.tikv.common.TiConfiguration; +import org.tikv.common.TiSession; +import org.tikv.common.key.RowKey; +import org.tikv.common.meta.TiColumnInfo; +import org.tikv.common.meta.TiTableInfo; +import org.tikv.kvproto.Cdcpb; +import org.tikv.kvproto.Coprocessor; +import org.tikv.shade.com.google.protobuf.ByteString; + +import java.io.Serializable; +import java.time.Instant; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.tikv.common.codec.TableCodec.decodeObjects; +import static org.tikv.common.codec.TiDBRowV2Decoder.decodeObjectsPreservingBinary; + +/** TiDB streaming change event source reader. */ +public class EventSourceReader + implements StreamingChangeEventSource<TiDBPartition, EventOffsetContext> { + private static final Logger LOG = LoggerFactory.getLogger(EventSourceReader.class); + private final StreamSplit split; + private final TiDBConnectorConfig connectorConfig; + private final TiConfiguration ticonf; + private final JdbcSourceEventDispatcher<TiDBPartition> eventDispatcher; + private final ErrorHandler errorHandler; + private final TiDBSourceFetchTaskContext taskContext; + private final Map<TableSchema, Map<String, Integer>> fieldIndexMap = new HashMap<>(); + public volatile ChangeEventSourceContext context; + + private static final long STREAMING_VERSION_START_EPOCH = 0L; + + /** Task local variables. */ + private transient TiSession session = null; + + private transient Coprocessor.KeyRange keyRange = null; + private transient CDCClient cdcClient = null; + private transient volatile long resolvedTs = -1L; + private transient TreeMap<RowKeyWithTs, Cdcpb.Event.Row> prewrites = null; + private transient TreeMap<RowKeyWithTs, Cdcpb.Event.Row> commits = null; + private transient BlockingQueue<Cdcpb.Event.Row> committedEvents = null; + private transient TableId tableId; + private transient TiTableInfo tableInfo; + + private transient volatile boolean running; + private transient volatile Thread executionThread; + private transient ExecutorService executorService; + private final AtomicBoolean closed = new AtomicBoolean(false); + + public EventSourceReader( + TiDBConnectorConfig connectorConfig, + JdbcSourceEventDispatcher eventDispatcher, + ErrorHandler errorHandler, + TiDBSourceFetchTaskContext taskContext, + StreamSplit split) { + this.connectorConfig = connectorConfig; + this.ticonf = connectorConfig.getSourceConfig().getTiConfiguration(); + this.eventDispatcher = eventDispatcher; + this.errorHandler = errorHandler; + this.taskContext = taskContext; + this.split = split; + } + + @Override + public synchronized void init() throws InterruptedException { + if (closed.get()) { + return; + } + StreamingChangeEventSource.super.init(); + try { + session = TiSession.create(ticonf); + Set<TableId> tableIds = this.split.getTableSchemas().keySet(); + if (tableIds.size() != 1) { + throw new IllegalStateException( + "Currently only single table ingest is supported, but found " + + tableIds.size() + + " tables."); + } + this.tableId = tableIds.stream().findFirst().get(); + this.tableInfo = session.getCatalog().getTable(tableId.catalog(), tableId.table()); + if (tableInfo == null) { + throw new RuntimeException( + String.format( + "Table %s.%s does not exist.", tableId.catalog(), tableId.table())); + } + keyRange = TableKeyRangeUtils.getTableKeyRange(tableInfo.getId(), 1, 0); + cdcClient = new CDCClient(session, keyRange); + prewrites = new TreeMap<>(); + commits = new TreeMap<>(); + // cdc event will lose if pull cdc event block when region split + // use queue to separate read and write to ensure pull event unblock. + // since sink jdbc is slow, 5000W queue size may be safe size. + committedEvents = new LinkedBlockingQueue<>(); + resolvedTs = EventOffset.getStartTs(this.split.getStartingOffset()); + ThreadFactory threadFactory = + new ThreadFactoryBuilder().setNameFormat("tidb-source-function-0").build(); + executorService = Executors.newSingleThreadExecutor(threadFactory); + } catch (RuntimeException e) { + close(); + throw e; + } + } + + @Override + public void execute( + ChangeEventSourceContext context, + TiDBPartition partition, + EventOffsetContext offsetContext) + throws InterruptedException { + if (closed.get()) { + return; + } + this.context = context; + this.executionThread = Thread.currentThread(); + running = true; + try { + if (connectorConfig.getSourceConfig().getStartupOptions().isSnapshotOnly()) { + LOG.info("Streaming is not enabled in current configuration"); + return; + } + this.taskContext.getDatabaseSchema().assureNonEmptySchema(); + cdcClient.start(resolvedTs); + EventOffsetContext effectiveOffsetContext = + offsetContext != null + ? offsetContext + : EventOffsetContext.initial(this.connectorConfig); + readChangeEvents(partition, effectiveOffsetContext); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + if (!closed.get()) { + throw e; + } + } catch (Exception e) { + if (!closed.get()) { + this.errorHandler.setProducerThrowable(e); + } + } finally { + running = false; + executionThread = null; + } + } + + protected void readChangeEvents(TiDBPartition partition, EventOffsetContext offsetContext) + throws Exception { + LOG.info("read change event from resolvedTs:{}", resolvedTs); + // child thread to sink committed rows. + executorService.execute( + () -> { + while (running && context.isRunning()) { + try { + Cdcpb.Event.Row committedRow = committedEvents.take(); + emitChangeEvent(partition, offsetContext, committedRow); + // use startTs of row as messageTs, use commitTs of row as fetchTs + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } catch (Exception e) { + if (running && context.isRunning()) { + LOG.error("Read change events error.", e); + } + } + } + }); + while (running && context.isRunning() && resolvedTs >= STREAMING_VERSION_START_EPOCH) { + for (int i = 0; i < 1000; i++) { + final Cdcpb.Event.Row row = cdcClient.get(); + if (row == null) { + break; + } + handleRow(row); + } + resolvedTs = cdcClient.getMaxResolvedTs(); + if (commits.size() > 0) { + flushRows(resolvedTs); + } + } + } + + protected void emitChangeEvent( + TiDBPartition partition, EventOffsetContext offsetContext, final Cdcpb.Event.Row row) + throws Exception { + if (!context.isRunning()) { + LOG.info("sourceContext is not running, skip emit change event."); + return; + } + if (tableId == null) { + LOG.warn("No valid tableId found, skipping log message: {}", row); + return; + } + TableSchema tableSchema = taskContext.getDatabaseSchema().schemaFor(tableId); + if (tableSchema == null) { + LOG.warn("No table schema found, skipping log message: {}", row); + return; + } + offsetContext.event(tableSchema.id(), Instant.ofEpochMilli(row.getCommitTs())); Review Comment: **Blocking: TSO vs epoch-millis unit mixing.** `row.getCommitTs()` is a packed TiKV TSO (physical << 18 | logical), not wall-clock milliseconds — the rest of this PR treats it as a TSO (e.g. `EventOffset.getStartTs`, the `TiTimestamp` conversions). Passing a packed TSO into `Instant.ofEpochMilli` yields nonsensical source timestamps that end up in `sourceInfo` and downstream metadata columns. Also note `EventOffsetContext.event()` only touches `sourceInfo` — it never advances `timestamp`/`commitVersion`, so the checkpointed offset doesn't reliably track the last emitted record; on restore we could replay from a stale offset and re-emit a large window. I'd suggest keeping the two representations strictly separated: exact `commitVersion` (TSO) for offsets, decoded physical time for display timestamps — and adding a checkpoint → kill → restore ITCase asserting no pre-checkpoint records are re-emitted. ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/source/offset/EventOffset.java: ########## @@ -0,0 +1,122 @@ +/* + * 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.offset; + +import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset; + +import org.tikv.common.meta.TiTimestamp; + +import javax.annotation.Nonnull; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +/** The offset of TiDB binlog. */ +public class EventOffset extends Offset { + public static final String TIMESTAMP_KEY = "timestamp"; + // TimeStamp Oracle from pd + public static final String COMMIT_VERSION_KEY = "commit_version"; + + public static final EventOffset INITIAL_OFFSET = + new EventOffset(Collections.singletonMap(TIMESTAMP_KEY, "0")); + public static final EventOffset NO_STOPPING_OFFSET = new EventOffset(Long.MAX_VALUE); + + public EventOffset(Map<String, ?> offset) { + Map<String, String> offsetMap = new HashMap<>(); + for (Map.Entry<String, ?> entry : offset.entrySet()) { + offsetMap.put( + entry.getKey(), entry.getValue() == null ? null : entry.getValue().toString()); + } + this.offset = offsetMap; + } + + public EventOffset(@Nonnull String timestamp, String commitVersion) { + Map<String, String> offsetMap = new HashMap<>(); + offsetMap.put(TIMESTAMP_KEY, timestamp); + if (commitVersion != null) { + offsetMap.put(COMMIT_VERSION_KEY, commitVersion); + } + this.offset = offsetMap; + } + + public EventOffset(long binlogEpochMill) { + Map<String, String> offsetMap = new HashMap<>(); + offsetMap.put(TIMESTAMP_KEY, String.valueOf(binlogEpochMill)); + offsetMap.put( + COMMIT_VERSION_KEY, + String.valueOf(new TiTimestamp(binlogEpochMill, 0).getVersion())); + this.offset = offsetMap; + } + + public String getTimestamp() { + return offset.get(TIMESTAMP_KEY); + } + + public String getCommitVersion() { + if (offset.get(COMMIT_VERSION_KEY) == null) { + String timestamp = getTimestamp(); + // timestamp to commit version. + return String.valueOf(new TiTimestamp(Long.parseLong(timestamp), 0).getVersion()); + } + return offset.get(COMMIT_VERSION_KEY); + } + + @Override + public int compareTo(@Nonnull Offset o) { + EventOffset that = (EventOffset) o; + + int flag; + flag = compareLong(getTimestamp(), that.getTimestamp()); + if (flag != 0) { + return flag; + } + return compareLong(getCommitVersion(), that.getCommitVersion()); + } + + private int compareLong(String a, String b) { + if (a == null && b == null) { + return 0; + } + if (a == null) { + return -1; + } + if (b == null) { + return 1; + } + return Long.compare(Long.parseLong(a), Long.parseLong(b)); Review Comment: `Long.parseLong` here (and in `getCommitVersion()`/`getStartTs()`) throws `NumberFormatException` on malformed or legacy offset strings — `compareTo` is used for split/watermark ordering, so a bad offset blows up deep inside the framework rather than failing at the parse boundary with a clear message. Related unit-mixing: `getCommitVersion()` (L75) feeds an epoch-millis timestamp into `new TiTimestamp(...)`, which is the same TSO-vs-millis conflation as in the reader. ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/io/debezium/connector/tidb/TiDBPartition.java: ########## @@ -0,0 +1,60 @@ +/* + * 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 io.debezium.connector.tidb; + +import io.debezium.pipeline.spi.Partition; + +import java.util.Collections; +import java.util.Map; +import java.util.Objects; + +public class TiDBPartition implements Partition { + private final String serverName; + + public TiDBPartition(String serverName) { + this.serverName = serverName; + } + + @Override + public Map<String, String> getSourcePartition() { + return Collections.singletonMap("server", serverName); + } + + @Override + public Map<String, String> getLoggingContext() { + return Partition.super.getLoggingContext(); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + final io.debezium.connector.tidb.TiDBPartition other = + (io.debezium.connector.tidb.TiDBPartition) obj; + return Objects.equals(serverName, other.serverName); Review Comment: `equals()` is overridden without `hashCode()` — this violates the equals/hashCode contract, and Debezium uses partitions as source identities in maps/sets (offset lookups etc.), where an identity hashCode breaks equal-partition lookups. Also worth giving this a meaningful `toString()` (currently `super.toString()`) — partitions show up in logs and error messages all the time. ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/source/fetch/TiDBSourceFetchTaskContext.java: ########## @@ -0,0 +1,238 @@ +/* + * 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.fetch; + +import org.apache.flink.cdc.connectors.base.WatermarkDispatcher; +import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfig; +import org.apache.flink.cdc.connectors.base.dialect.JdbcDataSourceDialect; +import org.apache.flink.cdc.connectors.base.relational.JdbcSourceEventDispatcher; +import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset; +import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase; +import org.apache.flink.cdc.connectors.base.source.reader.external.JdbcSourceFetchTaskContext; +import org.apache.flink.cdc.connectors.base.utils.SplitKeyUtils; +import org.apache.flink.cdc.connectors.tidb.source.config.TiDBConnectorConfig; +import org.apache.flink.cdc.connectors.tidb.source.connection.TiDBConnection; +import org.apache.flink.cdc.connectors.tidb.source.handler.TiDBErrorHandler; +import org.apache.flink.cdc.connectors.tidb.source.handler.TiDBSchemaChangeEventHandler; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffset; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetContext; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetFactory; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetUtils; +import org.apache.flink.cdc.connectors.tidb.source.schema.TiDBDatabaseSchema; +import org.apache.flink.cdc.connectors.tidb.utils.TiDBUtils; +import org.apache.flink.table.types.logical.RowType; + +import io.debezium.connector.base.ChangeEventQueue; +import io.debezium.connector.tidb.TiDBEventMetadataProvider; +import io.debezium.connector.tidb.TiDBPartition; +import io.debezium.connector.tidb.TiDBTaskContext; +import io.debezium.pipeline.DataChangeEvent; +import io.debezium.pipeline.ErrorHandler; +import io.debezium.pipeline.metrics.DefaultChangeEventSourceMetricsFactory; +import io.debezium.pipeline.metrics.SnapshotChangeEventSourceMetrics; +import io.debezium.pipeline.metrics.spi.ChangeEventSourceMetricsFactory; +import io.debezium.pipeline.source.spi.EventMetadataProvider; +import io.debezium.relational.Table; +import io.debezium.relational.TableId; +import io.debezium.relational.Tables; +import io.debezium.schema.TopicSelector; +import org.apache.kafka.connect.source.SourceRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** TiDB source fetch task context. */ +public class TiDBSourceFetchTaskContext extends JdbcSourceFetchTaskContext { + + private static final Logger LOG = LoggerFactory.getLogger(TiDBSourceFetchTaskContext.class); + + private TiDBTaskContext tidbTaskContext; + + private final TiDBConnection connection; + private TiDBDatabaseSchema tiDBDatabaseSchema; + private EventOffsetContext offsetContext; + private SnapshotChangeEventSourceMetrics<TiDBPartition> snapshotChangeEventSourceMetrics; + private TopicSelector<TableId> topicSelector; + private JdbcSourceEventDispatcher<TiDBPartition> dispatcher; + private TiDBPartition tiDBPartition; + private ChangeEventQueue<DataChangeEvent> queue; + private ErrorHandler errorHandler; + private EventMetadataProvider metadataProvider; + + public TiDBSourceFetchTaskContext( + JdbcSourceConfig sourceConfig, + JdbcDataSourceDialect dataSourceDialect, + TiDBConnection connection) { + super(sourceConfig, dataSourceDialect); + this.connection = connection; + this.metadataProvider = new TiDBEventMetadataProvider(); + } + + @Override + public void configure(SourceSplitBase sourceSplitBase) { + final TiDBConnectorConfig connectorConfig = getDbzConnectorConfig(); + final boolean tableIdCaseInsensitive = + dataSourceDialect.isDataCollectionIdCaseSensitive(sourceConfig); Review Comment: **Polarity check please.** The result of `isDataCollectionIdCaseSensitive(...)` is stored into a variable named `tableIdCaseInsensitive` and passed on to `TiDBUtils.newSchema(...)`. The dialect side looks correct in this push, but if `newSchema` really expects a case-*insensitive* flag, the inversion has just moved to the call site — same end behavior. Could you double-check which polarity `newSchema` expects and rename accordingly? ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/source/fetch/EventSourceReader.java: ########## @@ -0,0 +1,499 @@ +/* + * 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.fetch; + +import org.apache.flink.cdc.connectors.base.relational.JdbcSourceEventDispatcher; +import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit; +import org.apache.flink.cdc.connectors.tidb.source.config.TiDBConnectorConfig; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffset; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetContext; +import org.apache.flink.cdc.connectors.tidb.utils.TableKeyRangeUtils; +import org.apache.flink.util.Preconditions; + +import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder; + +import io.debezium.connector.tidb.TiDBPartition; +import io.debezium.data.Envelope; +import io.debezium.pipeline.ErrorHandler; +import io.debezium.pipeline.source.spi.StreamingChangeEventSource; +import io.debezium.relational.TableId; +import io.debezium.relational.TableSchema; +import io.debezium.util.Clock; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.tikv.cdc.CDCClient; +import org.tikv.common.TiConfiguration; +import org.tikv.common.TiSession; +import org.tikv.common.key.RowKey; +import org.tikv.common.meta.TiColumnInfo; +import org.tikv.common.meta.TiTableInfo; +import org.tikv.kvproto.Cdcpb; +import org.tikv.kvproto.Coprocessor; +import org.tikv.shade.com.google.protobuf.ByteString; + +import java.io.Serializable; +import java.time.Instant; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.tikv.common.codec.TableCodec.decodeObjects; +import static org.tikv.common.codec.TiDBRowV2Decoder.decodeObjectsPreservingBinary; + +/** TiDB streaming change event source reader. */ +public class EventSourceReader + implements StreamingChangeEventSource<TiDBPartition, EventOffsetContext> { + private static final Logger LOG = LoggerFactory.getLogger(EventSourceReader.class); + private final StreamSplit split; + private final TiDBConnectorConfig connectorConfig; + private final TiConfiguration ticonf; + private final JdbcSourceEventDispatcher<TiDBPartition> eventDispatcher; + private final ErrorHandler errorHandler; + private final TiDBSourceFetchTaskContext taskContext; + private final Map<TableSchema, Map<String, Integer>> fieldIndexMap = new HashMap<>(); + public volatile ChangeEventSourceContext context; + + private static final long STREAMING_VERSION_START_EPOCH = 0L; + + /** Task local variables. */ + private transient TiSession session = null; + + private transient Coprocessor.KeyRange keyRange = null; + private transient CDCClient cdcClient = null; + private transient volatile long resolvedTs = -1L; + private transient TreeMap<RowKeyWithTs, Cdcpb.Event.Row> prewrites = null; + private transient TreeMap<RowKeyWithTs, Cdcpb.Event.Row> commits = null; + private transient BlockingQueue<Cdcpb.Event.Row> committedEvents = null; + private transient TableId tableId; + private transient TiTableInfo tableInfo; + + private transient volatile boolean running; + private transient volatile Thread executionThread; + private transient ExecutorService executorService; + private final AtomicBoolean closed = new AtomicBoolean(false); + + public EventSourceReader( + TiDBConnectorConfig connectorConfig, + JdbcSourceEventDispatcher eventDispatcher, + ErrorHandler errorHandler, + TiDBSourceFetchTaskContext taskContext, + StreamSplit split) { + this.connectorConfig = connectorConfig; + this.ticonf = connectorConfig.getSourceConfig().getTiConfiguration(); + this.eventDispatcher = eventDispatcher; + this.errorHandler = errorHandler; + this.taskContext = taskContext; + this.split = split; + } + + @Override + public synchronized void init() throws InterruptedException { + if (closed.get()) { + return; + } + StreamingChangeEventSource.super.init(); + try { + session = TiSession.create(ticonf); + Set<TableId> tableIds = this.split.getTableSchemas().keySet(); + if (tableIds.size() != 1) { + throw new IllegalStateException( + "Currently only single table ingest is supported, but found " + + tableIds.size() + + " tables."); + } + this.tableId = tableIds.stream().findFirst().get(); + this.tableInfo = session.getCatalog().getTable(tableId.catalog(), tableId.table()); + if (tableInfo == null) { + throw new RuntimeException( + String.format( + "Table %s.%s does not exist.", tableId.catalog(), tableId.table())); + } + keyRange = TableKeyRangeUtils.getTableKeyRange(tableInfo.getId(), 1, 0); + cdcClient = new CDCClient(session, keyRange); + prewrites = new TreeMap<>(); + commits = new TreeMap<>(); + // cdc event will lose if pull cdc event block when region split + // use queue to separate read and write to ensure pull event unblock. + // since sink jdbc is slow, 5000W queue size may be safe size. + committedEvents = new LinkedBlockingQueue<>(); + resolvedTs = EventOffset.getStartTs(this.split.getStartingOffset()); + ThreadFactory threadFactory = + new ThreadFactoryBuilder().setNameFormat("tidb-source-function-0").build(); + executorService = Executors.newSingleThreadExecutor(threadFactory); + } catch (RuntimeException e) { + close(); + throw e; + } + } + + @Override + public void execute( + ChangeEventSourceContext context, + TiDBPartition partition, + EventOffsetContext offsetContext) + throws InterruptedException { + if (closed.get()) { + return; + } + this.context = context; + this.executionThread = Thread.currentThread(); + running = true; + try { + if (connectorConfig.getSourceConfig().getStartupOptions().isSnapshotOnly()) { + LOG.info("Streaming is not enabled in current configuration"); + return; + } + this.taskContext.getDatabaseSchema().assureNonEmptySchema(); + cdcClient.start(resolvedTs); + EventOffsetContext effectiveOffsetContext = + offsetContext != null + ? offsetContext + : EventOffsetContext.initial(this.connectorConfig); + readChangeEvents(partition, effectiveOffsetContext); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + if (!closed.get()) { + throw e; + } + } catch (Exception e) { + if (!closed.get()) { + this.errorHandler.setProducerThrowable(e); + } + } finally { + running = false; + executionThread = null; + } + } + + protected void readChangeEvents(TiDBPartition partition, EventOffsetContext offsetContext) + throws Exception { + LOG.info("read change event from resolvedTs:{}", resolvedTs); + // child thread to sink committed rows. + executorService.execute( + () -> { + while (running && context.isRunning()) { + try { + Cdcpb.Event.Row committedRow = committedEvents.take(); + emitChangeEvent(partition, offsetContext, committedRow); + // use startTs of row as messageTs, use commitTs of row as fetchTs + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } catch (Exception e) { + if (running && context.isRunning()) { + LOG.error("Read change events error.", e); + } + } + } + }); + while (running && context.isRunning() && resolvedTs >= STREAMING_VERSION_START_EPOCH) { + for (int i = 0; i < 1000; i++) { + final Cdcpb.Event.Row row = cdcClient.get(); + if (row == null) { + break; + } + handleRow(row); + } + resolvedTs = cdcClient.getMaxResolvedTs(); + if (commits.size() > 0) { + flushRows(resolvedTs); + } + } + } + + protected void emitChangeEvent( + TiDBPartition partition, EventOffsetContext offsetContext, final Cdcpb.Event.Row row) + throws Exception { + if (!context.isRunning()) { + LOG.info("sourceContext is not running, skip emit change event."); + return; + } + if (tableId == null) { + LOG.warn("No valid tableId found, skipping log message: {}", row); + return; + } + TableSchema tableSchema = taskContext.getDatabaseSchema().schemaFor(tableId); + if (tableSchema == null) { + LOG.warn("No table schema found, skipping log message: {}", row); + return; + } + offsetContext.event(tableSchema.id(), Instant.ofEpochMilli(row.getCommitTs())); + Set<Integer> fieldIndex = fieldIndexConverter(tableInfo.getColumns(), tableSchema); + + Serializable[] before = null; + Serializable[] after = null; + final RowKey rowKey = RowKey.decode(row.getKey().toByteArray()); + final long handle = rowKey.getHandle(); + Envelope.Operation operation = getOperation(row); + switch (operation) { + case CREATE: + after = + (Serializable[]) + getSerializableObject( + handle, row.getValue(), tableInfo, fieldIndex); + break; + case UPDATE: + before = + (Serializable[]) + getSerializableObject( + handle, row.getOldValue(), tableInfo, fieldIndex); + after = + (Serializable[]) + getSerializableObject( + handle, row.getValue(), tableInfo, fieldIndex); + break; + case DELETE: + before = + (Serializable[]) + getSerializableObject( + handle, row.getOldValue(), tableInfo, fieldIndex); + + break; + default: + LOG.error("Row data opType is not supported,row:{}.", row); + } + eventDispatcher.dispatchDataChangeEvent( + partition, + tableSchema.id(), + new EventEmitter(partition, offsetContext, Clock.SYSTEM, operation, before, after)); + } + + private Object[] getSerializableObject( + long handle, final ByteString value, TiTableInfo tableInfo, Set<Integer> fieldIndex) { + Object[] serializableObject = new Serializable[fieldIndex.size()]; + try { + if (value == null) { + return null; + } + + Object[] tiKVValueAfter; + if (value != null && !value.isEmpty()) { + byte[] encodedValue = value.toByteArray(); + tiKVValueAfter = + Byte.toUnsignedInt(encodedValue[0]) == org.tikv.common.codec.RowV2.CODEC_VER + ? decodeObjectsPreservingBinary(encodedValue, handle, tableInfo) + : decodeObjects(encodedValue, handle, tableInfo); + } else { + return null; + } + for (int index : fieldIndex) { + serializableObject[index] = tiKVValueAfter[index]; + } + return serializableObject; + } catch (Exception e) { + LOG.error("decode object error", e); + return null; + } Review Comment: **Projection column indexing mixes two index spaces.** `serializableObject` is sized by `fieldIndex.size()`, but the copy loop uses `tiKVValueAfter[index]` where the decoded values are in TiKV physical column order. For projections / non-contiguous column sets these index spaces diverge: either AIOOBE, or silently reading the wrong physical column into the wrong schema slot — and the catch-all at the bottom converts any failure into `return null`, corrupting before/after images silently (see also the swallowed-exceptions comment above). Could you add an explicit physical-offset → schema-position mapping here, plus a unit test with a projected/non-contiguous column subset? ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/source/offset/EventOffsetContext.java: ########## @@ -0,0 +1,212 @@ +/* + * 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.offset; + +import org.apache.flink.cdc.connectors.tidb.source.config.TiDBConnectorConfig; + +import io.debezium.connector.AbstractSourceInfo; +import io.debezium.connector.SnapshotRecord; +import io.debezium.connector.mysql.MySqlReadOnlyIncrementalSnapshotContext; +import io.debezium.pipeline.source.snapshot.incremental.IncrementalSnapshotContext; +import io.debezium.pipeline.source.snapshot.incremental.SignalBasedIncrementalSnapshotContext; +import io.debezium.pipeline.spi.OffsetContext; +import io.debezium.pipeline.txmetadata.TransactionContext; +import io.debezium.relational.TableId; +import io.debezium.schema.DataCollectionId; +import org.apache.kafka.connect.data.Schema; +import org.apache.kafka.connect.data.Struct; +import org.tikv.common.meta.TiTimestamp; + +import java.time.Instant; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; + +import static org.apache.flink.cdc.connectors.tidb.source.offset.EventOffset.COMMIT_VERSION_KEY; +import static org.apache.flink.cdc.connectors.tidb.source.offset.EventOffset.TIMESTAMP_KEY; + +/** The offset context for TiDB connector. */ +public class EventOffsetContext implements OffsetContext { + private static final String SNAPSHOT_COMPLETED_KEY = "snapshot_completed"; + + private final Schema sourceInfoSchema; + private final TiDBSourceInfo sourceInfo; + private final TransactionContext transactionContext; + private final IncrementalSnapshotContext<TableId> incrementalSnapshotContext; + private boolean snapshotCompleted; + private String commitVersion; + private String timestamp; + + public EventOffsetContext( + boolean snapshot, + boolean snapshotCompleted, + TransactionContext transactionContext, + IncrementalSnapshotContext<TableId> incrementalSnapshotContext, + TiDBSourceInfo sourceInfo) { + this.sourceInfo = sourceInfo; + this.sourceInfoSchema = sourceInfo.schema(); + this.snapshotCompleted = snapshotCompleted; + + this.transactionContext = transactionContext; + this.incrementalSnapshotContext = incrementalSnapshotContext; + + if (this.snapshotCompleted) { + postSnapshotCompletion(); + } else { + sourceInfo.setSnapshot(snapshot ? SnapshotRecord.TRUE : SnapshotRecord.FALSE); + } + } + + public static EventOffsetContext initial(TiDBConnectorConfig config) { + return new EventOffsetContext( + false, + false, + new TransactionContext(), + new SignalBasedIncrementalSnapshotContext<>(), + new TiDBSourceInfo(config)); + } + + @Override + public Map<String, ?> getOffset() { + HashMap<String, Object> offset = new HashMap<>(); + if (timestamp != null) { + offset.put(TIMESTAMP_KEY, timestamp); + } + + if (commitVersion != null) { + offset.put(COMMIT_VERSION_KEY, commitVersion); + } + if (sourceInfo.isSnapshot()) { + if (!snapshotCompleted) { + offset.put(AbstractSourceInfo.SNAPSHOT_KEY, true); + } + return offset; + } else { + return incrementalSnapshotContext.store(transactionContext.store(offset)); + } + } + + public void databaseEvent(String database, Instant timestamp) { + sourceInfo.setSourceTime(timestamp); + sourceInfo.databaseEvent(database); + sourceInfo.tableEvent((TableId) null); + } + + public void tableEvent(String database, Set<TableId> tableIds, Instant timestamp) { + sourceInfo.setSourceTime(timestamp); + sourceInfo.databaseEvent(database); + sourceInfo.tableEvent(tableIds); + } + + @Override + public Schema getSourceInfoSchema() { + return sourceInfoSchema; + } + + @Override + public Struct getSourceInfo() { + return sourceInfo.struct(); + } + + @Override + public boolean isSnapshotRunning() { + return sourceInfo.isSnapshot() && !snapshotCompleted; + } + + @Override + public void markLastSnapshotRecord() { + sourceInfo.setSnapshot(SnapshotRecord.LAST); + } + + @Override + public void preSnapshotStart() { + sourceInfo.setSnapshot(SnapshotRecord.TRUE); + snapshotCompleted = false; + } + + @Override + public void preSnapshotCompletion() { + snapshotCompleted = true; + } + + @Override + public void postSnapshotCompletion() { + snapshotCompleted = true; + } + + @Override + public void event(DataCollectionId collectionId, Instant timestamp) { + sourceInfo.setSourceTime(timestamp); + sourceInfo.tableEvent((TableId) collectionId); + } + + @Override + public TransactionContext getTransactionContext() { + return transactionContext; + } + + public void setCheckpoint(Instant timestamp, String commitVersion) { + this.timestamp = String.valueOf(timestamp.toEpochMilli()); + if (commitVersion == null) { + commitVersion = + String.valueOf(new TiTimestamp(timestamp.toEpochMilli(), 0).getVersion()); + } + this.commitVersion = commitVersion; + } + + /** The loader for TiDB offset context. */ + public static class Loader implements OffsetContext.Loader<EventOffsetContext> { + + private final TiDBConnectorConfig connectorConfig; + + public Loader(TiDBConnectorConfig connectorConfig) { + this.connectorConfig = connectorConfig; + } + + @SuppressWarnings("unchecked") + @Override + public EventOffsetContext load(Map<String, ?> offset) { + boolean snapshot = + Boolean.TRUE.equals(offset.get(TiDBSourceInfo.SNAPSHOT_KEY)) + || "true".equals(offset.get(TiDBSourceInfo.SNAPSHOT_KEY)); + boolean snapshotCompleted = + Boolean.TRUE.equals(offset.get(SNAPSHOT_COMPLETED_KEY)) + || "true".equals(offset.get(SNAPSHOT_COMPLETED_KEY)); + IncrementalSnapshotContext<TableId> incrementalSnapshotContext; + if (connectorConfig.isReadOnlyConnection()) { + incrementalSnapshotContext = MySqlReadOnlyIncrementalSnapshotContext.load(offset); + } else { + incrementalSnapshotContext = SignalBasedIncrementalSnapshotContext.load(offset); + } + final EventOffsetContext offsetContext = + new EventOffsetContext( + snapshot, + snapshotCompleted, + TransactionContext.load(offset), + incrementalSnapshotContext, + new TiDBSourceInfo(connectorConfig)); + String timestamp = (String) offset.get(TIMESTAMP_KEY); + offsetContext.setCheckpoint( + timestamp == null + ? Instant.now() + : Instant.ofEpochMilli(Long.parseLong(timestamp)), + (String) offset.get(COMMIT_VERSION_KEY)); Review Comment: **Offset load/store asymmetry.** A missing timestamp silently falls back to `Instant.now()`, so the restored position depends on recovery wall-clock time rather than persisted state — and `getOffset()` never writes the `snapshot_completed` key that this loader reads (L187-189). Stored and loaded fields should be strictly symmetric, and a malformed offset should fail deterministically instead of jumping to "now". ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/source/fetch/EventSourceReader.java: ########## @@ -0,0 +1,499 @@ +/* + * 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.fetch; + +import org.apache.flink.cdc.connectors.base.relational.JdbcSourceEventDispatcher; +import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit; +import org.apache.flink.cdc.connectors.tidb.source.config.TiDBConnectorConfig; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffset; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetContext; +import org.apache.flink.cdc.connectors.tidb.utils.TableKeyRangeUtils; +import org.apache.flink.util.Preconditions; + +import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder; + +import io.debezium.connector.tidb.TiDBPartition; +import io.debezium.data.Envelope; +import io.debezium.pipeline.ErrorHandler; +import io.debezium.pipeline.source.spi.StreamingChangeEventSource; +import io.debezium.relational.TableId; +import io.debezium.relational.TableSchema; +import io.debezium.util.Clock; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.tikv.cdc.CDCClient; +import org.tikv.common.TiConfiguration; +import org.tikv.common.TiSession; +import org.tikv.common.key.RowKey; +import org.tikv.common.meta.TiColumnInfo; +import org.tikv.common.meta.TiTableInfo; +import org.tikv.kvproto.Cdcpb; +import org.tikv.kvproto.Coprocessor; +import org.tikv.shade.com.google.protobuf.ByteString; + +import java.io.Serializable; +import java.time.Instant; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.tikv.common.codec.TableCodec.decodeObjects; +import static org.tikv.common.codec.TiDBRowV2Decoder.decodeObjectsPreservingBinary; + +/** TiDB streaming change event source reader. */ +public class EventSourceReader + implements StreamingChangeEventSource<TiDBPartition, EventOffsetContext> { + private static final Logger LOG = LoggerFactory.getLogger(EventSourceReader.class); + private final StreamSplit split; + private final TiDBConnectorConfig connectorConfig; + private final TiConfiguration ticonf; + private final JdbcSourceEventDispatcher<TiDBPartition> eventDispatcher; + private final ErrorHandler errorHandler; + private final TiDBSourceFetchTaskContext taskContext; + private final Map<TableSchema, Map<String, Integer>> fieldIndexMap = new HashMap<>(); + public volatile ChangeEventSourceContext context; + + private static final long STREAMING_VERSION_START_EPOCH = 0L; + + /** Task local variables. */ + private transient TiSession session = null; + + private transient Coprocessor.KeyRange keyRange = null; + private transient CDCClient cdcClient = null; + private transient volatile long resolvedTs = -1L; + private transient TreeMap<RowKeyWithTs, Cdcpb.Event.Row> prewrites = null; + private transient TreeMap<RowKeyWithTs, Cdcpb.Event.Row> commits = null; + private transient BlockingQueue<Cdcpb.Event.Row> committedEvents = null; + private transient TableId tableId; + private transient TiTableInfo tableInfo; + + private transient volatile boolean running; + private transient volatile Thread executionThread; + private transient ExecutorService executorService; + private final AtomicBoolean closed = new AtomicBoolean(false); + + public EventSourceReader( + TiDBConnectorConfig connectorConfig, + JdbcSourceEventDispatcher eventDispatcher, + ErrorHandler errorHandler, + TiDBSourceFetchTaskContext taskContext, + StreamSplit split) { + this.connectorConfig = connectorConfig; + this.ticonf = connectorConfig.getSourceConfig().getTiConfiguration(); + this.eventDispatcher = eventDispatcher; + this.errorHandler = errorHandler; + this.taskContext = taskContext; + this.split = split; + } + + @Override + public synchronized void init() throws InterruptedException { + if (closed.get()) { + return; + } + StreamingChangeEventSource.super.init(); + try { + session = TiSession.create(ticonf); + Set<TableId> tableIds = this.split.getTableSchemas().keySet(); + if (tableIds.size() != 1) { + throw new IllegalStateException( + "Currently only single table ingest is supported, but found " + + tableIds.size() + + " tables."); + } + this.tableId = tableIds.stream().findFirst().get(); + this.tableInfo = session.getCatalog().getTable(tableId.catalog(), tableId.table()); + if (tableInfo == null) { + throw new RuntimeException( + String.format( + "Table %s.%s does not exist.", tableId.catalog(), tableId.table())); + } + keyRange = TableKeyRangeUtils.getTableKeyRange(tableInfo.getId(), 1, 0); + cdcClient = new CDCClient(session, keyRange); + prewrites = new TreeMap<>(); + commits = new TreeMap<>(); + // cdc event will lose if pull cdc event block when region split + // use queue to separate read and write to ensure pull event unblock. + // since sink jdbc is slow, 5000W queue size may be safe size. + committedEvents = new LinkedBlockingQueue<>(); + resolvedTs = EventOffset.getStartTs(this.split.getStartingOffset()); + ThreadFactory threadFactory = + new ThreadFactoryBuilder().setNameFormat("tidb-source-function-0").build(); + executorService = Executors.newSingleThreadExecutor(threadFactory); + } catch (RuntimeException e) { + close(); + throw e; + } + } + + @Override + public void execute( + ChangeEventSourceContext context, + TiDBPartition partition, + EventOffsetContext offsetContext) + throws InterruptedException { + if (closed.get()) { + return; + } + this.context = context; + this.executionThread = Thread.currentThread(); + running = true; + try { + if (connectorConfig.getSourceConfig().getStartupOptions().isSnapshotOnly()) { + LOG.info("Streaming is not enabled in current configuration"); + return; + } + this.taskContext.getDatabaseSchema().assureNonEmptySchema(); + cdcClient.start(resolvedTs); + EventOffsetContext effectiveOffsetContext = + offsetContext != null + ? offsetContext + : EventOffsetContext.initial(this.connectorConfig); + readChangeEvents(partition, effectiveOffsetContext); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + if (!closed.get()) { + throw e; + } + } catch (Exception e) { + if (!closed.get()) { + this.errorHandler.setProducerThrowable(e); + } + } finally { + running = false; + executionThread = null; + } + } + + protected void readChangeEvents(TiDBPartition partition, EventOffsetContext offsetContext) + throws Exception { + LOG.info("read change event from resolvedTs:{}", resolvedTs); + // child thread to sink committed rows. + executorService.execute( + () -> { + while (running && context.isRunning()) { + try { + Cdcpb.Event.Row committedRow = committedEvents.take(); + emitChangeEvent(partition, offsetContext, committedRow); + // use startTs of row as messageTs, use commitTs of row as fetchTs + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } catch (Exception e) { + if (running && context.isRunning()) { + LOG.error("Read change events error.", e); + } + } + } + }); + while (running && context.isRunning() && resolvedTs >= STREAMING_VERSION_START_EPOCH) { + for (int i = 0; i < 1000; i++) { + final Cdcpb.Event.Row row = cdcClient.get(); + if (row == null) { + break; + } + handleRow(row); + } + resolvedTs = cdcClient.getMaxResolvedTs(); + if (commits.size() > 0) { + flushRows(resolvedTs); + } + } + } + + protected void emitChangeEvent( + TiDBPartition partition, EventOffsetContext offsetContext, final Cdcpb.Event.Row row) + throws Exception { + if (!context.isRunning()) { + LOG.info("sourceContext is not running, skip emit change event."); + return; + } + if (tableId == null) { + LOG.warn("No valid tableId found, skipping log message: {}", row); + return; + } + TableSchema tableSchema = taskContext.getDatabaseSchema().schemaFor(tableId); + if (tableSchema == null) { + LOG.warn("No table schema found, skipping log message: {}", row); + return; + } + offsetContext.event(tableSchema.id(), Instant.ofEpochMilli(row.getCommitTs())); + Set<Integer> fieldIndex = fieldIndexConverter(tableInfo.getColumns(), tableSchema); + + Serializable[] before = null; + Serializable[] after = null; + final RowKey rowKey = RowKey.decode(row.getKey().toByteArray()); + final long handle = rowKey.getHandle(); + Envelope.Operation operation = getOperation(row); + switch (operation) { + case CREATE: + after = + (Serializable[]) + getSerializableObject( + handle, row.getValue(), tableInfo, fieldIndex); + break; + case UPDATE: + before = + (Serializable[]) + getSerializableObject( + handle, row.getOldValue(), tableInfo, fieldIndex); + after = + (Serializable[]) + getSerializableObject( + handle, row.getValue(), tableInfo, fieldIndex); + break; + case DELETE: + before = + (Serializable[]) + getSerializableObject( + handle, row.getOldValue(), tableInfo, fieldIndex); + + break; + default: + LOG.error("Row data opType is not supported,row:{}.", row); + } + eventDispatcher.dispatchDataChangeEvent( + partition, + tableSchema.id(), + new EventEmitter(partition, offsetContext, Clock.SYSTEM, operation, before, after)); + } + + private Object[] getSerializableObject( + long handle, final ByteString value, TiTableInfo tableInfo, Set<Integer> fieldIndex) { + Object[] serializableObject = new Serializable[fieldIndex.size()]; + try { + if (value == null) { + return null; + } + + Object[] tiKVValueAfter; + if (value != null && !value.isEmpty()) { + byte[] encodedValue = value.toByteArray(); + tiKVValueAfter = + Byte.toUnsignedInt(encodedValue[0]) == org.tikv.common.codec.RowV2.CODEC_VER + ? decodeObjectsPreservingBinary(encodedValue, handle, tableInfo) + : decodeObjects(encodedValue, handle, tableInfo); + } else { + return null; + } + for (int index : fieldIndex) { + serializableObject[index] = tiKVValueAfter[index]; + } + return serializableObject; + } catch (Exception e) { + LOG.error("decode object error", e); + return null; + } + } + + private Set<Integer> fieldIndexConverter( + List<TiColumnInfo> tiColumnInfos, TableSchema tableSchema) { + Map<String, Integer> fieldIndex = + fieldIndexMap.computeIfAbsent( + tableSchema, + schema -> + IntStream.range(0, schema.valueSchema().fields().size()) + .boxed() + .collect( + Collectors.toMap( + i -> + schema.valueSchema() + .fields() + .get(i) + .name(), + i -> i))); + Set<Integer> fieldIndexSet = new HashSet<>(); + for (TiColumnInfo tiColumnInfo : tiColumnInfos) { + if (fieldIndex.keySet().stream() + .anyMatch(key -> key.equalsIgnoreCase(tiColumnInfo.getName()))) { + fieldIndexSet.add(tiColumnInfo.getOffset()); + } + } + return fieldIndexSet; + } + + private Envelope.Operation getOperation(final Cdcpb.Event.Row row) { + if (row.getOpType() == Cdcpb.Event.Row.OpType.PUT) { // create ，update + if (row.getValue() != null && !row.getOldValue().isEmpty()) { + return Envelope.Operation.UPDATE; + } else { + return Envelope.Operation.CREATE; + } + } else if (row.getOpType() == Cdcpb.Event.Row.OpType.DELETE) { // delete + return Envelope.Operation.DELETE; + } else { + LOG.error("Row data opType is not supported,row:{}.", row); + return null; + } + } + + protected void flushRows(final long timestamp) throws Exception { + Preconditions.checkState(context != null, "sourceContext shouldn't be null"); + synchronized (context) { + while (!commits.isEmpty() && commits.firstKey().timestamp <= timestamp) { + final Cdcpb.Event.Row commitRow = commits.pollFirstEntry().getValue(); + final Cdcpb.Event.Row prewriteRow = + prewrites.remove(RowKeyWithTs.ofStart(commitRow)); + // if pull cdc event block when region split, cdc event will lose. + committedEvents.offer(prewriteRow); Review Comment: **`prewrites.remove(...)` can return `null`** — reconnects, region transitions, or a COMMIT arriving without its PREWRITE — and `LinkedBlockingQueue.offer(null)` throws NPE, which would kill the reader inside `flushRows`. Worth defining the expected PREWRITE/COMMIT guarantees across reconnects (the comment above already hints events can be lost during region splits) and handling the missing-prewrite case explicitly (skip + warn, or fail, but not NPE). ########## flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/source/fetch/EventSourceReader.java: ########## @@ -0,0 +1,499 @@ +/* + * 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.fetch; + +import org.apache.flink.cdc.connectors.base.relational.JdbcSourceEventDispatcher; +import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit; +import org.apache.flink.cdc.connectors.tidb.source.config.TiDBConnectorConfig; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffset; +import org.apache.flink.cdc.connectors.tidb.source.offset.EventOffsetContext; +import org.apache.flink.cdc.connectors.tidb.utils.TableKeyRangeUtils; +import org.apache.flink.util.Preconditions; + +import org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder; + +import io.debezium.connector.tidb.TiDBPartition; +import io.debezium.data.Envelope; +import io.debezium.pipeline.ErrorHandler; +import io.debezium.pipeline.source.spi.StreamingChangeEventSource; +import io.debezium.relational.TableId; +import io.debezium.relational.TableSchema; +import io.debezium.util.Clock; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.tikv.cdc.CDCClient; +import org.tikv.common.TiConfiguration; +import org.tikv.common.TiSession; +import org.tikv.common.key.RowKey; +import org.tikv.common.meta.TiColumnInfo; +import org.tikv.common.meta.TiTableInfo; +import org.tikv.kvproto.Cdcpb; +import org.tikv.kvproto.Coprocessor; +import org.tikv.shade.com.google.protobuf.ByteString; + +import java.io.Serializable; +import java.time.Instant; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.tikv.common.codec.TableCodec.decodeObjects; +import static org.tikv.common.codec.TiDBRowV2Decoder.decodeObjectsPreservingBinary; + +/** TiDB streaming change event source reader. */ +public class EventSourceReader + implements StreamingChangeEventSource<TiDBPartition, EventOffsetContext> { + private static final Logger LOG = LoggerFactory.getLogger(EventSourceReader.class); + private final StreamSplit split; + private final TiDBConnectorConfig connectorConfig; + private final TiConfiguration ticonf; + private final JdbcSourceEventDispatcher<TiDBPartition> eventDispatcher; + private final ErrorHandler errorHandler; + private final TiDBSourceFetchTaskContext taskContext; + private final Map<TableSchema, Map<String, Integer>> fieldIndexMap = new HashMap<>(); + public volatile ChangeEventSourceContext context; + + private static final long STREAMING_VERSION_START_EPOCH = 0L; + + /** Task local variables. */ + private transient TiSession session = null; + + private transient Coprocessor.KeyRange keyRange = null; + private transient CDCClient cdcClient = null; + private transient volatile long resolvedTs = -1L; + private transient TreeMap<RowKeyWithTs, Cdcpb.Event.Row> prewrites = null; + private transient TreeMap<RowKeyWithTs, Cdcpb.Event.Row> commits = null; + private transient BlockingQueue<Cdcpb.Event.Row> committedEvents = null; + private transient TableId tableId; + private transient TiTableInfo tableInfo; + + private transient volatile boolean running; + private transient volatile Thread executionThread; + private transient ExecutorService executorService; + private final AtomicBoolean closed = new AtomicBoolean(false); + + public EventSourceReader( + TiDBConnectorConfig connectorConfig, + JdbcSourceEventDispatcher eventDispatcher, + ErrorHandler errorHandler, + TiDBSourceFetchTaskContext taskContext, + StreamSplit split) { + this.connectorConfig = connectorConfig; + this.ticonf = connectorConfig.getSourceConfig().getTiConfiguration(); + this.eventDispatcher = eventDispatcher; + this.errorHandler = errorHandler; + this.taskContext = taskContext; + this.split = split; + } + + @Override + public synchronized void init() throws InterruptedException { + if (closed.get()) { + return; + } + StreamingChangeEventSource.super.init(); + try { + session = TiSession.create(ticonf); + Set<TableId> tableIds = this.split.getTableSchemas().keySet(); + if (tableIds.size() != 1) { + throw new IllegalStateException( + "Currently only single table ingest is supported, but found " + + tableIds.size() + + " tables."); + } Review Comment: **Multi-table jobs fail late.** The dialect/config accept multi-table table-lists, and snapshotting works for them — but this check throws at runtime when entering the streaming phase, so such a job would snapshot for hours on a large table and only then die. Either support multi-table ranges in the stream reader, or fail fast at configuration/validation time with a documented limitation. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
