You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "Caideyipi (via GitHub)" <gi...@apache.org> on 2026/09/01 11:29:03 UTC
[PR] [Pipe] Add reliable DataNode completion metric [iotdb]
Caideyipi opened a new pull request, #18563:
URL: https://github.com/apache/iotdb/pull/18563
# Pipe 鍙戦€佸畬鎴愭寚鏍囪璁¤鏄?
## 1. 缁撹
鏈疄鐜颁繚鐣欎簡 [Apache IoTDB PR #18280](https://github.com/apache/iotdb/pull/18280) 鐨勫畬鏁寸鍒扮灞忛殰鏂规锛屽苟鍦ㄥ綋鍓嶅垎鏀笂琛ュ厖浜?expected DataRegion 瀹屾暣鎬ф鏌ャ€?
鏂板鎸囨爣锛?
```text
pipe_datanode_completion_ready{name=<pipeName>,creation_time=<creationTime>}
```
鎸囨爣鍊煎彧鏈?`0` 鍜?`1`锛?
- `1`锛氳繖涓?sender DataNode 涓婏紝璇?Pipe 褰撳墠鎵€鏈夐鏈?DataRegion 鐨勬渶鏂?full-FLUSH 灞忛殰鍧囧凡缁忚繃 source銆乸rocessor銆乻ink锛屽苟鍦?sink 鎴愬姛 ACK 鍚庢寜搴忔彁浜わ紱鍚屾椂娌℃湁寰呭鐞嗙殑闈炲績璺充簨浠讹紝涔熸病鏈夋娴嬪埌浠诲姟銆乻ource銆乤ssigner銆乧ommitter銆佸紓甯告垨闄嶇骇鐘舵€佸彉鍖栥€?- `0`锛氭湭瀹屾垚銆佷笉鏀寔銆佺姸鎬佹湭鐭ャ€佸彂鐢熺珵鎬佹垨鍋ュ悍妫€鏌ュけ璐ャ€俙0` 涓嶅尯鍒嗗叿浣撳師鍥犮€?
璁捐鍘熷垯鏄?**鍏佽鏆傛椂鍋囬槾鎬э紝浣嗕笉鍏佽鍋囬槼鎬?*銆傚師鏈?`remaining_event_count` 缁х画浣滀负杩涘害鎸囨爣锛涘畠涓?`0` 鏄畬鎴愮殑蹇呰鏉′欢锛屼絾涓嶈兘鍗曠嫭浣滀负鈥滃凡鍙戦€佸畬鎴愨€濈殑璇佹槑銆?
## 2. 涓轰粈涔堜笉鑳藉彧缁?remaining event 鍔犱笂鈥滅瓑寰?flush鈥濆拰鈥滃凡鎹曡幏 TsFile鈥?
涓€涓緝灏忕殑瀹炵幇鍙互鎶?storage engine 涓瓑寰呭叧闂殑 processor銆佸凡鎹曡幏鐨?TsFile 鍜?Pipe 闃熷垪鐩稿姞锛屼絾瀹冨彧鑳藉緱鍒拌繎浼肩Н鍘嬮噺锛屾棤娉曞彲闈犺瘉鏄庡彂閫佸畬鎴愩€?
### 2.1 鐘舵€佽縼绉诲瓨鍦ㄨ鏁扮┖绐?
鍚屼竴涓?TsFile 浼氱粡鍘嗭細
```text
working/closing processor -> close callback -> assigner -> source queue -> processor queue -> sink/in-flight RPC
```
濡傛灉鍒嗗埆璇诲彇杩欎簺缁撴瀯鍐嶇浉鍔狅紝瀵硅薄鍦ㄤ袱涓粨鏋勪箣闂寸Щ鍔ㄦ椂鍙兘鏆傛椂涓嶅睘浜庝换浣曚竴涓凡璇诲彇鐨勫揩鐓э紝浠庤€岀灛鏃跺緱鍒?`0`銆傜粰鏇村瀹瑰櫒鍔犺鏁板彧鑳界缉灏忕獥鍙ｏ紝涓嶈兘浠庢牴鏈笂娑堥櫎璺ㄧ粍浠跺揩鐓х珵鎬併€?
### 2.2 闃熷垪涓虹┖涓嶇瓑浜?sink 宸茬‘璁?
浜嬩欢浠?sink queue 鍙栧嚭鍚庯紝鍒扮洰鏍囩杩斿洖鎴愬姛 ACK 涔嬪墠锛岄槦鍒楀彲浠ュ凡缁忎负绌恒€傛鏃?`remaining_event_count == 0` 浠嶄笉鑳借瘉鏄庣洰鏍囩宸茬粡鎺ユ敹鎴栧姞杞藉畬鎴愩€?
### 2.3 FLUSH 涓庡苟鍙戝啓鍏ュ瓨鍦ㄨ竟鐣岀珵鎬?
full FLUSH 寮€濮嬪悗锛屽鏋滄湁骞跺彂 insert 鍒涘缓浜嗘柊鐨?working processor锛岃€岃 processor 娌¤繘鍏ユ湰娆?flush 鐨勬崟鑾烽泦鍚堬紝鍒欐棫 TsFile 鍏ㄩ儴鍙戦€佸畬涔熶笉鑳戒唬琛ㄦ湰杞啓鍏ュ凡瀹屾垚銆傚繀椤讳负姣忔 full FLUSH 寤虹珛 token锛屽苟鍦ㄥ苟鍙?insert 鍑虹幇鏃朵娇 token 澶辨晥銆?
### 2.4 鐢熷懡鍛ㄦ湡鍙樺寲浼氳鏃х姸鎬佽鎶?
Pipe task銆乺ealtime source銆乤ssigner 鎴?committer 琚浛鎹㈠悗锛屾棫瀹炰緥宸叉彁浜ょ殑鐘舵€佷笉鑳界敤浜庤瘉鏄庢柊瀹炰緥瀹屾垚銆備换鍔″垵濮嬪寲澶辫触閫犳垚 DataRegion 缂哄け鏃讹紝涔熶笉鑳芥妸鈥滄病鏈夋湰鍦?task鈥濆綋鎴愬畬鎴愩€?
鍥犳锛屽鏋滅洰鏍囧彧鏄€滆繎浼肩Н鍘嬮噺鈥濓紝灏忔敼 `remaining_event_count` 瓒冲锛涘鏋滅洰鏍囨槸纭畾鈥淧ipe 宸插彂閫佸畬鎴愨€濓紝鍒欏繀椤诲缓绔嬬鍒扮鏈夊簭灞忛殰銆?18280 鐨勪富瑕佷綋閲忔鏄敤鏉ュ叧闂笂杩板亣闃虫€х獥鍙ｏ紝涓嶈兘鍙繚鐣欏叾涓煇涓€涓鏁扮偣銆?
## 3. 瀹屾垚灞忛殰娴佺▼
```text
鍋滄骞?join writers
|
v
鎵ц瑕嗙洊鎵€鏈夌浉鍏?DataRegion 鐨?full FLUSH
|
+-- 浣挎棫 completion token 澶辨晥锛岃褰曟柊 token
+-- 鎹曡幏褰撴椂 working + closing TsFileProcessor
+-- 绛夊緟宸插湪杩愯鐨勬櫘閫?async flush
+-- 鍏抽棴骞剁瓑寰呮崟鑾风殑 processor 鍙?close callback
+-- 浠呭湪 token 鏈骞跺彂 insert 澶辨晥鏃跺彂甯?barrier
|
v
assigner锛堢粦瀹?assigner epoch + data generation锛? |
v
realtime source -> processor锛坆arrier 涓嶅厑璁歌鍚炴帀/鏀瑰啓锛? |
v
sink queue锛坆arrier 涓嶅弬涓庢櫘閫?heartbeat 鍚堝苟锛? |
v
sink 鎴愬姛 ACK -> ordered commit -> onCommitted hook
|
v
completion operator 鍙岄噸蹇収鏍￠獙
|
v
pipe_datanode_completion_ready = 1
```
鍏抽敭鐐癸細barrier 鎺掑湪鏈 flush 鎹曡幏骞跺彂甯冪殑 TsFile 浜嬩欢涔嬪悗锛涘彧鏈?sink 鎴愬姛澶勭悊 barrier 骞跺畬鎴?ordered commit锛孌ataRegion 鎵嶄細琚爣璁板畬鎴愩€?
## 4. fail-closed 鏉′欢
浠ヤ笅浠讳竴鏉′欢鎴愮珛锛屾寚鏍囬兘杩斿洖 `0`锛?
- `remaining_event_count` 涓粛鏈夐潪蹇冭烦浜嬩欢锛?- Pipe 涓嶅瓨鍦ㄣ€佷笉鏄?RUNNING USER Pipe锛屾垨瀛樺湪 Pipe/runtime/task 寮傚父锛?- source銆乸rocessor銆乻ink 鎴?TsFile load strategy 涓嶅湪鏀寔鑼冨洿锛?- historical source 灏氭湭娑堣垂瀹岋紝鎴?realtime source 灏氭湭瀹屾暣鍚姩锛?- 褰撳墠瀹為檯 DataRegion task 闆嗗悎涓庢牴鎹?PipeMeta銆乴eader 鍜屾湰鍦?StorageEngine 璁＄畻鍑虹殑棰勬湡闆嗗悎涓嶄竴鑷达紱
- task/source/assigner/committer 瀹炰緥鍙戠敓鏇挎崲锛?- full-FLUSH token 琚苟鍙?insert 澶辨晥锛?- barrier 鐨?generation 钀藉悗浜庢渶鏂版暟鎹?generation锛?- 浜嬩欢鍙戝竷銆佸紩鐢ㄨ鏁般€佸叆闃熸垨渚涚粰鍙戠敓澶辫触锛?- hybrid source 姝ｅ湪绛夊緟 TsFile 鎭㈠宸蹭涪寮冪殑 tablet锛坉egraded锛夛紱
- 涓ゆ浠诲姟鎷撴墤蹇収鎴栦袱娆＄姸鎬佸揩鐓т箣闂村彂鐢熶换浣曞彉鍖栵紱
- 鎸囨爣璁＄畻鏈熼棿鍙戠敓杩愯鏃跺紓甯告垨鑾峰彇浠诲姟璇婚攣瓒呮椂銆?
浜嬩欢涓㈠け鎴?publication failure 灞炰簬涓嶅彲鑷姩璇佹槑鎭㈠鐨勬儏鍐碉紝鎸囨爣浼氫繚鎸?`0`锛岄€氬父闇€瑕佸厛淇闂锛屽啀閲嶅惎/閲嶅缓鐩稿簲 Pipe task 骞堕噸鏂版墽琛屽畬鎴愬崗璁€?
## 5. 鏀寔鑼冨洿
褰撳墠鍙湁浠ヤ笅缁勫悎鑳借繑鍥?`1`锛?
- RUNNING 鐨?USER Pipe锛?- source/extractor锛歚iotdb-extractor` 鎴?`iotdb-source`锛?- processor锛歚do-nothing-processor`锛?- sink/connector锛氬唴缃?IoTDB Thrift connector/sink 鐨?sync銆乤sync銆丼SL 绛夊埆鍚嶏紱
- TsFile load strategy锛歚sync`锛?- source 宸插惎鍔紝historical 闃舵宸叉秷璐瑰畬鎴愶紱
- DataRegion/DML 鍙戦€佸畬鎴愩€?
鐩墠瀹冧笉鑳借瘉鏄?SchemaRegion/DDL 宸插湪鐩爣绔畬鎴愶紝涔熶笉涓鸿嚜瀹氫箟 processor銆佽嚜瀹氫箟 sink 鎴栧紓姝?TsFile load strategy 鎻愪緵瀹屾垚淇濊瘉銆傝繖浜涚粍鍚堜細淇濆畧杩斿洖 `0`銆?
## 6. 姝ｇ‘浣跨敤鍗忚
1. 鍋滄鎵€鏈夊彲鑳藉悜鐩稿叧 DataRegion 鍐欏叆鐨?writer锛屽苟绛夊緟 writer 绾跨▼缁撴潫銆傚綋鍓?generation 鏄?DataRegion 绾х殑锛涘嵆浣挎槸 Pipe pattern 涔嬪鐨勫苟鍙戝啓鍏ワ紝涔熷彲鑳戒娇鎸囨爣淇濆畧鍦颁繚鎸?`0`銆?2. 鎵ц鎴愬姛鐨?full `FLUSH`锛岀‘淇濊鐩栬 Pipe 鍙兘鍙戦€佺殑鎵€鏈?DataRegion銆傛帹鑽愭墽琛屽叏灞€ full FLUSH锛涗笉瑕佷娇鐢ㄥ彧鍏抽棴 `SEQ` 鎴?`UNSEQ` 鐨勯儴鍒?flush 浣滀负瀹屾垚灞忛殰銆?3. FLUSH 鎴愬姛杩斿洖鍚庯紝鍦ㄦ瘡涓鏈?sender DataNode 涓婅幏鍙栨柊椴滅殑鎸囨爣鏍锋湰銆?4. 鍙湁鎵€鏈夐鏈?series 閮藉瓨鍦ㄤ笖鍊煎潎涓?`1`锛屾墠鑳藉垽瀹氭湰杞?Pipe DML 鍙戦€佸畬鎴愩€?
Prometheus 鍦烘櫙杩樺繀椤诲悓鏃剁‘璁わ細
- sample timestamp 鏅氫簬鏈 FLUSH锛?- 姣忎釜鐩爣鐨?`up == 1`锛?- series 鏁伴噺涓庨鏈?sender DataNode 鏁颁竴鑷达紱
- 涓嶆妸缂哄け series銆佹姄鍙栧け璐ユ垨鏃ф牱鏈綋浣滃畬鎴愩€?
`creation_time` 鐢ㄤ簬鍖哄垎鍚屽悕 Pipe 鐨勪笉鍚?incarnation銆傚垽鏂椂搴旈攣瀹氭湰杞?Pipe 鐨勫噯纭?`name + creation_time`锛屼笉鑳藉彧鎸?`name` 鑱氬悎鏃?series銆?
## 7. 涓?PR #18280 鐨勫叧绯诲強褰撳墠鍒嗘敮閫傞厤
鏈疄鐜伴噰鐢?#18280 鐨勬牳蹇冩彁浜?`12bc3f3a5c00265f5c04dc28a4e6c76affe326bf`锛坄[Pipe] Add reliable DataNode completion metric`锛夈€傚畬鏁村疄鐜板寘鍚細
- full FLUSH 瀵?working銆乧losing processor 鍜岄噸鍙?ordinary async flush 鐨勭瓑寰咃紱
- completion token銆乨ata generation銆乤ssigner epoch 鍜?publication failure epoch锛?- completion barrier 鍦?source銆乸rocessor銆乻ink queue 涓殑淇濆簭锛?- sink ACK 鍚庣殑 ordered commit hook锛?- task/source/assigner/committer 鐢熷懡鍛ㄦ湡鏍￠獙锛?- fail-closed completion operator 鍙婂苟鍙戞祴璇曘€?
褰撳墠鍒嗘敮棰濆澶嶇敤浜?master 宸叉湁鐨?expected-DataRegion 璁＄畻锛屽苟瑕佹眰锛?
```text
actual DataRegion source ids == expected DataRegion ids
```
杩欓伩鍏嶄簡鏌愪釜 DataRegion task 鍒濆鍖栧け璐ユ垨缂哄け鏃讹紝鎸囨爣鍥犫€滄湰鍦伴泦鍚堜负绌?涓嶅畬鏁粹€濊€岃鎶?`1`銆?
## 8. 楠岃瘉缁撴灉
- Spotless锛氶€氳繃锛?- 鐩稿叧妯″潡 clean test-compile锛氶€氳繃锛?- 灞忛殰銆佺敓鍛藉懆鏈熴€侀槦鍒楀拰 full-flush 骞跺彂閽堝鎬ф祴璇曪細18 涓紝Failures 0锛孍rrors 0锛孲kipped 0锛?- 鑻辨枃 locale 鍏?reactor `test-compile`锛?2/52 閫氳繃锛?- 涓枃 locale 鍏?reactor `test-compile`锛?2/52 閫氳繃锛?- `git diff --check`锛氶€氳繃銆?
閽堝鎬ф祴璇曡鐩栦簡 completion generation銆乫ail-closed銆佹垚鍛樺彉鍖栥€乻ource 鏇挎崲銆乧ommitter 鏇挎崲銆乥arrier 涓嶈 heartbeat 鍚堝苟銆佷簨浠舵敹闆嗗け璐ャ€乮nsert 浣?flush token 澶辨晥锛屼互鍙?full FLUSH 涓?ordinary async flush 閲嶅彔绛夊満鏅€?
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Pipe] Add reliable DataNode completion metric [iotdb]
Posted by "Caideyipi (via GitHub)" <gi...@apache.org>.
Caideyipi commented on code in PR #18563:
URL: https://github.com/apache/iotdb/pull/18563#discussion_r4244554150
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java:
##########
@@ -212,6 +387,9 @@ private void assignToSource(
}
Review Comment:
Addressed in c803c30b8229.
Applied as Event.bindSource(long sourceId), implemented by EnrichedEvent. The assigner binds every copied event through the common method. PipeHeartbeatEvent adds its completion commit hook when a barrier first receives a valid source binding; ordinary events use the same source identity without completion-specific behavior. Added coverage that repeated binding neither changes the identity nor duplicates the hook.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java:
##########
@@ -2538,6 +2538,78 @@ public void syncCloseAllWorkingTsFileProcessors() {
}
}
+ /**
+ * Closes all working TsFile processors and waits for processors that were already closing.
+ *
+ * <p>This stronger variant is used by a full flush before a Pipe completion barrier is published.
+ * Unlike {@link #syncCloseAllWorkingTsFileProcessors()}, failures are propagated to the caller so
+ * a failed flush can never publish a completion barrier.
+ */
+ public void syncCloseAllWorkingAndClosingTsFileProcessors()
+ throws InterruptedException, ExecutionException {
+ final Set<TsFileProcessor> targets = new HashSet<>();
+ final Set<TsFileProcessor> targetsToClose = new HashSet<>();
+ final List<Future<?>> closingFutures = new ArrayList<>();
+
+ writeLock("syncCloseAllWorkingAndClosingTsFileProcessors");
+ try {
+ targets.addAll(closingSequenceTsFileProcessor);
+ targets.addAll(closingUnSequenceTsFileProcessor);
+ for (final TsFileProcessor target : targets) {
+ closingFutures.add(target.getCloseFuture());
+ }
+
+ targetsToClose.addAll(workSequenceTsFileProcessors.values());
+ targetsToClose.addAll(workUnsequenceTsFileProcessors.values());
+ targets.addAll(targetsToClose);
+ } finally {
+ writeUnlock();
+ }
+
+ int closedProcessorCount = 0;
+ while (!targetsToClose.isEmpty()) {
+ final List<Future<?>> currentFutures = new ArrayList<>();
+ final Set<TsFileProcessor> targetsWaitingForOrdinaryFlush = new HashSet<>();
+
+ writeLock("syncCloseAllWorkingAndClosingTsFileProcessors");
+ try {
+ final Iterator<TsFileProcessor> iterator = targetsToClose.iterator();
+ while (iterator.hasNext()) {
+ final TsFileProcessor target = iterator.next();
+ final Future<?> future =
+ asyncCloseOneTsFileProcessorForFullFlush(target.isSequence(), target);
+ currentFutures.add(future);
+ if (target.alreadyMarkedClosing()) {
+ closingFutures.add(future);
+ iterator.remove();
+ closedProcessorCount++;
+ } else {
+ targetsWaitingForOrdinaryFlush.add(target);
+ }
+ }
+ } finally {
+ writeUnlock();
Review Comment:
Addressed in c803c30b8229.
Added a comment beside targetsWaitingForOrdinaryFlush. Here 'ordinary flush' means TsFileProcessor.asyncFlush(), which flushes memtables while keeping the file open. A full FLUSH must await that operation outside the DataRegion write lock and then retry closing the captured processor, so its close callback precedes the completion barrier. Both the PR's full-FLUSH tests and master's already-closing-processor test are retained.
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitManager.java:
##########
@@ -173,6 +173,16 @@ public CommitterKey getCommitterKey(
return generateCommitterKey(pipeName, creationTime, regionId);
}
+ public boolean isCurrentCommitterKey(final CommitterKey committerKey) {
+ return committerKey != null
+ && eventCommitterMap.containsKey(committerKey)
+ && committerKey.equals(
+ generateCommitterKey(
+ committerKey.getPipeName(),
+ committerKey.getCreationTime(),
+ committerKey.getRegionId()));
Review Comment:
Addressed in c803c30b8229.
Added Javadoc explaining the two lifecycle checks. containsKey verifies that this committer is still registered, rejecting dropped tasks. Comparing with the generated key also checks its current restart generation, rejecting a barrier from an older committer after the same Pipe/DataRegion task restarts. A committed barrier from that older instance cannot prove completion for its successor.
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Pipe] Add reliable DataNode completion metric [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #18563:
URL: https://github.com/apache/iotdb/pull/18563#discussion_r4268389507
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java:
##########
@@ -2538,6 +2538,78 @@ public void syncCloseAllWorkingTsFileProcessors() {
}
}
+ /**
+ * Closes all working TsFile processors and waits for processors that were already closing.
+ *
+ * <p>This stronger variant is used by a full flush before a Pipe completion barrier is published.
+ * Unlike {@link #syncCloseAllWorkingTsFileProcessors()}, failures are propagated to the caller so
+ * a failed flush can never publish a completion barrier.
+ */
+ public void syncCloseAllWorkingAndClosingTsFileProcessors()
+ throws InterruptedException, ExecutionException {
+ final Set<TsFileProcessor> targets = new HashSet<>();
+ final Set<TsFileProcessor> targetsToClose = new HashSet<>();
+ final List<Future<?>> closingFutures = new ArrayList<>();
+
+ writeLock("syncCloseAllWorkingAndClosingTsFileProcessors");
+ try {
+ targets.addAll(closingSequenceTsFileProcessor);
+ targets.addAll(closingUnSequenceTsFileProcessor);
+ for (final TsFileProcessor target : targets) {
+ closingFutures.add(target.getCloseFuture());
+ }
+
+ targetsToClose.addAll(workSequenceTsFileProcessors.values());
+ targetsToClose.addAll(workUnsequenceTsFileProcessors.values());
+ targets.addAll(targetsToClose);
+ } finally {
+ writeUnlock();
+ }
+
+ int closedProcessorCount = 0;
+ while (!targetsToClose.isEmpty()) {
+ final List<Future<?>> currentFutures = new ArrayList<>();
+ final Set<TsFileProcessor> targetsWaitingForOrdinaryFlush = new HashSet<>();
+
+ writeLock("syncCloseAllWorkingAndClosingTsFileProcessors");
+ try {
+ final Iterator<TsFileProcessor> iterator = targetsToClose.iterator();
+ while (iterator.hasNext()) {
+ final TsFileProcessor target = iterator.next();
+ final Future<?> future =
+ asyncCloseOneTsFileProcessorForFullFlush(target.isSequence(), target);
+ currentFutures.add(future);
+ if (target.alreadyMarkedClosing()) {
+ closingFutures.add(future);
+ iterator.remove();
+ closedProcessorCount++;
+ } else {
+ targetsWaitingForOrdinaryFlush.add(target);
+ }
+ }
+ } finally {
+ writeUnlock();
Review Comment:
The file is no longer kept open after the first flush.
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Pipe] Add reliable DataNode completion metric [iotdb]
Posted by "Caideyipi (via GitHub)" <gi...@apache.org>.
Caideyipi commented on code in PR #18563:
URL: https://github.com/apache/iotdb/pull/18563#discussion_r4244553046
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -677,6 +696,90 @@ public Set<Integer> getPipeTaskRegionIdSet(final String pipeName, final long cre
: pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().keySet();
}
+ public Pair<Boolean, Map<Integer, PipeRealtimeDataRegionSource>> getPipeCompletionSnapshot(
+ final String pipeName, final long creationTime) {
+ if (!tryReadLockWithTimeOutInMs(10)) {
+ return new Pair<>(false, Collections.emptyMap());
+ }
+
+ try {
+ final PipeMeta pipeMeta = pipeMetaKeeper.getPipeMeta(pipeName, creationTime);
+ if (pipeMeta == null) {
+ return new Pair<>(false, Collections.emptyMap());
+ }
+
+ final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta();
+ final String sourceName =
+ staticMeta
+ .getSourceParameters()
+ .getStringOrDefault(
+ Arrays.asList(PipeSourceConstant.EXTRACTOR_KEY, PipeSourceConstant.SOURCE_KEY),
+ BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName());
+ final String processorName =
+ staticMeta
+ .getProcessorParameters()
+ .getStringOrDefault(
+ PipeProcessorConstant.PROCESSOR_KEY,
+ BuiltinPipePlugin.DO_NOTHING_PROCESSOR.getPipePluginName());
+ final PipeParameters sinkParameters = staticMeta.getSinkParameters();
+ final String sinkName = PipeSinkConstant.getConnectorOrSinkNameWithDefault(sinkParameters);
+ boolean supported =
+ staticMeta.getPipeType() == PipeType.USER
+ && COMPLETION_SUPPORTED_SOURCES.stream()
+ .anyMatch(name -> name.equalsIgnoreCase(sourceName))
+ && BuiltinPipePlugin.DO_NOTHING_PROCESSOR
+ .getPipePluginName()
+ .equalsIgnoreCase(processorName)
+ && COMPLETION_SUPPORTED_SINKS.stream()
+ .anyMatch(name -> name.equalsIgnoreCase(sinkName))
+ && PipeSinkConstant.CONNECTOR_LOAD_TSFILE_STRATEGY_SYNC_VALUE.equalsIgnoreCase(
+ sinkParameters.getStringOrDefault(
+ Arrays.asList(
+ PipeSinkConstant.CONNECTOR_LOAD_TSFILE_STRATEGY_KEY,
+ PipeSinkConstant.SINK_LOAD_TSFILE_STRATEGY_KEY),
+ PipeSinkConstant.CONNECTOR_LOAD_TSFILE_STRATEGY_SYNC_VALUE))
+ && pipeMeta.getRuntimeMeta().getStatus().get() == PipeStatus.RUNNING
+ && pipeMeta.getRuntimeMeta().getNodeId2PipeRuntimeExceptionMap().isEmpty()
+ && pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().values().stream()
+ .noneMatch(PipeTaskMeta::hasExceptionMessages);
Review Comment:
Addressed in c803c30b8229.
Cached the static plugin/load-strategy eligibility in PipeStaticMeta. It is initialized from constructor parameters and recomputed by both deserialization paths, without changing the serialized format. ALTER PIPE installs a new static metadata incarnation and therefore obtains a fresh value. Running status, exceptions, source readiness, and expected/actual task membership are still checked on each snapshot, since caching those runtime values could produce a stale completion result. Added supported/unsupported and round-trip tests.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskSourceStage.java:
##########
@@ -107,4 +109,17 @@ public void dropSubtask() throws PipeException {
public EventSupplier getEventSupplier() {
return pipeExtractor::supply;
}
+
+ public PipeExtractor getPipeExtractor() {
+ return pipeExtractor;
+ }
+
+ public long getCompletionSourceId() {
+ if (!(pipeExtractor instanceof IoTDBDataRegionSource)) {
+ return Long.MIN_VALUE;
+ }
+ final PipeRealtimeDataRegionSource realtimeSource =
+ ((IoTDBDataRegionSource) pipeExtractor).getRealtimeSourceForCompletion();
+ return realtimeSource == null ? Long.MIN_VALUE : realtimeSource.getCompletionSourceId();
+ }
Review Comment:
Addressed in c803c30b8229.
Added the explanation in PipeTaskSourceStage. The full-FLUSH barrier travels through the realtime assigner/source, so the collector must use that same source instance ID for completion invalidation and stale-callback rejection. The outer IoTDBDataRegionSource separately requires the historical phase to be fully consumed before the snapshot is ready; this does not omit historical data from the completion check.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetrics.java:
##########
@@ -194,26 +222,100 @@ private void removeAutoGauge(final String pipeID) {
Metric.PIPE_TSFILE_EVENT_TRANSFER_TIME.toString(),
Tag.NAME.toString(),
operator.getPipeName());
- remainingEventAndTimeOperatorMap.remove(pipeID);
}
//////////////////////////// register & deregister (pipe integration) ////////////////////////////
public void register(final IoTDBDataRegionSource source) {
// The metric is global thus the regionId is omitted
final String pipeID = source.getPipeName() + "_" + source.getCreationTime();
- remainingEventAndTimeOperatorMap
- .computeIfAbsent(
+ final PipeDataNodeRemainingEventAndTimeOperator remainingOperator =
+ remainingEventAndTimeOperatorMap.computeIfAbsent(
pipeID,
k ->
new PipeDataNodeRemainingEventAndTimeOperator(
- source.getPipeName(), source.getCreationTime()))
- .register(source);
+ source.getPipeName(), source.getCreationTime()));
+ remainingOperator.register(source);
+ final PipeDataNodeCompletionOperator completionOperator =
+ completionOperatorMap.computeIfAbsent(
+ pipeID,
+ k ->
+ new PipeDataNodeCompletionOperator(
+ () -> remainingOperator.getRemainingNonHeartbeatEvents() == 0,
+ () ->
+ PipeDataNodeAgent.task()
+ .getPipeCompletionSnapshot(
+ source.getPipeName(), source.getCreationTime())));
+ if (source.getRealtimeSourceForCompletion() != null) {
+ completionOperator.registerDataRegionSource(source.getRealtimeSourceForCompletion());
+ }
if (Objects.nonNull(metricService)) {
createMetrics(pipeID);
}
}
+ public void register(
+ final PipeRealtimeDataRegionSource source, final PipeDataRegionAssigner assigner) {
+ final String pipeID = source.getPipeName() + "_" + source.getCreationTime();
+ final PipeDataNodeCompletionOperator operator = completionOperatorMap.get(pipeID);
+ if (operator != null) {
+ operator.register(source, assigner);
+ }
+ }
+
+ public void deregister(final IoTDBDataRegionSource source) {
+ final String pipeID = source.getPipeName() + "_" + source.getCreationTime();
+ final PipeDataNodeRemainingEventAndTimeOperator remainingOperator =
+ remainingEventAndTimeOperatorMap.get(pipeID);
+ if (remainingOperator != null) {
+ remainingOperator.deregister(source);
+ }
+
+ final PipeDataNodeCompletionOperator operator = completionOperatorMap.get(pipeID);
+ if (operator != null && source.getRealtimeSourceForCompletion() != null) {
+ operator.deregisterDataRegionSource(source.getRealtimeSourceForCompletion());
+ }
+ }
+
+ public void deregister(
+ final PipeRealtimeDataRegionSource source, final PipeDataRegionAssigner assigner) {
+ final PipeDataNodeCompletionOperator operator =
+ completionOperatorMap.get(source.getPipeName() + "_" + source.getCreationTime());
+ if (operator != null) {
+ operator.deregister(source, assigner);
+ }
+ }
+
+ public void markDataRegionCompleted(
+ final String pipeName,
+ final long creationTime,
+ final int dataRegionId,
+ final PipeTaskMeta pipeTaskMeta,
+ final long assignerEpoch,
+ final long generation,
+ final long completionSourceId,
+ final CommitterKey committerKey) {
+ final PipeDataNodeCompletionOperator operator =
+ completionOperatorMap.get(pipeName + "_" + creationTime);
+ if (operator != null) {
+ operator.markCompleted(
+ dataRegionId, pipeTaskMeta, assignerEpoch, generation, completionSourceId, committerKey);
+ }
+ }
+
+ public void markDataRegionInvalid(
Review Comment:
Addressed in c803c30b8229.
Applied: renamed the API to markDataRegionCompletionInvalid and updated its callers and tests.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java:
##########
@@ -94,42 +104,189 @@ public PipeDataRegionAssigner(final int dataRegionId) {
}
public void publishToAssign(final PipeRealtimeEvent event) {
+ publicationLock.lock();
+ try {
+ final EnrichedEvent innerEvent = event.getEvent();
+ final boolean isDataEvent = !(innerEvent instanceof PipeHeartbeatEvent);
Review Comment:
Addressed in c803c30b8229.
Added Event.isControlEvent() and moved the classification out of the assigner. Heartbeat, progress-report, and termination events identify themselves as control events. The assigner and collector now use the event abstraction for data-generation/failure classification, so future control events do not require another instanceof check here.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java:
##########
@@ -64,6 +69,11 @@ public class PipeDataRegionAssigner implements Closeable {
private final DisruptorQueue disruptor;
private final int dataRegionId;
+ private final long assignerEpoch = ASSIGNER_EPOCH_GENERATOR.incrementAndGet();
+ private final AtomicLong publishedDataGeneration = new AtomicLong(0);
+ private final AtomicLong publicationFailureEpoch = new AtomicLong(0);
+ private final AtomicLong completionInvalidationEpoch = new AtomicLong(0);
+ private final ReentrantLock publicationLock = new ReentrantLock();
Review Comment:
Addressed in c803c30b8229.
Added comments for all five fields: assignerEpoch identifies the assigner incarnation; publishedDataGeneration counts data mutations/publication attempts; publicationFailureEpoch records unrecovered delivery failures; completionInvalidationEpoch invalidates a full-FLUSH token when an insert may escape its snapshot; publicationLock serializes these changes with publication so a barrier cannot overtake data.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java:
##########
@@ -94,42 +104,189 @@ public PipeDataRegionAssigner(final int dataRegionId) {
}
public void publishToAssign(final PipeRealtimeEvent event) {
+ publicationLock.lock();
+ try {
+ final EnrichedEvent innerEvent = event.getEvent();
+ final boolean isDataEvent = !(innerEvent instanceof PipeHeartbeatEvent);
+ if (isDataEvent) {
+ publishedDataGeneration.incrementAndGet();
+ }
+ publishToAssignInternal(event, isDataEvent);
+ } finally {
+ publicationLock.unlock();
+ }
+ }
+
+ public void publishDataEventToAssign(final Supplier<PipeRealtimeEvent> eventSupplier) {
+ publishDataEventToAssign(eventSupplier, false);
+ }
+
+ public void publishInsertDataEventToAssign(final Supplier<PipeRealtimeEvent> eventSupplier) {
+ publishDataEventToAssign(eventSupplier, true);
+ }
Review Comment:
Addressed in c803c30b8229.
Added Javadoc to both entry points. publishDataEventToAssign handles TsFile-close/deletion publication and counts the attempt before supplier execution, including supplier failure. A close callback belongs to the in-progress flush and preserves its token. publishInsertDataEventToAssign additionally invalidates the token because an insert can create a working processor outside the flush snapshot. Token changes and publication share the same lock.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java:
##########
@@ -94,42 +104,189 @@ public PipeDataRegionAssigner(final int dataRegionId) {
}
public void publishToAssign(final PipeRealtimeEvent event) {
+ publicationLock.lock();
+ try {
+ final EnrichedEvent innerEvent = event.getEvent();
+ final boolean isDataEvent = !(innerEvent instanceof PipeHeartbeatEvent);
+ if (isDataEvent) {
+ publishedDataGeneration.incrementAndGet();
+ }
+ publishToAssignInternal(event, isDataEvent);
+ } finally {
+ publicationLock.unlock();
+ }
+ }
+
+ public void publishDataEventToAssign(final Supplier<PipeRealtimeEvent> eventSupplier) {
+ publishDataEventToAssign(eventSupplier, false);
+ }
+
+ public void publishInsertDataEventToAssign(final Supplier<PipeRealtimeEvent> eventSupplier) {
+ publishDataEventToAssign(eventSupplier, true);
+ }
+
+ private void publishDataEventToAssign(
+ final Supplier<PipeRealtimeEvent> eventSupplier, final boolean invalidateCompletionBarrier) {
+ publicationLock.lock();
+ try {
+ publishedDataGeneration.incrementAndGet();
+ if (invalidateCompletionBarrier) {
+ completionInvalidationEpoch.incrementAndGet();
+ }
+ final PipeRealtimeEvent event;
+ try {
+ event = eventSupplier.get();
+ } catch (final RuntimeException | Error e) {
+ markPublicationFailed();
+ throw e;
+ }
+ if (event != null) {
+ publishToAssignInternal(event, true);
+ } else {
+ markPublicationFailed();
+ }
+ } finally {
+ publicationLock.unlock();
+ }
+ }
+
+ private void publishToAssignInternal(final PipeRealtimeEvent event, final boolean isDataEvent) {
+ final EnrichedEvent innerEvent = event.getEvent();
if (!event.increaseReferenceCount(PipeDataRegionAssigner.class.getName())) {
+ if (isDataEvent) {
+ markPublicationFailed();
+ }
LOGGER.warn(DataNodePipeMessages.THE_REFERENCE_COUNT_OF_THE_REALTIME_EVENT, event);
return;
}
- final EnrichedEvent innerEvent = event.getEvent();
eventCounter.increaseEventCount(innerEvent);
if (innerEvent instanceof PipeHeartbeatEvent) {
((PipeHeartbeatEvent) innerEvent).onPublished();
}
- synchronized (this) {
- if (disruptor.isClosed()) {
- onAssignedHook(event);
- return;
- }
- inFlightPublishCount++;
- }
-
+ boolean shouldReleaseDirectly = false;
boolean isPublished = false;
try {
- isPublished = disruptor.publishOrDrop(event);
- } finally {
+ if (innerEvent instanceof PipeHeartbeatEvent) {
+ final PipeHeartbeatEvent heartbeatEvent = (PipeHeartbeatEvent) innerEvent;
+ if (heartbeatEvent.isCompletionBarrier()) {
+ heartbeatEvent.bindCompletionBarrier(assignerEpoch, publishedDataGeneration.get());
+ }
+ }
+
synchronized (this) {
- inFlightPublishCount--;
- if (inFlightPublishCount == 0) {
- notifyAll();
+ if (disruptor.isClosed()) {
+ shouldReleaseDirectly = true;
+ } else {
+ inFlightPublishCount++;
+ }
+ }
+
+ if (!shouldReleaseDirectly) {
+ try {
+ isPublished = disruptor.publishOrDrop(event);
+ } finally {
+ synchronized (this) {
+ inFlightPublishCount--;
+ if (inFlightPublishCount == 0) {
+ notifyAll();
+ }
+ }
}
}
+ } catch (final RuntimeException | Error e) {
+ if (isDataEvent) {
+ markPublicationFailed();
+ }
+ throw e;
}
- if (!isPublished) {
+ if (shouldReleaseDirectly || !isPublished) {
+ if (isDataEvent) {
+ markPublicationFailed();
+ }
onAssignedHook(event);
}
}
+ private void markPublicationFailed() {
+ publicationFailureEpoch.incrementAndGet();
+ }
+
+ /**
+ * Advances the published data generation for a data event that was ignored before publication.
+ * This deliberately keeps the current completion token valid: full-flush close callbacks may be
+ * ignored when no source listens to TsFile events, but their generations still need to be covered
+ * by the barrier for that flush.
+ */
+ public void invalidateCompletion() {
+ invalidateCompletion(false);
+ }
+
+ /**
+ * Advances the published data generation and invalidates the current completion token for an
+ * ignored insert. The insert may create a working TsFile processor that the in-progress full
+ * flush does not cover, so its barrier must not be published.
+ */
+ public void invalidateCompletionBarrier() {
+ invalidateCompletion(true);
+ }
Review Comment:
Addressed in c803c30b8229.
Documented the term at publishDataEventToAssign: data events carry insert/tablet, TsFile, or deletion data. Heartbeats and progress reports are internal control events and do not advance the data generation. The distinction now lives in Event.isControlEvent(), with a test showing that a non-heartbeat control event preserves the generation and current flush token.
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Pipe] Add reliable DataNode completion metric [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 merged PR #18563:
URL: https://github.com/apache/iotdb/pull/18563
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Pipe] Add reliable DataNode completion metric [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #18563:
URL: https://github.com/apache/iotdb/pull/18563#discussion_r4128937701
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java:
##########
@@ -60,6 +61,8 @@ public class PipeEventCollector implements EventCollector {
private final int regionId;
+ private final long completionSourceId;
+
Review Comment:
Is it possible to make the ID not only for completion?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java:
##########
@@ -246,6 +251,7 @@ private void collectEvent(final Event event) {
if (event instanceof EnrichedEvent) {
final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
if (!enrichedEvent.increaseReferenceCount(PipeEventCollector.class.getName())) {
+ markDataRegionCompletionInvalid(enrichedEvent);
Review Comment:
Why would a skipped event invalidate the completion mark.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java:
##########
@@ -228,7 +228,13 @@ protected boolean executeOnce() throws Exception {
.markTsFileCollectInvocationCount(
pipeNameWithCreationTime, outputEventCollector.getCollectInvocationCount());
} else if (event instanceof PipeHeartbeatEvent) {
- pipeProcessor.process(event, outputEventCollector);
+ // A completion barrier is an internal ordering event. It must not be swallowed or
+ // transformed by a user processor.
+ if (((PipeHeartbeatEvent) event).isCompletionBarrier()) {
+ outputEventCollector.collect(event);
+ } else {
+ pipeProcessor.process(event, outputEventCollector);
+ }
Review Comment:
May use a higher-level abstraction like Event.isUserVisible or Event.shouldSkipProcessing
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java:
##########
@@ -275,6 +284,20 @@ private void collectEvent(final Event event) {
if (pendingQueue.offer(event)) {
collectInvocationCount.incrementAndGet();
+ } else if (event instanceof EnrichedEvent) {
+ markDataRegionCompletionInvalid((EnrichedEvent) event);
+ }
+ }
+
+ private void markDataRegionCompletionInvalid(final EnrichedEvent event) {
+ if (!(event instanceof PipeHeartbeatEvent) && event.getPipeName() != null && regionId >= 0) {
+ PipeDataNodeSinglePipeMetrics.getInstance()
+ .markDataRegionInvalid(
Review Comment:
Is the method name `markDataRegionInvalid` proper?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -118,6 +123,20 @@
public class PipeDataNodeTaskAgent extends PipeTaskAgent {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeDataNodeTaskAgent.class);
+ private static final Set<String> COMPLETION_SUPPORTED_SOURCES =
+ Set.of(
+ BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_SOURCE.getPipePluginName());
+ private static final Set<String> COMPLETION_SUPPORTED_SINKS =
+ Set.of(
+ BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SSL_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SINK.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SSL_SINK.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SYNC_SINK.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK.getPipePluginName());
Review Comment:
Is it possible to support air-gap?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -118,6 +123,20 @@
public class PipeDataNodeTaskAgent extends PipeTaskAgent {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeDataNodeTaskAgent.class);
+ private static final Set<String> COMPLETION_SUPPORTED_SOURCES =
+ Set.of(
+ BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_SOURCE.getPipePluginName());
+ private static final Set<String> COMPLETION_SUPPORTED_SINKS =
+ Set.of(
+ BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SSL_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SINK.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SSL_SINK.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SYNC_SINK.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK.getPipePluginName());
Review Comment:
How about adding a field `supportCompletionMark` for these enums?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java:
##########
@@ -64,6 +69,11 @@ public class PipeDataRegionAssigner implements Closeable {
private final DisruptorQueue disruptor;
private final int dataRegionId;
+ private final long assignerEpoch = ASSIGNER_EPOCH_GENERATOR.incrementAndGet();
+ private final AtomicLong publishedDataGeneration = new AtomicLong(0);
+ private final AtomicLong publicationFailureEpoch = new AtomicLong(0);
+ private final AtomicLong completionInvalidationEpoch = new AtomicLong(0);
+ private final ReentrantLock publicationLock = new ReentrantLock();
Review Comment:
Add some comments to explain these vars.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetrics.java:
##########
@@ -194,26 +222,100 @@ private void removeAutoGauge(final String pipeID) {
Metric.PIPE_TSFILE_EVENT_TRANSFER_TIME.toString(),
Tag.NAME.toString(),
operator.getPipeName());
- remainingEventAndTimeOperatorMap.remove(pipeID);
}
//////////////////////////// register & deregister (pipe integration) ////////////////////////////
public void register(final IoTDBDataRegionSource source) {
// The metric is global thus the regionId is omitted
final String pipeID = source.getPipeName() + "_" + source.getCreationTime();
- remainingEventAndTimeOperatorMap
- .computeIfAbsent(
+ final PipeDataNodeRemainingEventAndTimeOperator remainingOperator =
+ remainingEventAndTimeOperatorMap.computeIfAbsent(
pipeID,
k ->
new PipeDataNodeRemainingEventAndTimeOperator(
- source.getPipeName(), source.getCreationTime()))
- .register(source);
+ source.getPipeName(), source.getCreationTime()));
+ remainingOperator.register(source);
+ final PipeDataNodeCompletionOperator completionOperator =
+ completionOperatorMap.computeIfAbsent(
+ pipeID,
+ k ->
+ new PipeDataNodeCompletionOperator(
+ () -> remainingOperator.getRemainingNonHeartbeatEvents() == 0,
+ () ->
+ PipeDataNodeAgent.task()
+ .getPipeCompletionSnapshot(
+ source.getPipeName(), source.getCreationTime())));
+ if (source.getRealtimeSourceForCompletion() != null) {
+ completionOperator.registerDataRegionSource(source.getRealtimeSourceForCompletion());
+ }
if (Objects.nonNull(metricService)) {
createMetrics(pipeID);
}
}
+ public void register(
+ final PipeRealtimeDataRegionSource source, final PipeDataRegionAssigner assigner) {
+ final String pipeID = source.getPipeName() + "_" + source.getCreationTime();
+ final PipeDataNodeCompletionOperator operator = completionOperatorMap.get(pipeID);
+ if (operator != null) {
+ operator.register(source, assigner);
+ }
+ }
+
+ public void deregister(final IoTDBDataRegionSource source) {
+ final String pipeID = source.getPipeName() + "_" + source.getCreationTime();
+ final PipeDataNodeRemainingEventAndTimeOperator remainingOperator =
+ remainingEventAndTimeOperatorMap.get(pipeID);
+ if (remainingOperator != null) {
+ remainingOperator.deregister(source);
+ }
+
+ final PipeDataNodeCompletionOperator operator = completionOperatorMap.get(pipeID);
+ if (operator != null && source.getRealtimeSourceForCompletion() != null) {
+ operator.deregisterDataRegionSource(source.getRealtimeSourceForCompletion());
+ }
+ }
+
+ public void deregister(
+ final PipeRealtimeDataRegionSource source, final PipeDataRegionAssigner assigner) {
+ final PipeDataNodeCompletionOperator operator =
+ completionOperatorMap.get(source.getPipeName() + "_" + source.getCreationTime());
+ if (operator != null) {
+ operator.deregister(source, assigner);
+ }
+ }
+
+ public void markDataRegionCompleted(
+ final String pipeName,
+ final long creationTime,
+ final int dataRegionId,
+ final PipeTaskMeta pipeTaskMeta,
+ final long assignerEpoch,
+ final long generation,
+ final long completionSourceId,
+ final CommitterKey committerKey) {
+ final PipeDataNodeCompletionOperator operator =
+ completionOperatorMap.get(pipeName + "_" + creationTime);
+ if (operator != null) {
+ operator.markCompleted(
+ dataRegionId, pipeTaskMeta, assignerEpoch, generation, completionSourceId, committerKey);
+ }
+ }
+
+ public void markDataRegionInvalid(
Review Comment:
markDataRegionCompletionInvalid
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java:
##########
@@ -94,42 +104,189 @@ public PipeDataRegionAssigner(final int dataRegionId) {
}
public void publishToAssign(final PipeRealtimeEvent event) {
+ publicationLock.lock();
+ try {
+ final EnrichedEvent innerEvent = event.getEvent();
+ final boolean isDataEvent = !(innerEvent instanceof PipeHeartbeatEvent);
Review Comment:
Better to push down the method to Event, in case we add more control-flow events in the future.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java:
##########
@@ -2538,6 +2538,78 @@ public void syncCloseAllWorkingTsFileProcessors() {
}
}
+ /**
+ * Closes all working TsFile processors and waits for processors that were already closing.
+ *
+ * <p>This stronger variant is used by a full flush before a Pipe completion barrier is published.
+ * Unlike {@link #syncCloseAllWorkingTsFileProcessors()}, failures are propagated to the caller so
+ * a failed flush can never publish a completion barrier.
+ */
+ public void syncCloseAllWorkingAndClosingTsFileProcessors()
+ throws InterruptedException, ExecutionException {
+ final Set<TsFileProcessor> targets = new HashSet<>();
+ final Set<TsFileProcessor> targetsToClose = new HashSet<>();
+ final List<Future<?>> closingFutures = new ArrayList<>();
+
+ writeLock("syncCloseAllWorkingAndClosingTsFileProcessors");
+ try {
+ targets.addAll(closingSequenceTsFileProcessor);
+ targets.addAll(closingUnSequenceTsFileProcessor);
+ for (final TsFileProcessor target : targets) {
+ closingFutures.add(target.getCloseFuture());
+ }
+
+ targetsToClose.addAll(workSequenceTsFileProcessors.values());
+ targetsToClose.addAll(workUnsequenceTsFileProcessors.values());
+ targets.addAll(targetsToClose);
+ } finally {
+ writeUnlock();
+ }
+
+ int closedProcessorCount = 0;
+ while (!targetsToClose.isEmpty()) {
+ final List<Future<?>> currentFutures = new ArrayList<>();
+ final Set<TsFileProcessor> targetsWaitingForOrdinaryFlush = new HashSet<>();
+
+ writeLock("syncCloseAllWorkingAndClosingTsFileProcessors");
+ try {
+ final Iterator<TsFileProcessor> iterator = targetsToClose.iterator();
+ while (iterator.hasNext()) {
+ final TsFileProcessor target = iterator.next();
+ final Future<?> future =
+ asyncCloseOneTsFileProcessorForFullFlush(target.isSequence(), target);
+ currentFutures.add(future);
+ if (target.alreadyMarkedClosing()) {
+ closingFutures.add(future);
+ iterator.remove();
+ closedProcessorCount++;
+ } else {
+ targetsWaitingForOrdinaryFlush.add(target);
+ }
+ }
+ } finally {
+ writeUnlock();
Review Comment:
What does an ordinary flush refer to?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -677,6 +696,90 @@ public Set<Integer> getPipeTaskRegionIdSet(final String pipeName, final long cre
: pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().keySet();
}
+ public Pair<Boolean, Map<Integer, PipeRealtimeDataRegionSource>> getPipeCompletionSnapshot(
+ final String pipeName, final long creationTime) {
Review Comment:
Add a comment to explain the return value.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java:
##########
@@ -94,42 +104,189 @@ public PipeDataRegionAssigner(final int dataRegionId) {
}
public void publishToAssign(final PipeRealtimeEvent event) {
+ publicationLock.lock();
+ try {
+ final EnrichedEvent innerEvent = event.getEvent();
+ final boolean isDataEvent = !(innerEvent instanceof PipeHeartbeatEvent);
+ if (isDataEvent) {
+ publishedDataGeneration.incrementAndGet();
+ }
+ publishToAssignInternal(event, isDataEvent);
+ } finally {
+ publicationLock.unlock();
+ }
+ }
+
+ public void publishDataEventToAssign(final Supplier<PipeRealtimeEvent> eventSupplier) {
+ publishDataEventToAssign(eventSupplier, false);
+ }
+
+ public void publishInsertDataEventToAssign(final Supplier<PipeRealtimeEvent> eventSupplier) {
+ publishDataEventToAssign(eventSupplier, true);
+ }
Review Comment:
Explain these.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java:
##########
@@ -94,42 +104,189 @@ public PipeDataRegionAssigner(final int dataRegionId) {
}
public void publishToAssign(final PipeRealtimeEvent event) {
+ publicationLock.lock();
+ try {
+ final EnrichedEvent innerEvent = event.getEvent();
+ final boolean isDataEvent = !(innerEvent instanceof PipeHeartbeatEvent);
+ if (isDataEvent) {
+ publishedDataGeneration.incrementAndGet();
+ }
+ publishToAssignInternal(event, isDataEvent);
+ } finally {
+ publicationLock.unlock();
+ }
+ }
+
+ public void publishDataEventToAssign(final Supplier<PipeRealtimeEvent> eventSupplier) {
+ publishDataEventToAssign(eventSupplier, false);
+ }
+
+ public void publishInsertDataEventToAssign(final Supplier<PipeRealtimeEvent> eventSupplier) {
+ publishDataEventToAssign(eventSupplier, true);
+ }
+
+ private void publishDataEventToAssign(
+ final Supplier<PipeRealtimeEvent> eventSupplier, final boolean invalidateCompletionBarrier) {
+ publicationLock.lock();
+ try {
+ publishedDataGeneration.incrementAndGet();
+ if (invalidateCompletionBarrier) {
+ completionInvalidationEpoch.incrementAndGet();
+ }
+ final PipeRealtimeEvent event;
+ try {
+ event = eventSupplier.get();
+ } catch (final RuntimeException | Error e) {
+ markPublicationFailed();
+ throw e;
+ }
+ if (event != null) {
+ publishToAssignInternal(event, true);
+ } else {
+ markPublicationFailed();
+ }
+ } finally {
+ publicationLock.unlock();
+ }
+ }
+
+ private void publishToAssignInternal(final PipeRealtimeEvent event, final boolean isDataEvent) {
+ final EnrichedEvent innerEvent = event.getEvent();
if (!event.increaseReferenceCount(PipeDataRegionAssigner.class.getName())) {
+ if (isDataEvent) {
+ markPublicationFailed();
+ }
LOGGER.warn(DataNodePipeMessages.THE_REFERENCE_COUNT_OF_THE_REALTIME_EVENT, event);
return;
}
- final EnrichedEvent innerEvent = event.getEvent();
eventCounter.increaseEventCount(innerEvent);
if (innerEvent instanceof PipeHeartbeatEvent) {
((PipeHeartbeatEvent) innerEvent).onPublished();
}
- synchronized (this) {
- if (disruptor.isClosed()) {
- onAssignedHook(event);
- return;
- }
- inFlightPublishCount++;
- }
-
+ boolean shouldReleaseDirectly = false;
boolean isPublished = false;
try {
- isPublished = disruptor.publishOrDrop(event);
- } finally {
+ if (innerEvent instanceof PipeHeartbeatEvent) {
+ final PipeHeartbeatEvent heartbeatEvent = (PipeHeartbeatEvent) innerEvent;
+ if (heartbeatEvent.isCompletionBarrier()) {
+ heartbeatEvent.bindCompletionBarrier(assignerEpoch, publishedDataGeneration.get());
+ }
+ }
+
synchronized (this) {
- inFlightPublishCount--;
- if (inFlightPublishCount == 0) {
- notifyAll();
+ if (disruptor.isClosed()) {
+ shouldReleaseDirectly = true;
+ } else {
+ inFlightPublishCount++;
+ }
+ }
+
+ if (!shouldReleaseDirectly) {
+ try {
+ isPublished = disruptor.publishOrDrop(event);
+ } finally {
+ synchronized (this) {
+ inFlightPublishCount--;
+ if (inFlightPublishCount == 0) {
+ notifyAll();
+ }
+ }
}
}
+ } catch (final RuntimeException | Error e) {
+ if (isDataEvent) {
+ markPublicationFailed();
+ }
+ throw e;
}
- if (!isPublished) {
+ if (shouldReleaseDirectly || !isPublished) {
+ if (isDataEvent) {
+ markPublicationFailed();
+ }
onAssignedHook(event);
}
}
+ private void markPublicationFailed() {
+ publicationFailureEpoch.incrementAndGet();
+ }
+
+ /**
+ * Advances the published data generation for a data event that was ignored before publication.
+ * This deliberately keeps the current completion token valid: full-flush close callbacks may be
+ * ignored when no source listens to TsFile events, but their generations still need to be covered
+ * by the barrier for that flush.
+ */
+ public void invalidateCompletion() {
+ invalidateCompletion(false);
+ }
+
+ /**
+ * Advances the published data generation and invalidates the current completion token for an
+ * ignored insert. The insert may create a working TsFile processor that the in-progress full
+ * flush does not cover, so its barrier must not be published.
+ */
+ public void invalidateCompletionBarrier() {
+ invalidateCompletion(true);
+ }
Review Comment:
What does a data event refer to?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskSourceStage.java:
##########
@@ -107,4 +109,17 @@ public void dropSubtask() throws PipeException {
public EventSupplier getEventSupplier() {
return pipeExtractor::supply;
}
+
+ public PipeExtractor getPipeExtractor() {
+ return pipeExtractor;
+ }
+
+ public long getCompletionSourceId() {
+ if (!(pipeExtractor instanceof IoTDBDataRegionSource)) {
+ return Long.MIN_VALUE;
+ }
+ final PipeRealtimeDataRegionSource realtimeSource =
+ ((IoTDBDataRegionSource) pipeExtractor).getRealtimeSourceForCompletion();
+ return realtimeSource == null ? Long.MIN_VALUE : realtimeSource.getCompletionSourceId();
+ }
Review Comment:
Why must it be a realtime source
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java:
##########
@@ -212,6 +387,9 @@ private void assignToSource(
}
Review Comment:
Is it possible to abstract these as Event.bindSource?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -677,6 +696,90 @@ public Set<Integer> getPipeTaskRegionIdSet(final String pipeName, final long cre
: pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().keySet();
}
+ public Pair<Boolean, Map<Integer, PipeRealtimeDataRegionSource>> getPipeCompletionSnapshot(
+ final String pipeName, final long creationTime) {
+ if (!tryReadLockWithTimeOutInMs(10)) {
+ return new Pair<>(false, Collections.emptyMap());
+ }
+
+ try {
+ final PipeMeta pipeMeta = pipeMetaKeeper.getPipeMeta(pipeName, creationTime);
+ if (pipeMeta == null) {
+ return new Pair<>(false, Collections.emptyMap());
+ }
+
+ final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta();
+ final String sourceName =
+ staticMeta
+ .getSourceParameters()
+ .getStringOrDefault(
+ Arrays.asList(PipeSourceConstant.EXTRACTOR_KEY, PipeSourceConstant.SOURCE_KEY),
+ BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName());
+ final String processorName =
+ staticMeta
+ .getProcessorParameters()
+ .getStringOrDefault(
+ PipeProcessorConstant.PROCESSOR_KEY,
+ BuiltinPipePlugin.DO_NOTHING_PROCESSOR.getPipePluginName());
+ final PipeParameters sinkParameters = staticMeta.getSinkParameters();
+ final String sinkName = PipeSinkConstant.getConnectorOrSinkNameWithDefault(sinkParameters);
+ boolean supported =
+ staticMeta.getPipeType() == PipeType.USER
+ && COMPLETION_SUPPORTED_SOURCES.stream()
+ .anyMatch(name -> name.equalsIgnoreCase(sourceName))
+ && BuiltinPipePlugin.DO_NOTHING_PROCESSOR
+ .getPipePluginName()
+ .equalsIgnoreCase(processorName)
+ && COMPLETION_SUPPORTED_SINKS.stream()
+ .anyMatch(name -> name.equalsIgnoreCase(sinkName))
+ && PipeSinkConstant.CONNECTOR_LOAD_TSFILE_STRATEGY_SYNC_VALUE.equalsIgnoreCase(
+ sinkParameters.getStringOrDefault(
+ Arrays.asList(
+ PipeSinkConstant.CONNECTOR_LOAD_TSFILE_STRATEGY_KEY,
+ PipeSinkConstant.SINK_LOAD_TSFILE_STRATEGY_KEY),
+ PipeSinkConstant.CONNECTOR_LOAD_TSFILE_STRATEGY_SYNC_VALUE))
+ && pipeMeta.getRuntimeMeta().getStatus().get() == PipeStatus.RUNNING
+ && pipeMeta.getRuntimeMeta().getNodeId2PipeRuntimeExceptionMap().isEmpty()
+ && pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().values().stream()
+ .noneMatch(PipeTaskMeta::hasExceptionMessages);
Review Comment:
How about caching the result in pipeMeta?
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitManager.java:
##########
@@ -173,6 +173,16 @@ public CommitterKey getCommitterKey(
return generateCommitterKey(pipeName, creationTime, regionId);
}
+ public boolean isCurrentCommitterKey(final CommitterKey committerKey) {
+ return committerKey != null
+ && eventCommitterMap.containsKey(committerKey)
+ && committerKey.equals(
+ generateCommitterKey(
+ committerKey.getPipeName(),
+ committerKey.getCreationTime(),
+ committerKey.getRegionId()));
Review Comment:
Explain this
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Pipe] Add reliable DataNode completion metric [iotdb]
Posted by "Caideyipi (via GitHub)" <gi...@apache.org>.
Caideyipi commented on code in PR #18563:
URL: https://github.com/apache/iotdb/pull/18563#discussion_r4244552174
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java:
##########
@@ -246,6 +251,7 @@ private void collectEvent(final Event event) {
if (event instanceof EnrichedEvent) {
final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
if (!enrichedEvent.increaseReferenceCount(PipeEventCollector.class.getName())) {
+ markDataRegionCompletionInvalid(enrichedEvent);
Review Comment:
Addressed in c803c30b8229.
Added an explanation at the failure path. This skip means increaseReferenceCount failed: the event's data is already released/unavailable, rather than intentionally filtered by a pattern or privilege check. A later barrier cannot prove that the missing data was delivered, so the current source's completion state must stay invalid. The collector also invalidates completion on enqueue/committer failures; stale source instances cannot invalidate a replacement.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java:
##########
@@ -275,6 +284,20 @@ private void collectEvent(final Event event) {
if (pendingQueue.offer(event)) {
collectInvocationCount.incrementAndGet();
+ } else if (event instanceof EnrichedEvent) {
+ markDataRegionCompletionInvalid((EnrichedEvent) event);
+ }
+ }
+
+ private void markDataRegionCompletionInvalid(final EnrichedEvent event) {
+ if (!(event instanceof PipeHeartbeatEvent) && event.getPipeName() != null && regionId >= 0) {
+ PipeDataNodeSinglePipeMetrics.getInstance()
+ .markDataRegionInvalid(
Review Comment:
Addressed in c803c30b8229.
Renamed it to markDataRegionCompletionInvalid at the metrics API and all call sites. This makes clear that only the completion proof is invalidated, not the DataRegion itself.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java:
##########
@@ -60,6 +61,8 @@ public class PipeEventCollector implements EventCollector {
private final int regionId;
+ private final long completionSourceId;
+
Review Comment:
Addressed in c803c30b8229.
Renamed completionSourceId/getCompletionSourceId to sourceId/getSourceId. The ID now identifies the source instance generally: EnrichedEvent stores it, Event.bindSource attaches it during assignment/collection, and ordinary data/progress/heartbeat events can carry it. Completion uses the same lifecycle identity to reject callbacks from replaced sources.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java:
##########
@@ -228,7 +228,13 @@ protected boolean executeOnce() throws Exception {
.markTsFileCollectInvocationCount(
pipeNameWithCreationTime, outputEventCollector.getCollectInvocationCount());
} else if (event instanceof PipeHeartbeatEvent) {
- pipeProcessor.process(event, outputEventCollector);
+ // A completion barrier is an internal ordering event. It must not be swallowed or
+ // transformed by a user processor.
+ if (((PipeHeartbeatEvent) event).isCompletionBarrier()) {
+ outputEventCollector.collect(event);
+ } else {
+ pipeProcessor.process(event, outputEventCollector);
+ }
Review Comment:
Addressed in c803c30b8229.
Added Event.shouldSkipProcessing() with a default of false. Completion barriers override it to true, and PipeProcessorSubtask forwards such events unchanged before type-specific processor dispatch. Periodic heartbeats still go through the processor. Added coverage using a generic event, so the bypass no longer depends on a PipeHeartbeatEvent cast.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -118,6 +123,20 @@
public class PipeDataNodeTaskAgent extends PipeTaskAgent {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeDataNodeTaskAgent.class);
+ private static final Set<String> COMPLETION_SUPPORTED_SOURCES =
+ Set.of(
+ BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_SOURCE.getPipePluginName());
+ private static final Set<String> COMPLETION_SUPPORTED_SINKS =
+ Set.of(
+ BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SSL_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SINK.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SSL_SINK.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SYNC_SINK.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK.getPipePluginName());
Review Comment:
Addressed in c803c30b8229.
Added support for both iotdb-air-gap-connector and iotdb-air-gap-sink, retaining the sync TsFile-load requirement. Air-gap transfer waits synchronously for the receiver's one-byte ACK, and ordered commit cannot pass an unacknowledged earlier data event. Added a regression test in which a batch loses its ACK and is retried by a completion barrier before its data references are released.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -677,6 +696,90 @@ public Set<Integer> getPipeTaskRegionIdSet(final String pipeName, final long cre
: pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().keySet();
}
+ public Pair<Boolean, Map<Integer, PipeRealtimeDataRegionSource>> getPipeCompletionSnapshot(
+ final String pipeName, final long creationTime) {
Review Comment:
Addressed in c803c30b8229.
Added Javadoc explaining both values: the map contains the current realtime DataRegion sources by region ID; the boolean is true only for a supported, running, healthy Pipe with all expected DataRegion tasks present and ready. On false, the map may be partial/empty and cannot be used as evidence of completion.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java:
##########
@@ -118,6 +123,20 @@
public class PipeDataNodeTaskAgent extends PipeTaskAgent {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeDataNodeTaskAgent.class);
+ private static final Set<String> COMPLETION_SUPPORTED_SOURCES =
+ Set.of(
+ BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_SOURCE.getPipePluginName());
+ private static final Set<String> COMPLETION_SUPPORTED_SINKS =
+ Set.of(
+ BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SSL_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SINK.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SSL_SINK.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_SYNC_SINK.getPipePluginName(),
+ BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK.getPipePluginName());
Review Comment:
Addressed in c803c30b8229.
Added the supportCompletionMark field and supportsCompletionMark accessors to BuiltinPipePlugin. The supported source, do-nothing processor, Thrift aliases, and Air-gap aliases declare the capability alongside their enum entries; plugins without a proven ordered completion path default to false. Sync TsFile-load compatibility is checked separately in PipeStaticMeta.
--
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: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org