This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 94b4ac59dd [Fix][Connector-V2] Fix TiDB CDC silent row event loss when
resolvedT… (#11113)
94b4ac59dd is described below
commit 94b4ac59ddcdb57524d455a9178de53b6cff395c
Author: zhaoysg <[email protected]>
AuthorDate: Sun Jul 12 20:12:56 2026 +0800
[Fix][Connector-V2] Fix TiDB CDC silent row event loss when resolvedT…
(#11113)
Co-authored-by: Cursor <[email protected]>
Co-authored-by: David Zollo <[email protected]>
---
.../enumerator/TiDBSourceCheckpointState.java | 68 ++++-
.../enumerator/TiDBSourceSplitEnumerator.java | 56 +++--
.../cdc/tidb/source/reader/TiDBSourceReader.java | 277 +++++++++++++++++++--
.../cdc/tidb/source/utils/TableKeyRangeUtils.java | 19 +-
.../enumerator/TiDBSourceSplitEnumeratorTest.java | 183 ++++++++++++++
.../tidb/source/reader/TiDBSourceReaderTest.java | 121 +++++++++
6 files changed, 662 insertions(+), 62 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/enumerator/TiDBSourceCheckpointState.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/enumerator/TiDBSourceCheckpointState.java
index 6869f73ea8..2c796343a4 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/enumerator/TiDBSourceCheckpointState.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/enumerator/TiDBSourceCheckpointState.java
@@ -19,20 +19,78 @@ package
org.apache.seatunnel.connectors.seatunnel.cdc.tidb.source.enumerator;
import
org.apache.seatunnel.connectors.seatunnel.cdc.tidb.source.split.TiDBSourceSplit;
-import lombok.AllArgsConstructor;
import lombok.Getter;
-import lombok.Setter;
import lombok.ToString;
import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
import java.util.Map;
@Getter
-@Setter
-@AllArgsConstructor
@ToString
public class TiDBSourceCheckpointState implements Serializable {
private static final long serialVersionUID = 6292978509042158791L;
private boolean shouldEnumerate;
- private Map<Integer, TiDBSourceSplit> pendingSplit;
+ private Map<Integer, List<TiDBSourceSplit>> pendingSplit;
+
+ public TiDBSourceCheckpointState(boolean shouldEnumerate, Map<Integer, ?>
pendingSplit) {
+ this.shouldEnumerate = shouldEnumerate;
+ this.pendingSplit = normalizePendingSplit(pendingSplit);
+ }
+
+ public void setShouldEnumerate(boolean shouldEnumerate) {
+ this.shouldEnumerate = shouldEnumerate;
+ }
+
+ public void setPendingSplit(Map<Integer, ?> pendingSplit) {
+ this.pendingSplit = normalizePendingSplit(pendingSplit);
+ }
+
+ public Map<Integer, List<TiDBSourceSplit>> getPendingSplit() {
+ pendingSplit = normalizePendingSplit(pendingSplit);
+ return pendingSplit;
+ }
+
+ private static Map<Integer, List<TiDBSourceSplit>> normalizePendingSplit(
+ Map<Integer, ?> pendingSplit) {
+ Map<Integer, List<TiDBSourceSplit>> normalizedPendingSplit = new
HashMap<>();
+ if (pendingSplit == null) {
+ return normalizedPendingSplit;
+ }
+ for (Map.Entry<Integer, ?> entry : pendingSplit.entrySet()) {
+ Object value = entry.getValue();
+ if (value instanceof TiDBSourceSplit) {
+ normalizedPendingSplit.put(
+ entry.getKey(),
+ new
ArrayList<>(Collections.singletonList((TiDBSourceSplit) value)));
+ } else if (value instanceof List) {
+ normalizedPendingSplit.put(entry.getKey(),
copyPendingSplits((List<?>) value));
+ } else if (value == null) {
+ normalizedPendingSplit.put(entry.getKey(), new ArrayList<>());
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported pending split value type %s for
reader %s.",
+ value.getClass().getName(), entry.getKey()));
+ }
+ }
+ return normalizedPendingSplit;
+ }
+
+ private static List<TiDBSourceSplit> copyPendingSplits(List<?>
pendingSplits) {
+ List<TiDBSourceSplit> copiedPendingSplits = new ArrayList<>();
+ for (Object pendingSplit : pendingSplits) {
+ if (!(pendingSplit instanceof TiDBSourceSplit)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported pending split list value type
%s.",
+ pendingSplit == null ? "null" :
pendingSplit.getClass().getName()));
+ }
+ copiedPendingSplits.add((TiDBSourceSplit) pendingSplit);
+ }
+ return copiedPendingSplits;
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/enumerator/TiDBSourceSplitEnumerator.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/enumerator/TiDBSourceSplitEnumerator.java
index fc4b7dd332..9adb1d700a 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/enumerator/TiDBSourceSplitEnumerator.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/enumerator/TiDBSourceSplitEnumerator.java
@@ -17,8 +17,6 @@
package org.apache.seatunnel.connectors.seatunnel.cdc.tidb.source.enumerator;
-import org.apache.seatunnel.shade.com.google.common.collect.Lists;
-
import org.apache.seatunnel.api.source.SourceSplitEnumerator;
import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
import
org.apache.seatunnel.connectors.seatunnel.cdc.tidb.source.config.TiDBSourceConfig;
@@ -32,6 +30,7 @@ import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -43,9 +42,10 @@ import java.util.Set;
public class TiDBSourceSplitEnumerator
implements SourceSplitEnumerator<TiDBSourceSplit,
TiDBSourceCheckpointState> {
+ private static final String CDC_DIAG_PREFIX = "[TiDB-CDC-DIAG]";
+
private final TiDBSourceConfig sourceConfig;
- private final Map<Integer, TiDBSourceSplit> assignedSplit;
- private final Map<Integer, TiDBSourceSplit> pendingSplit;
+ private final Map<Integer, List<TiDBSourceSplit>> pendingSplit;
private final Context<TiDBSourceSplit> context;
private TiSession tiSession;
private long tableId;
@@ -65,7 +65,6 @@ public class TiDBSourceSplitEnumerator
TiDBSourceCheckpointState restoreState) {
this.context = context;
this.sourceConfig = sourceConfig;
- this.assignedSplit = new HashMap<>();
this.pendingSplit = new HashMap<>();
this.shouldEnumerate = (restoreState == null);
if (restoreState != null) {
@@ -82,6 +81,15 @@ public class TiDBSourceSplitEnumerator
.getCatalog()
.getTable(sourceConfig.getDatabaseName(),
sourceConfig.getTableName())
.getId();
+ log.info(
+ "{} Enumerator opened, database={}, table={}, tableId={},
startupMode={},"
+ + " parallelism={}.",
+ CDC_DIAG_PREFIX,
+ sourceConfig.getDatabaseName(),
+ sourceConfig.getTableName(),
+ tableId,
+ sourceConfig.getStartupMode(),
+ context.currentParallelism());
}
/** The method is executed by the engine only once. */
@@ -90,9 +98,14 @@ public class TiDBSourceSplitEnumerator
Set<Integer> readers = context.registeredReaders();
if (shouldEnumerate) {
List<TiDBSourceSplit> sourceSplits = getTiDBSourceSplit();
+ log.info(
+ "{} Enumerated TiDB CDC splits, database={}, table={},
splitCount={}.",
+ CDC_DIAG_PREFIX,
+ sourceConfig.getDatabaseName(),
+ sourceConfig.getTableName(),
+ sourceSplits.size());
synchronized (stateLock) {
addPendingSplit(sourceSplits);
- fetchAssignedSplit();
shouldEnumerate = false;
assignSplit(readers);
}
@@ -102,28 +115,27 @@ public class TiDBSourceSplitEnumerator
readers.forEach(context::signalNoMoreSplits);
}
- private void fetchAssignedSplit() {
- for (Map.Entry<Integer, TiDBSourceSplit> split :
pendingSplit.entrySet()) {
- if (assignedSplit.containsKey(split.getKey())) {
- // override split
- pendingSplit.put(split.getKey(), split.getValue());
- }
- }
- }
-
private synchronized void addPendingSplit(List<TiDBSourceSplit> splits) {
splits.forEach(
split -> {
- pendingSplit.put(
- getSplitOwner(split.splitId(),
context.currentParallelism()), split);
+ pendingSplit
+ .computeIfAbsent(
+ getSplitOwner(split.splitId(),
context.currentParallelism()),
+ ignored -> new ArrayList<>())
+ .add(split);
});
}
private void assignSplit(Collection<Integer> readers) {
for (Integer reader : readers) {
- final TiDBSourceSplit assignmentForReader =
pendingSplit.remove(reader);
- if (assignmentForReader != null) {
- log.debug("Assign splits {} to reader {}",
assignmentForReader, reader);
+ final List<TiDBSourceSplit> assignmentForReader =
pendingSplit.remove(reader);
+ if (assignmentForReader != null && !assignmentForReader.isEmpty())
{
+ log.info(
+ "{} Assign split to reader, reader={}, splits={},
remainingPending={}.",
+ CDC_DIAG_PREFIX,
+ reader,
+ assignmentForReader,
+ currentUnassignedSplitSize());
context.assignSplit(reader, assignmentForReader);
}
}
@@ -134,7 +146,7 @@ public class TiDBSourceSplitEnumerator
}
private List<TiDBSourceSplit> getTiDBSourceSplit() {
- List<TiDBSourceSplit> sourceSplits = Lists.newArrayList();
+ List<TiDBSourceSplit> sourceSplits = new ArrayList<>();
List<Coprocessor.KeyRange> keyRanges =
TableKeyRangeUtils.getTableKeyRanges(this.tableId,
context.currentParallelism());
for (Coprocessor.KeyRange keyRange : keyRanges) {
@@ -190,7 +202,7 @@ public class TiDBSourceSplitEnumerator
@Override
public int currentUnassignedSplitSize() {
- return pendingSplit.size();
+ return pendingSplit.values().stream().mapToInt(List::size).sum();
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/reader/TiDBSourceReader.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/reader/TiDBSourceReader.java
index 2c2d71a54e..30b4b412ff 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/reader/TiDBSourceReader.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/reader/TiDBSourceReader.java
@@ -30,13 +30,15 @@ import
org.apache.seatunnel.connectors.seatunnel.cdc.tidb.source.utils.TableKeyR
import org.tikv.cdc.CDCClient;
import org.tikv.common.TiSession;
-import org.tikv.common.key.RowKey;
+import org.tikv.common.key.Key;
import org.tikv.common.meta.TiTableInfo;
+import org.tikv.common.region.RegionStoreClient;
+import org.tikv.common.region.TiRegion;
+import org.tikv.common.util.ConcreteBackOffer;
import org.tikv.kvproto.Cdcpb;
import org.tikv.kvproto.Coprocessor;
import org.tikv.kvproto.Kvrpcpb;
import org.tikv.shade.com.google.protobuf.ByteString;
-import org.tikv.txn.KVClient;
import lombok.extern.slf4j.Slf4j;
@@ -53,6 +55,12 @@ import java.util.concurrent.LinkedBlockingQueue;
@Slf4j
public class TiDBSourceReader implements SourceReader<SeaTunnelRow,
TiDBSourceSplit> {
+ private static final String CDC_DIAG_PREFIX = "[TiDB-CDC-DIAG]";
+ private static final long STREAMING_STATS_LOG_INTERVAL_MS = 10_000L;
+ private static final long SLOW_STREAMING_BATCH_MS = 1_000L;
+ private static final int METADATA_DRAIN_ATTEMPT_MULTIPLIER = 10;
+ private static final long EMPTY_POLL_BREAK_THRESHOLD_MS = 50L;
+
private final SourceReader.Context context;
private final TiDBSourceConfig config;
private final List<TiDBSourceSplit> sourceSplits;
@@ -70,6 +78,11 @@ public class TiDBSourceReader implements
SourceReader<SeaTunnelRow, TiDBSourceSp
private CatalogTable catalogTable;
+ private long lastStreamingStatsLogTime;
+ private long totalPolledRows;
+ private long totalCommittedRows;
+ private long totalEmittedRows;
+
public TiDBSourceReader(Context context, TiDBSourceConfig config,
CatalogTable catalogTable) {
this.context = context;
this.config = config;
@@ -96,6 +109,16 @@ public class TiDBSourceReader implements
SourceReader<SeaTunnelRow, TiDBSourceSp
new SeaTunnelRowSnapshotRecordDeserializer(tableInfo,
catalogTable);
this.streamingRecordDeserializer =
new SeaTunnelRowStreamingRecordDeserializer(tableInfo,
catalogTable);
+ log.info(
+ "{} Reader opened, database={}, table={}, startupMode={},
batchSize={},"
+ + " scanTimeout={}, requestTimeout={}.",
+ CDC_DIAG_PREFIX,
+ config.getDatabaseName(),
+ config.getTableName(),
+ config.getStartupMode(),
+ config.getBatchSize(),
+ config.getTiConfiguration().getScanTimeout(),
+ config.getTiConfiguration().getTimeout());
}
/**
@@ -140,55 +163,144 @@ public class TiDBSourceReader implements
SourceReader<SeaTunnelRow, TiDBSourceSp
throws Exception {
log.info(String.format("[%s] Snapshot events start.",
split.splitId()));
Coprocessor.KeyRange keyRange = split.getKeyRange();
- try (KVClient scanClient = session.createKVClient()) {
- // start timestamp
- long startTs = session.getTimestamp().getVersion();
- ByteString start = split.getSnapshotStart();
- while (true) {
+ // start timestamp
+ long startTs = session.getTimestamp().getVersion();
+ ByteString start = split.getSnapshotStart();
+ Key end = Key.toRawKey(keyRange.getEnd());
+ long scannedCount = 0;
+ long emittedCount = 0;
+ while (Key.toRawKey(start).compareTo(end) < 0) {
+ try (RegionStoreClient scanClient =
+ session.getRegionStoreClientBuilder().build(start)) {
+
scanClient.setTimeout(config.getTiConfiguration().getScanTimeout());
+ TiRegion region = scanClient.getRegion();
+ ByteString regionEnd = region.getEndKey();
final List<Kvrpcpb.KvPair> segment =
- scanClient.scan(start, keyRange.getEnd(), startTs);
- if (segment.isEmpty()) {
- split.setResolvedTs(startTs);
- break;
+ scanClient.scan(
+ ConcreteBackOffer.newScannerNextMaxBackOff(),
start, startTs);
+ if (segment == null || segment.isEmpty()) {
+ if (regionEnd.isEmpty() ||
Key.toRawKey(regionEnd).compareTo(end) >= 0) {
+ break;
+ }
+ start = regionEnd;
+ split.setSnapshotStart(start);
+ continue;
}
+
+ boolean reachEnd = false;
+ ByteString nextStart = null;
for (Kvrpcpb.KvPair record : segment) {
+ Key recordKey = Key.toRawKey(record.getKey());
+ if (recordKey.compareTo(end) >= 0) {
+ reachEnd = true;
+ break;
+ }
+ scannedCount++;
if
(TableKeyRangeUtils.isRecordKey(record.getKey().toByteArray())) {
snapshotRecordDeserializer.deserialize(record, output);
+ emittedCount++;
}
+ nextStart = recordKey.next().toByteString();
+ }
+ if (reachEnd || nextStart == null) {
+ break;
}
- start =
- RowKey.toRawKey(segment.get(segment.size() -
1).getKey())
- .next()
- .toByteString();
+ start = nextStart;
// set snapshot offset
split.setSnapshotStart(start);
}
}
+ split.setResolvedTs(startTs);
+ log.info(
+ "[{}] Snapshot events end, scanned kv count: {}, emitted row
count: {}.",
+ split.splitId(),
+ scannedCount,
+ emittedCount);
}
protected void captureStreamingEvents(TiDBSourceSplit split,
Collector<SeaTunnelRow> output)
throws Exception {
+ long batchStartTime = System.currentTimeMillis();
+ long pullStartNanos = System.nanoTime();
long resolvedTs = split.getResolvedTs();
- log.info("Capture streaming event from resolvedTs:{}", resolvedTs);
+ long startResolvedTs = resolvedTs;
CDCClient cdcClient = getCdcClient(split, resolvedTs);
- for (int i = 0; i < config.getBatchSize(); i++) {
+ int polledRows = 0;
+ int ignoredRows = 0;
+ int resolvedTsAdvances = 0;
+ int metadataEvents = 0;
+ int emptyPolls = 0;
+ int pollAttempts = 0;
+ long currentMaxResolvedTs = cdcClient.getMaxResolvedTs();
+ int maxPollAttempts = maxPollAttempts(config.getBatchSize());
+ for (int i = 0, attempts = 0;
+ i < config.getBatchSize() && attempts < maxPollAttempts;
+ attempts++) {
+ long beforeGetResolvedTs = currentMaxResolvedTs;
+ long singlePollStartNanos = System.nanoTime();
final Cdcpb.Event.Row row = cdcClient.get();
+ pollAttempts++;
+ long singlePollCostMs = nanosToMillis(System.nanoTime() -
singlePollStartNanos);
+ currentMaxResolvedTs = cdcClient.getMaxResolvedTs();
if (row == null) {
+ if (currentMaxResolvedTs != beforeGetResolvedTs) {
+ resolvedTsAdvances++;
+ metadataEvents++;
+ continue;
+ }
+ if (singlePollCostMs < EMPTY_POLL_BREAK_THRESHOLD_MS) {
+ metadataEvents++;
+ continue;
+ }
+ emptyPolls++;
break;
}
- handleRow(row);
+ if (handleRow(row)) {
+ polledRows++;
+ } else {
+ ignoredRows++;
+ }
+ i++;
}
- resolvedTs = cdcClient.getMaxResolvedTs();
+ long pullCostMs = nanosToMillis(System.nanoTime() - pullStartNanos);
+ long flushStartNanos = System.nanoTime();
+ resolvedTs = currentMaxResolvedTs;
+ int pendingCommitsBeforeFlush = commits.size();
+ int committedEventsBeforeFlush = committedEvents.size();
if (commits.size() > 0) {
- flushRows(resolvedTs);
+ resolvedTs = flushRowsAndGetSafeResolvedTs(resolvedTs);
}
- // ouput data
+ long flushCostMs = nanosToMillis(System.nanoTime() - flushStartNanos);
+ long emitStartNanos = System.nanoTime();
+ int emittedRows = 0;
+ // output data
while (!committedEvents.isEmpty()) {
Cdcpb.Event.Row row = committedEvents.take();
this.streamingRecordDeserializer.deserialize(row, output);
+ emittedRows++;
}
- // reset resolvedTs
- log.info("Capture streaming event next resolvedTs:{}", resolvedTs);
+ long emitCostMs = nanosToMillis(System.nanoTime() - emitStartNanos);
+ long batchCostMs = System.currentTimeMillis() - batchStartTime;
+ totalPolledRows += polledRows;
+ totalEmittedRows += emittedRows;
+ logStreamingStats(
+ split,
+ startResolvedTs,
+ resolvedTs,
+ polledRows,
+ ignoredRows,
+ emittedRows,
+ resolvedTsAdvances,
+ metadataEvents,
+ emptyPolls,
+ pollAttempts,
+ maxPollAttempts,
+ pendingCommitsBeforeFlush,
+ committedEventsBeforeFlush,
+ pullCostMs,
+ flushCostMs,
+ emitCostMs,
+ batchCostMs);
split.setResolvedTs(resolvedTs);
}
@@ -199,6 +311,13 @@ public class TiDBSourceReader implements
SourceReader<SeaTunnelRow, TiDBSourceSp
k -> {
CDCClient client = new CDCClient(session,
k.getKeyRange());
client.start(finalResolvedTs);
+ log.info(
+ "{} CDC client started, split={},
startResolvedTs={},"
+ + " startLagMs={}.",
+ CDC_DIAG_PREFIX,
+ k.splitId(),
+ finalResolvedTs,
+ resolvedLagMs(finalResolvedTs));
return client;
});
return cdcClient;
@@ -241,10 +360,10 @@ public class TiDBSourceReader implements
SourceReader<SeaTunnelRow, TiDBSourceSp
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {}
- private void handleRow(final Cdcpb.Event.Row row) {
+ private boolean handleRow(final Cdcpb.Event.Row row) {
if (!TableKeyRangeUtils.isRecordKey(row.getKey().toByteArray())) {
// Don't handle index key for now
- return;
+ return false;
}
log.debug("binlog record, type: {}, data: {}", row.getType(), row);
switch (row.getType()) {
@@ -264,14 +383,120 @@ public class TiDBSourceReader implements
SourceReader<SeaTunnelRow, TiDBSourceSp
default:
log.warn("Unsupported row type:" + row.getType());
}
+ return true;
}
protected void flushRows(final long resolvedTs) throws Exception {
+ flushRowsAndGetSafeResolvedTs(resolvedTs);
+ }
+
+ private long flushRowsAndGetSafeResolvedTs(final long resolvedTs) throws
Exception {
+ long safeResolvedTs = resolvedTs;
while (!commits.isEmpty() && commits.firstKey().getTimestamp() <=
resolvedTs) {
- final Cdcpb.Event.Row commitRow =
commits.pollFirstEntry().getValue();
+ final RowKeyWithTs commitKey = commits.firstKey();
+ final Cdcpb.Event.Row commitRow = commits.firstEntry().getValue();
final Cdcpb.Event.Row prewriteRow =
preWrites.remove(RowKeyWithTs.ofStart(commitRow));
+ if (prewriteRow == null) {
+ safeResolvedTs = Math.min(safeResolvedTs,
commitKey.getTimestamp() - 1);
+ log.warn(
+ "Commit row has no matched prewrite row yet, hold it
and stop advancing resolvedTs. "
+ + "commitTs: {}, startTs: {}, key: {}",
+ commitRow.getCommitTs(),
+ commitRow.getStartTs(),
+ commitRow.getKey());
+ break;
+ }
+ commits.pollFirstEntry();
// if pull cdc event block when region split, cdc event will lose.
committedEvents.offer(prewriteRow);
+ totalCommittedRows++;
}
+ return safeResolvedTs;
+ }
+
+ private void logStreamingStats(
+ TiDBSourceSplit split,
+ long startResolvedTs,
+ long endResolvedTs,
+ int polledRows,
+ int ignoredRows,
+ int emittedRows,
+ int resolvedTsAdvances,
+ int metadataEvents,
+ int emptyPolls,
+ int pollAttempts,
+ int maxPollAttempts,
+ int pendingCommitsBeforeFlush,
+ int committedEventsBeforeFlush,
+ long pullCostMs,
+ long flushCostMs,
+ long emitCostMs,
+ long batchCostMs) {
+ long now = System.currentTimeMillis();
+ boolean slowBatch = batchCostMs >= SLOW_STREAMING_BATCH_MS;
+ boolean shouldLog =
+ slowBatch || now - lastStreamingStatsLogTime >=
STREAMING_STATS_LOG_INTERVAL_MS;
+ if (!shouldLog) {
+ return;
+ }
+ lastStreamingStatsLogTime = now;
+ log.info(
+ "{} Streaming stats, split={}, startResolvedTs={},
endResolvedTs={},"
+ + " resolvedLagMs={}, polledRows={}, ignoredRows={},
emittedRows={},"
+ + " resolvedTsAdvances={}, metadataEvents={},
emptyPolls={},"
+ + " pollAttempts={}, maxPollAttempts={},
pendingPrewrites={},"
+ + " pendingCommitsBeforeFlush={},"
+ + " committedQueueBeforeEmit={},
committedQueueAfterEmit={},"
+ + " totalPolledRows={}, totalCommittedRows={},
totalEmittedRows={},"
+ + " pullCostMs={}, flushCostMs={}, emitCostMs={},
batchCostMs={}.",
+ CDC_DIAG_PREFIX,
+ split.splitId(),
+ startResolvedTs,
+ endResolvedTs,
+ resolvedLagMs(endResolvedTs),
+ polledRows,
+ ignoredRows,
+ emittedRows,
+ resolvedTsAdvances,
+ metadataEvents,
+ emptyPolls,
+ pollAttempts,
+ maxPollAttempts,
+ preWrites.size(),
+ pendingCommitsBeforeFlush,
+ committedEventsBeforeFlush,
+ committedEvents.size(),
+ totalPolledRows,
+ totalCommittedRows,
+ totalEmittedRows,
+ pullCostMs,
+ flushCostMs,
+ emitCostMs,
+ batchCostMs);
+ }
+
+ private int maxPollAttempts(int batchSize) {
+ if (batchSize <= 0) {
+ return 0;
+ }
+ if (batchSize > Integer.MAX_VALUE / METADATA_DRAIN_ATTEMPT_MULTIPLIER)
{
+ return Integer.MAX_VALUE;
+ }
+ return Math.max(batchSize, batchSize *
METADATA_DRAIN_ATTEMPT_MULTIPLIER);
+ }
+
+ private long resolvedLagMs(long resolvedTs) {
+ if (resolvedTs <= 0) {
+ return -1L;
+ }
+ return Math.max(0L, System.currentTimeMillis() -
tsoPhysicalMillis(resolvedTs));
+ }
+
+ private long tsoPhysicalMillis(long tso) {
+ return tso >> 18;
+ }
+
+ private long nanosToMillis(long nanos) {
+ return nanos / 1_000_000L;
}
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/utils/TableKeyRangeUtils.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/utils/TableKeyRangeUtils.java
index 1be07634be..0772677a3a 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/utils/TableKeyRangeUtils.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/utils/TableKeyRangeUtils.java
@@ -17,14 +17,13 @@
package org.apache.seatunnel.connectors.seatunnel.cdc.tidb.source.utils;
-import org.apache.seatunnel.shade.com.google.common.base.Preconditions;
-import org.apache.seatunnel.shade.com.google.common.collect.ImmutableList;
-
import org.tikv.common.key.RowKey;
import org.tikv.common.util.KeyRangeUtils;
import org.tikv.kvproto.Coprocessor.KeyRange;
import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
/** Utils to obtain the keyRange of table. */
@@ -36,10 +35,12 @@ public class TableKeyRangeUtils {
}
public static List<KeyRange> getTableKeyRanges(final long tableId, final
int num) {
- Preconditions.checkArgument(num > 0, "Illegal value of num");
+ if (num <= 0) {
+ throw new IllegalArgumentException("Illegal value of num");
+ }
if (num == 1) {
- return ImmutableList.of(getTableKeyRange(tableId));
+ return Collections.singletonList(getTableKeyRange(tableId));
}
final long delta =
@@ -47,7 +48,7 @@ public class TableKeyRangeUtils {
.subtract(BigInteger.valueOf(Long.MIN_VALUE + 1))
.divide(BigInteger.valueOf(num))
.longValueExact();
- final ImmutableList.Builder<KeyRange> builder =
ImmutableList.builder();
+ final List<KeyRange> keyRanges = new ArrayList<>(num);
for (int i = 0; i < num; i++) {
final RowKey startKey =
(i == 0)
@@ -57,13 +58,13 @@ public class TableKeyRangeUtils {
(i == num - 1)
? RowKey.createBeyondMax(tableId)
: RowKey.toRowKey(tableId, Long.MIN_VALUE + delta
* (i + 1));
- builder.add(
+ keyRanges.add(
KeyRangeUtils.makeCoprocRange(startKey.toByteString(),
endKey.toByteString()));
}
- return builder.build();
+ return Collections.unmodifiableList(keyRanges);
}
public static boolean isRecordKey(final byte[] key) {
- return key[9] == '_' && key[10] == 'r';
+ return key.length > 10 && key[9] == '_' && key[10] == 'r';
}
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/enumerator/TiDBSourceSplitEnumeratorTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/enumerator/TiDBSourceSplitEnumeratorTest.java
new file mode 100644
index 0000000000..778cf207ed
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/enumerator/TiDBSourceSplitEnumeratorTest.java
@@ -0,0 +1,183 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.cdc.tidb.source.enumerator;
+
+import org.apache.seatunnel.api.common.metrics.AbstractMetricsContext;
+import org.apache.seatunnel.api.common.metrics.MetricsContext;
+import org.apache.seatunnel.api.event.Event;
+import org.apache.seatunnel.api.event.EventListener;
+import org.apache.seatunnel.api.source.SourceEvent;
+import org.apache.seatunnel.api.source.SourceSplitEnumerator;
+import org.apache.seatunnel.common.utils.SerializationUtils;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.tidb.source.config.TiDBSourceConfig;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.tidb.source.split.TiDBSourceSplit;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.tikv.kvproto.Coprocessor;
+import org.tikv.shade.com.google.protobuf.ByteString;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+class TiDBSourceSplitEnumeratorTest {
+
+ @Test
+ void addSplitsBackShouldNotOverwriteMultipleSplitsForSameReader() {
+ TestingEnumeratorContext context =
+ new TestingEnumeratorContext(1, Collections.singleton(0));
+ TiDBSourceSplitEnumerator enumerator =
+ new TiDBSourceSplitEnumerator(context, new TiDBSourceConfig());
+ List<TiDBSourceSplit> splits = Arrays.asList(newSplit("a"),
newSplit("b"), newSplit("c"));
+
+ enumerator.addSplitsBack(splits, 0);
+
+ Assertions.assertEquals(0, enumerator.currentUnassignedSplitSize());
+ Assertions.assertEquals(splits, context.getAssignedSplits(0));
+ }
+
+ @Test
+ void currentUnassignedSplitSizeShouldCountAllSplitsForSameReader() {
+ TestingEnumeratorContext context = new TestingEnumeratorContext(1,
Collections.emptySet());
+ TiDBSourceSplitEnumerator enumerator =
+ new TiDBSourceSplitEnumerator(context, new TiDBSourceConfig());
+
+ enumerator.addSplitsBack(Arrays.asList(newSplit("a"), newSplit("b"),
newSplit("c")), 0);
+
+ Assertions.assertEquals(3, enumerator.currentUnassignedSplitSize());
+ }
+
+ @Test
+ void checkpointStateShouldRestoreLegacySingleSplitPendingMap() throws
Exception {
+ TiDBSourceSplit split = newSplit("legacy");
+ TiDBSourceCheckpointState legacyState =
+ new TiDBSourceCheckpointState(false, Collections.emptyMap());
+ Map<Integer, TiDBSourceSplit> legacyPendingSplit = new HashMap<>();
+ legacyPendingSplit.put(0, split);
+ setPendingSplitField(legacyState, legacyPendingSplit);
+
+ TiDBSourceCheckpointState restoredState =
+
SerializationUtils.deserialize(SerializationUtils.serialize(legacyState));
+
+ Assertions.assertFalse(restoredState.isShouldEnumerate());
+ Assertions.assertEquals(1, restoredState.getPendingSplit().size());
+ Assertions.assertEquals(1,
restoredState.getPendingSplit().get(0).size());
+ assertSplitEquals(split,
restoredState.getPendingSplit().get(0).get(0));
+
+ TestingEnumeratorContext context =
+ new TestingEnumeratorContext(1, Collections.singleton(0));
+ TiDBSourceSplitEnumerator enumerator =
+ new TiDBSourceSplitEnumerator(context, new TiDBSourceConfig(),
restoredState);
+
+ Assertions.assertEquals(1, enumerator.currentUnassignedSplitSize());
+ enumerator.registerReader(0);
+ Assertions.assertEquals(0, enumerator.currentUnassignedSplitSize());
+ Assertions.assertEquals(1, context.getAssignedSplits(0).size());
+ assertSplitEquals(split, context.getAssignedSplits(0).get(0));
+ }
+
+ private static TiDBSourceSplit newSplit(String suffix) {
+ Coprocessor.KeyRange keyRange =
+ Coprocessor.KeyRange.newBuilder()
+ .setStart(ByteString.copyFromUtf8("start-" + suffix))
+ .setEnd(ByteString.copyFromUtf8("end-" + suffix))
+ .build();
+ return new TiDBSourceSplit("db", "table", keyRange, -1L,
keyRange.getStart(), false);
+ }
+
+ private static void setPendingSplitField(
+ TiDBSourceCheckpointState state, Map<Integer, TiDBSourceSplit>
pendingSplit)
+ throws Exception {
+ Field pendingSplitField =
TiDBSourceCheckpointState.class.getDeclaredField("pendingSplit");
+ pendingSplitField.setAccessible(true);
+ pendingSplitField.set(state, pendingSplit);
+ }
+
+ private static void assertSplitEquals(TiDBSourceSplit expected,
TiDBSourceSplit actual) {
+ Assertions.assertEquals(expected.getDatabase(), actual.getDatabase());
+ Assertions.assertEquals(expected.getTable(), actual.getTable());
+ Assertions.assertEquals(expected.getKeyRange(), actual.getKeyRange());
+ Assertions.assertEquals(expected.getResolvedTs(),
actual.getResolvedTs());
+ Assertions.assertEquals(expected.getSnapshotStart(),
actual.getSnapshotStart());
+ Assertions.assertEquals(expected.isSnapshotCompleted(),
actual.isSnapshotCompleted());
+ }
+
+ private static class TestingEnumeratorContext
+ implements SourceSplitEnumerator.Context<TiDBSourceSplit> {
+ private final int parallelism;
+ private final Set<Integer> registeredReaders;
+ private final Map<Integer, List<TiDBSourceSplit>>
assignedSplitsByReader = new HashMap<>();
+ private final MetricsContext metricsContext = new
AbstractMetricsContext() {};
+ private final EventListener eventListener =
+ new EventListener() {
+ @Override
+ public void onEvent(Event event) {
+ // no-op
+ }
+ };
+
+ private TestingEnumeratorContext(int parallelism, Set<Integer>
registeredReaders) {
+ this.parallelism = parallelism;
+ this.registeredReaders = new HashSet<>(registeredReaders);
+ }
+
+ @Override
+ public int currentParallelism() {
+ return parallelism;
+ }
+
+ @Override
+ public Set<Integer> registeredReaders() {
+ return registeredReaders;
+ }
+
+ @Override
+ public void assignSplit(int subtaskId, List<TiDBSourceSplit> splits) {
+ assignedSplitsByReader
+ .computeIfAbsent(subtaskId, ignored -> new ArrayList<>())
+ .addAll(splits);
+ }
+
+ @Override
+ public void signalNoMoreSplits(int subtask) {}
+
+ @Override
+ public void sendEventToSourceReader(int subtaskId, SourceEvent event)
{}
+
+ @Override
+ public MetricsContext getMetricsContext() {
+ return metricsContext;
+ }
+
+ @Override
+ public EventListener getEventListener() {
+ return eventListener;
+ }
+
+ private List<TiDBSourceSplit> getAssignedSplits(int subtaskId) {
+ return assignedSplitsByReader.getOrDefault(subtaskId,
Collections.emptyList());
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/reader/TiDBSourceReaderTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/reader/TiDBSourceReaderTest.java
new file mode 100644
index 0000000000..b7915f0e77
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/reader/TiDBSourceReaderTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.cdc.tidb.source.reader;
+
+import org.junit.jupiter.api.Test;
+import org.tikv.common.key.RowKey;
+import org.tikv.kvproto.Cdcpb;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.TreeMap;
+import java.util.concurrent.BlockingQueue;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TiDBSourceReaderTest {
+
+ private static final long TABLE_ID = 42L;
+ private static final long ROW_HANDLE = 7L;
+ private static final long START_TS = 100L;
+ private static final long COMMIT_TS = 200L;
+ private static final long RESOLVED_TS = 300L;
+
+ @Test
+ void flushRowsShouldHoldCommitUntilMatchingPrewriteArrives() throws
Exception {
+ TiDBSourceReader reader = new TiDBSourceReader(null, null, null);
+ Cdcpb.Event.Row commit = row(Cdcpb.Event.LogType.COMMIT, START_TS,
COMMIT_TS);
+
+ assertTrue(handleRow(reader, commit));
+
+ long safeResolvedTs = flushRowsAndGetSafeResolvedTs(reader,
RESOLVED_TS);
+
+ assertEquals(COMMIT_TS - 1, safeResolvedTs);
+ assertEquals(1, commits(reader).size());
+ assertTrue(committedEvents(reader).isEmpty());
+
+ Cdcpb.Event.Row prewrite = row(Cdcpb.Event.LogType.PREWRITE, START_TS,
0L);
+
+ assertTrue(handleRow(reader, prewrite));
+
+ safeResolvedTs = flushRowsAndGetSafeResolvedTs(reader, RESOLVED_TS);
+
+ assertEquals(RESOLVED_TS, safeResolvedTs);
+ assertTrue(commits(reader).isEmpty());
+ assertTrue(preWrites(reader).isEmpty());
+ BlockingQueue<Cdcpb.Event.Row> committedEvents =
committedEvents(reader);
+ assertFalse(committedEvents.isEmpty());
+ assertSame(prewrite, committedEvents.poll());
+ assertTrue(committedEvents.isEmpty());
+ }
+
+ private static Cdcpb.Event.Row row(Cdcpb.Event.LogType type, long startTs,
long commitTs) {
+ return Cdcpb.Event.Row.newBuilder()
+ .setType(type)
+ .setStartTs(startTs)
+ .setCommitTs(commitTs)
+ .setKey(RowKey.toRowKey(TABLE_ID, ROW_HANDLE).toByteString())
+ .build();
+ }
+
+ private static boolean handleRow(TiDBSourceReader reader, Cdcpb.Event.Row
row)
+ throws Exception {
+ Method handleRow =
+ TiDBSourceReader.class.getDeclaredMethod("handleRow",
Cdcpb.Event.Row.class);
+ handleRow.setAccessible(true);
+ return (Boolean) handleRow.invoke(reader, row);
+ }
+
+ private static long flushRowsAndGetSafeResolvedTs(TiDBSourceReader reader,
long resolvedTs)
+ throws Exception {
+ Method flushRowsAndGetSafeResolvedTs =
+ TiDBSourceReader.class.getDeclaredMethod(
+ "flushRowsAndGetSafeResolvedTs", long.class);
+ flushRowsAndGetSafeResolvedTs.setAccessible(true);
+ return (Long) flushRowsAndGetSafeResolvedTs.invoke(reader, resolvedTs);
+ }
+
+ private static TreeMap<RowKeyWithTs, Cdcpb.Event.Row>
commits(TiDBSourceReader reader)
+ throws Exception {
+ return rowBuffer(reader, "commits");
+ }
+
+ private static TreeMap<RowKeyWithTs, Cdcpb.Event.Row>
preWrites(TiDBSourceReader reader)
+ throws Exception {
+ return rowBuffer(reader, "preWrites");
+ }
+
+ @SuppressWarnings("unchecked")
+ private static TreeMap<RowKeyWithTs, Cdcpb.Event.Row> rowBuffer(
+ TiDBSourceReader reader, String fieldName) throws Exception {
+ Field field = TiDBSourceReader.class.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return (TreeMap<RowKeyWithTs, Cdcpb.Event.Row>) field.get(reader);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static BlockingQueue<Cdcpb.Event.Row>
committedEvents(TiDBSourceReader reader)
+ throws Exception {
+ Field field =
TiDBSourceReader.class.getDeclaredField("committedEvents");
+ field.setAccessible(true);
+ return (BlockingQueue<Cdcpb.Event.Row>) field.get(reader);
+ }
+}
