Skip to content

fix: TAP-12454 fix for starrocks cold table pinning breakpoint #841

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jarad0628 merged 1 commit into from
Aug 8, 2026
Merged

fix: TAP-12454 fix for starrocks cold table pinning breakpoint #841

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions ...nector/src/main/java/io/tapdata/connector/starrocks/streamload/StarrocksStreamLoader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -279,6 +279,9 @@ private void initializeFlushScheduler() {
*/
private void checkAndFlushIfNeeded() throws StarrocksRetryableException {
synchronized (writeLock) {
// 驱逐队首连续的、已全部落库(不在 pendingFlushTables)的陈旧 offset 锚点，避免长期无变更的冷表占据队首，导致断点被永久钉死在其旧位点上
evictStaleFirstOffsets();

if (pendingFlushTables.isEmpty()) {
return; // 没有数据需要刷新
}
Expand Down Expand Up @@ -317,6 +320,35 @@ private void checkAndFlushIfNeeded() throws StarrocksRetryableException {
}
}

/**
* 驱逐 firstOffsetByTable 队首连续的、已全部落库的陈旧 offset 锚点。
* <p>
* flushTable 回传断点时取的是队首表的 offset。若一张长期无变更的冷表在全量/增量早期
* 进入队首后再无新数据，它就不会再进入 pendingFlushTables，也就永远不会被 flush，
* 从而永远不会被移出队首(见 flushTable 中 tableName.equals(firstTableName) 分支)，
* 导致所有其它表 flush 回传的都是这张冷表的旧位点，断点被永久钉死。
* <p>
* 判据：队首表若不在 pendingFlushTables，说明它此刻没有待落库数据、之前的数据均已成功
* 导入，丢弃其 offset 锚点不会丢数，可直接驱逐；一旦遇到队首表在 pendingFlushTables 中，
* 则停止驱逐——该表会被定时/大小阈值 flush 后按正常流程移出队首并推进断点。
*/
private void evictStaleFirstOffsets() {
synchronized (firstOffsetByTable) {
Iterator<Map.Entry<String, TapCallbackOffset>> iterator = firstOffsetByTable.entrySet().iterator();
while (iterator.hasNext()) {
Map.Entry<String, TapCallbackOffset> firstEntry = iterator.next();
String firstTableName = firstEntry.getKey();
if (pendingFlushTables.contains(firstTableName)) {
// 队首表仍有待落库数据，等它被正常 flush 后自然推进，停止驱逐
break;
}
iterator.remove();
taplogger.info("Evicted stale offset anchor of table {} from queue head " +
"(already flushed, no pending data), allowing breakpoint to advance", firstTableName);
}
}
}

/**
* 刷新指定的表
*/
Expand Down