You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ju...@apache.org on 2026/02/09 09:13:59 UTC
(iotdb) branch air-gap-local updated: mipl
This is an automated email from the ASF dual-hosted git repository.
justinchen pushed a commit to branch air-gap-local
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/air-gap-local by this push:
new 3122f197b65 mipl
3122f197b65 is described below
commit 3122f197b652bf16fe5093a9d17c261294d4de23
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Mon Feb 9 17:13:41 2026 +0800
mipl
---
.../protocol/airgap/IoTDBAirGapReceiver.java | 30 +---------------------
.../apache/iotdb/commons/conf/CommonConfig.java | 8 +++++-
2 files changed, 8 insertions(+), 30 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiver.java
index 49e642371c6..5d2e7013b34 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiver.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiver.java
@@ -133,35 +133,7 @@ public class IoTDBAirGapReceiver extends WrappedRunnable {
.setVersion(ReadWriteIOUtils.readByte(byteBuffer))
.setType(ReadWriteIOUtils.readShort(byteBuffer))
.setBody(byteBuffer.slice());
- final TPipeTransferResp resp = agent.receive(req);
-
- final TSStatus status = resp.getStatus();
- if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- ok();
- } else if (status.getCode() == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()
- || status.getCode()
- == TSStatusCode.PIPE_RECEIVER_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode()) {
- LOGGER.info(
- "Pipe air gap receiver {}: TSStatus {} is encountered at the air gap receiver, will ignore.",
- receiverId,
- resp.getStatus());
- ok();
- } else if (status.getCode()
- == TSStatusCode.PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode()) {
- final long startTime = System.currentTimeMillis();
- while (System.currentTimeMillis() - startTime
- < PipeConfig.getInstance().getPipeAirGapRetryMaxMs()) {
- agent.receive(req);
- Thread.sleep(PipeConfig.getInstance().getPipeAirGapRetryLocalIntervalMs());
- }
- } else {
- LOGGER.warn(
- "Pipe air gap receiver {}: Handle data failed, status: {}, req: {}",
- receiverId,
- resp.getStatus(),
- req);
- fail();
- }
+ handleReq(req, System.currentTimeMillis());
} catch (final PipeConnectionException e) {
LOGGER.info(
"Pipe air gap receiver {}: Socket {} closed when listening to data. Because: {}",
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index b5cf9cd1a3c..cf68da89553 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.commons.client.property.ClientPoolProperty.DefaultProper
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.enums.HandleSystemErrorStrategy;
import org.apache.iotdb.commons.enums.PipeRateAverage;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.commons.utils.KillPoint.KillPoint;
import org.apache.iotdb.rpc.RpcUtils;
@@ -1630,8 +1631,13 @@ public class CommonConfig {
logger.info("pipeAirGapRetryLocalIntervalMs is set to {}.", pipeAirGapRetryLocalIntervalMs);
}
+ // < 0 : 0.8 * transfer timeout to avoid timeout
+ // = 0 : Disable retry
+ // > 0 : Explicit configuration
public long getPipeAirGapRetryMaxMs() {
- return pipeAirGapRetryMaxMs;
+ return pipeAirGapRetryMaxMs >= 0
+ ? pipeAirGapRetryMaxMs
+ : (long) (PipeConfig.getInstance().getPipeSinkTransferTimeoutMs() * 0.8);
}
public void setPipeAirGapRetryMaxMs(long pipeAirGapRetryMaxMs) {