This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 79c94598031 [To dev/1.3] Pipe: Handle transient resource failures
locally (#18381) (#18430)
79c94598031 is described below
commit 79c945980319bf0c9b7208d9b0e946d634093bf3
Author: Caideyipi <[email protected]>
AuthorDate: Wed Aug 12 09:32:47 2026 +0800
[To dev/1.3] Pipe: Handle transient resource failures locally (#18381)
(#18430)
* Pipe: Handle transient resource failures locally (#18381)
* Fix pipe auto-drop IT expected header
---
.../pipe/it/autocreate/IoTDBPipeAutoDropIT.java | 2 +-
.../heartbeat/DataNodeHeartbeatHandler.java | 3 +-
.../response/pipe/task/PipeTableResp.java | 1 +
.../iotdb/confignode/manager/ConfigManager.java | 3 +-
.../pipe/agent/task/PipeConfigNodeSubtask.java | 8 ++
.../pipe/agent/task/PipeConfigNodeTaskAgent.java | 5 ++
.../runtime/PipeRuntimeCoordinator.java | 7 +-
.../runtime/heartbeat/PipeHeartbeat.java | 27 +++++++
.../runtime/heartbeat/PipeHeartbeatParser.java | 1 +
.../runtime/heartbeat/PipeHeartbeatScheduler.java | 6 +-
.../consensus/response/pipe/PipeTableRespTest.java | 21 +++++
.../runtime/heartbeat/PipeHeartbeatParserTest.java | 55 +++++++++++++
.../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 9 +++
.../subtask/processor/PipeProcessorSubtask.java | 12 +++
.../agent/task/subtask/sink/PipeSinkSubtask.java | 8 ++
.../thrift/async/IoTDBDataRegionAsyncSink.java | 57 ++++++++++++-
.../common/header/ColumnHeaderConstant.java | 4 +-
.../execution/config/sys/pipe/ShowPipeTask.java | 9 +++
.../task/subtask/sink/PipeSinkSubtaskTest.java | 49 +++++++++++
.../config/sys/pipe/ShowPipeTaskTest.java | 73 +++++++++++++++++
.../pipe/PipeRuntimeSinkResourceException.java | 51 ++++++++++++
.../commons/pipe/agent/task/PipeTaskAgent.java | 10 +++
.../agent/task/meta/PipeTemporaryMetaInAgent.java | 16 +++-
.../task/meta/PipeTemporaryMetaInCoordinator.java | 84 ++++++++++++++++++-
.../task/subtask/PipeAbstractSinkSubtask.java | 28 +++++--
.../pipe/receiver/PipeReceiverStatusHandler.java | 21 ++---
.../pipe/resource/PipeRecentFailureCounter.java | 94 ++++++++++++++++++++++
.../pipe/resource/PipeResourceFailureType.java | 36 +++++++++
.../commons/pipe/resource/PipeStopStrategy.java | 76 +++++++++++++++++
.../agent/task/meta/PipeTemporaryMetaTest.java | 58 +++++++++++++
.../resource/PipeRecentFailureCounterTest.java | 48 +++++++++++
.../pipe/resource/PipeStopStrategyTest.java | 83 +++++++++++++++++++
.../thrift-commons/src/main/thrift/common.thrift | 1 +
.../src/main/thrift/confignode.thrift | 2 +-
.../src/main/thrift/datanode.thrift | 3 +-
35 files changed, 939 insertions(+), 32 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoDropIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoDropIT.java
index 09657e2deb2..8f3ca8895c7 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoDropIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoDropIT.java
@@ -145,7 +145,7 @@ public class IoTDBPipeAutoDropIT extends
AbstractPipeDualAutoIT {
TestUtils.assertDataEventuallyOnEnv(
senderEnv,
"show pipes",
-
"ID,CreationTime,State,PipeSource,PipeProcessor,PipeSink,ExceptionMessage,RemainingEventCount,EstimatedRemainingSeconds,",
+
"ID,CreationTime,State,PipeSource,PipeProcessor,PipeSink,ExceptionMessage,RemainingEventCount,EstimatedRemainingSeconds,RecentFailures,",
Collections.emptySet());
}
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java
index 3613c337862..0acfb9ec7c2 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java
@@ -147,7 +147,8 @@ public class DataNodeHeartbeatHandler implements
AsyncMethodCallback<TDataNodeHe
heartbeatResp.getPipeMetaList(),
heartbeatResp.getPipeCompletedList(),
heartbeatResp.getPipeRemainingEventCountList(),
- heartbeatResp.getPipeRemainingTimeList());
+ heartbeatResp.getPipeRemainingTimeList(),
+ heartbeatResp.getPipeRecentFailureList());
}
if (heartbeatResp.isSetConfirmedConfigNodeEndPoints()) {
loadManager
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
index f7a76b6174c..9293a57c208 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
@@ -204,6 +204,7 @@ public class PipeTableResp implements DataSet {
canCalculateOnLocal ? -1 : temporaryMeta.getGlobalRemainingEvents());
showPipeInfo.setEstimatedRemainingTime(
canCalculateOnLocal ? -1 : temporaryMeta.getGlobalRemainingTime());
+ showPipeInfo.setRecentFailures(temporaryMeta.getGlobalRecentFailures());
showPipeInfoList.add(showPipeInfo);
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index 2d34c20e8ae..61819bd6509 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -2576,7 +2576,8 @@ public class ConfigManager implements IManager {
resp.getPipeMetaList(),
resp.getPipeCompletedList(),
resp.getPipeRemainingEventCountList(),
- resp.getPipeRemainingTimeList());
+ resp.getPipeRemainingTimeList(),
+ resp.getPipeRecentFailureList());
return StatusUtils.OK;
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java
index 6daf3160c0c..bd11e27d9ee 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java
@@ -32,6 +32,7 @@ import
org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSinkRuntimeEnviro
import
org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
+import org.apache.iotdb.commons.pipe.resource.PipeResourceFailureType;
import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent;
import
org.apache.iotdb.confignode.manager.pipe.metric.sink.PipeConfigRegionSinkMetrics;
import org.apache.iotdb.confignode.manager.pipe.source.IoTDBConfigRegionSource;
@@ -240,6 +241,13 @@ public class PipeConfigNodeSubtask extends
PipeAbstractSinkSubtask {
PipeConfigNodeAgent.runtime().report(event, exception);
}
+ @Override
+ protected void reportResourceFailure(
+ final EnrichedEvent event, final PipeResourceFailureType failureType) {
+ PipeConfigNodeAgent.task()
+ .recordPipeResourceFailure(event.getPipeName(),
event.getCreationTime(), failureType);
+ }
+
//////////////////////////// APIs provided for metric framework
////////////////////////////
public String getPipeName() {
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgent.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgent.java
index 3fe0b1757cf..d60b8b3b4c7 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgent.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgent.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInAgent;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent;
@@ -218,6 +219,7 @@ public class PipeConfigNodeTaskAgent extends PipeTaskAgent {
final List<ByteBuffer> pipeMetaBinaryList = new ArrayList<>();
final List<Long> pipeRemainingEventCountList = new ArrayList<>();
final List<Double> pipeRemainingTimeList = new ArrayList<>();
+ final List<Map<String, Long>> pipeRecentFailureList = new ArrayList<>();
try {
for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) {
pipeMetaBinaryList.add(pipeMeta.serialize());
@@ -232,6 +234,8 @@ public class PipeConfigNodeTaskAgent extends PipeTaskAgent {
pipeRemainingEventCountList.add(remainingEventCount);
pipeRemainingTimeList.add(estimatedRemainingTime);
+ pipeRecentFailureList.add(
+ ((PipeTemporaryMetaInAgent)
pipeMeta.getTemporaryMeta()).getRecentFailures());
logger.ifPresent(
l ->
@@ -248,6 +252,7 @@ public class PipeConfigNodeTaskAgent extends PipeTaskAgent {
resp.setPipeMetaList(pipeMetaBinaryList);
resp.setPipeRemainingEventCountList(pipeRemainingEventCountList);
resp.setPipeRemainingTimeList(pipeRemainingTimeList);
+ resp.setPipeRecentFailureList(pipeRecentFailureList);
}
@Override
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java
index eae99126523..83ee13d6407 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java
@@ -31,6 +31,7 @@ import
org.apache.iotdb.confignode.manager.pipe.coordinator.runtime.heartbeat.Pi
import java.nio.ByteBuffer;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicReference;
@@ -107,13 +108,15 @@ public class PipeRuntimeCoordinator implements
IClusterStatusSubscriber {
final List<ByteBuffer> pipeMetaByteBufferListFromDataNode,
/* @Nullable */ final List<Boolean> pipeCompletedListFromAgent,
/* @Nullable */ final List<Long> pipeRemainingEventCountListFromAgent,
- /* @Nullable */ final List<Double> pipeRemainingTimeListFromAgent) {
+ /* @Nullable */ final List<Double> pipeRemainingTimeListFromAgent,
+ /* @Nullable */ final List<Map<String, Long>>
pipeRecentFailureListFromAgent) {
pipeHeartbeatScheduler.parseHeartbeat(
dataNodeId,
new PipeHeartbeat(
pipeMetaByteBufferListFromDataNode,
pipeCompletedListFromAgent,
pipeRemainingEventCountListFromAgent,
- pipeRemainingTimeListFromAgent));
+ pipeRemainingTimeListFromAgent,
+ pipeRecentFailureListFromAgent));
}
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeat.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeat.java
index e7e9d2cd97d..933e7f04e1d 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeat.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeat.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import java.nio.ByteBuffer;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -33,12 +34,27 @@ public class PipeHeartbeat {
private final Map<PipeStaticMeta, Boolean> isCompletedMap = new HashMap<>();
private final Map<PipeStaticMeta, Long> remainingEventCountMap = new
HashMap<>();
private final Map<PipeStaticMeta, Double> remainingTimeMap = new HashMap<>();
+ private final Map<PipeStaticMeta, Map<String, Long>> recentFailuresMap = new
HashMap<>();
public PipeHeartbeat(
final List<ByteBuffer> pipeMetaByteBufferListFromAgent,
/* @Nullable */ final List<Boolean> pipeCompletedListFromAgent,
/* @Nullable */ final List<Long> pipeRemainingEventCountListFromAgent,
/* @Nullable */ final List<Double> pipeRemainingTimeListFromAgent) {
+ this(
+ pipeMetaByteBufferListFromAgent,
+ pipeCompletedListFromAgent,
+ pipeRemainingEventCountListFromAgent,
+ pipeRemainingTimeListFromAgent,
+ null);
+ }
+
+ public PipeHeartbeat(
+ final List<ByteBuffer> pipeMetaByteBufferListFromAgent,
+ /* @Nullable */ final List<Boolean> pipeCompletedListFromAgent,
+ /* @Nullable */ final List<Long> pipeRemainingEventCountListFromAgent,
+ /* @Nullable */ final List<Double> pipeRemainingTimeListFromAgent,
+ /* @Nullable */ final List<Map<String, Long>>
pipeRecentFailureListFromAgent) {
// Shall not reach here, just in case
if (Objects.isNull(pipeMetaByteBufferListFromAgent)) {
return;
@@ -63,6 +79,13 @@ public class PipeHeartbeat {
Objects.nonNull(pipeRemainingTimeListFromAgent)
? pipeRemainingTimeListFromAgent.get(i)
: 0d);
+ recentFailuresMap.put(
+ pipeMeta.getStaticMeta(),
+ Objects.nonNull(pipeRecentFailureListFromAgent)
+ && i < pipeRecentFailureListFromAgent.size()
+ && Objects.nonNull(pipeRecentFailureListFromAgent.get(i))
+ ? new HashMap<>(pipeRecentFailureListFromAgent.get(i))
+ : Collections.emptyMap());
}
}
@@ -86,6 +109,10 @@ public class PipeHeartbeat {
return remainingTimeMap.get(pipeStaticMeta);
}
+ public Map<String, Long> getRecentFailures(final PipeStaticMeta
pipeStaticMeta) {
+ return recentFailuresMap.get(pipeStaticMeta);
+ }
+
public boolean isEmpty() {
return pipeMetaMap.isEmpty();
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
index e67a832ea3e..55bb018f6d0 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
@@ -191,6 +191,7 @@ public class PipeHeartbeatParser {
// Record statistics
temporaryMeta.setRemainingEvent(nodeId,
pipeHeartbeat.getRemainingEventCount(staticMeta));
temporaryMeta.setRemainingTime(nodeId,
pipeHeartbeat.getRemainingTime(staticMeta));
+ temporaryMeta.setRecentFailures(nodeId,
pipeHeartbeat.getRecentFailures(staticMeta));
final Map<Integer, PipeTaskMeta> pipeTaskMetaMapFromCoordinator =
pipeMetaFromCoordinator.getRuntimeMeta().getConsensusGroupId2TaskMetaMap();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatScheduler.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatScheduler.java
index d4611bd2f2e..9e41ab771f5 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatScheduler.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatScheduler.java
@@ -112,7 +112,8 @@ public class PipeHeartbeatScheduler {
resp.getPipeMetaList(),
resp.getPipeCompletedList(),
resp.getPipeRemainingEventCountList(),
- resp.getPipeRemainingTimeList())));
+ resp.getPipeRemainingTimeList(),
+ resp.getPipeRecentFailureList())));
// config node heartbeat
try {
@@ -124,7 +125,8 @@ public class PipeHeartbeatScheduler {
configNodeResp.getPipeMetaList(),
null,
configNodeResp.getPipeRemainingEventCountList(),
- configNodeResp.getPipeRemainingTimeList()));
+ configNodeResp.getPipeRemainingTimeList(),
+ configNodeResp.getPipeRecentFailureList()));
} catch (final Exception e) {
LOGGER.warn("Failed to collect pipe meta list from config node task
agent", e);
}
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
index 072f98a7500..068f85a0a3d 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java
@@ -25,9 +25,11 @@ import
org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import
org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInCoordinator;
import org.apache.iotdb.confignode.consensus.response.pipe.task.PipeTableResp;
import org.apache.iotdb.confignode.manager.ConfigManager;
import org.apache.iotdb.confignode.manager.node.NodeManager;
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo;
import org.apache.iotdb.confignode.service.ConfigNode;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -146,6 +148,25 @@ public class PipeTableRespTest {
Assert.assertEquals(3, allPipeTableResp.getAllPipeMeta().size());
}
+ @Test
+ public void testConvertToTShowPipeRespAggregatesRecentFailures() {
+ final PipeTableResp pipeTableResp = constructPipeTableResp();
+ final PipeTemporaryMetaInCoordinator temporaryMeta =
+ (PipeTemporaryMetaInCoordinator)
pipeTableResp.getAllPipeMeta().get(0).getTemporaryMeta();
+ final Map<String, Long> firstNodeFailures = new HashMap<>();
+ firstNodeFailures.put("network_timeout", 10L);
+ firstNodeFailures.put("memory_timeout", 15L);
+ temporaryMeta.setRecentFailures(1, firstNodeFailures);
+ final Map<String, Long> secondNodeFailures = new HashMap<>();
+ secondNodeFailures.put("network_timeout", 2L);
+ temporaryMeta.setRecentFailures(2, secondNodeFailures);
+
+ final TShowPipeInfo showPipeInfo =
+ pipeTableResp.convertToTShowPipeResp().getPipeInfoList().get(0);
+ Assert.assertEquals(Long.valueOf(12),
showPipeInfo.getRecentFailures().get("network_timeout"));
+ Assert.assertEquals(Long.valueOf(15),
showPipeInfo.getRecentFailures().get("memory_timeout"));
+ }
+
@Test
public void testConvertToTShowPipeRespIncludesPreDeleteStatus() {
final PipeTableResp pipeTableResp = constructPipeTableResp();
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
index b080f8f51b1..ebe8cc19573 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
@@ -27,6 +27,7 @@ import
org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import
org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInCoordinator;
import
org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipePlanV2;
import org.apache.iotdb.confignode.manager.ConfigManager;
import org.apache.iotdb.confignode.manager.ProcedureManager;
@@ -243,6 +244,50 @@ public class PipeHeartbeatParserTest {
verify(context.procedureManager,
never()).pipeHandleMetaChange(anyBoolean(), anyBoolean());
}
+ @Test
+ public void testParseHeartbeatAggregatesRecentFailuresFromAllDataNodes()
throws Exception {
+
CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false);
+
+ final String pipeName = "resourceFailurePipe";
+ final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo();
+ createPipe(pipeTaskInfo, pipeName, PipeStatus.RUNNING);
+ final PipeMeta pipeMeta = pipeTaskInfo.getPipeMetaByPipeName(pipeName);
+ final ParserTestContext context = createParserTestContext(2, pipeTaskInfo);
+
+ final Map<String, Long> firstNodeFailures = new HashMap<>();
+ firstNodeFailures.put("network_timeout", 10L);
+ final Map<String, Long> secondNodeFailures = new HashMap<>();
+ secondNodeFailures.put("network_timeout", 2L);
+ secondNodeFailures.put("memory_timeout", 15L);
+
+ context.parser.parseHeartbeat(1, createPipeHeartbeat(pipeMeta,
firstNodeFailures));
+ context.parser.parseHeartbeat(2, createPipeHeartbeat(pipeMeta,
secondNodeFailures));
+
+ final PipeTemporaryMetaInCoordinator temporaryMeta =
+ (PipeTemporaryMetaInCoordinator) pipeMeta.getTemporaryMeta();
+ Assert.assertEquals(
+ Long.valueOf(12),
temporaryMeta.getGlobalRecentFailures().get("network_timeout"));
+ Assert.assertEquals(
+ Long.valueOf(15),
temporaryMeta.getGlobalRecentFailures().get("memory_timeout"));
+ verify(context.procedureManager,
never()).pipeHandleMetaChange(anyBoolean(), anyBoolean());
+ }
+
+ @Test
+ public void testPipeHeartbeatTreatsNullRecentFailureMapAsEmpty() throws
Exception {
+ final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo();
+ createPipe(pipeTaskInfo, "nullFailureMapPipe", PipeStatus.RUNNING);
+ final PipeMeta pipeMeta =
pipeTaskInfo.getPipeMetaByPipeName("nullFailureMapPipe");
+ final PipeHeartbeat heartbeat =
+ new PipeHeartbeat(
+ Collections.singletonList(pipeMeta.serialize()),
+ Collections.singletonList(false),
+ Collections.singletonList(0L),
+ Collections.singletonList(0d),
+ Collections.<Map<String, Long>>singletonList(null));
+
+
Assert.assertTrue(heartbeat.getRecentFailures(pipeMeta.getStaticMeta()).isEmpty());
+ }
+
private ParserTestContext createParserTestContext(final int
registeredDataNodeCount) {
return createParserTestContext(registeredDataNodeCount, new
PipeTaskInfo());
}
@@ -331,6 +376,16 @@ public class PipeHeartbeatParserTest {
return new PipeHeartbeat(Collections.emptyList(), null, null, null);
}
+ private PipeHeartbeat createPipeHeartbeat(
+ final PipeMeta pipeMeta, final Map<String, Long> recentFailures) throws
Exception {
+ return new PipeHeartbeat(
+ Collections.singletonList(pipeMeta.serialize()),
+ Collections.singletonList(false),
+ Collections.singletonList(0L),
+ Collections.singletonList(0d),
+ Collections.singletonList(recentFailures));
+ }
+
private static class ParserTestContext {
private final PipeHeartbeatParser parser;
private final ProcedureManager procedureManager;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
index 54e8f1752ea..a28cdb95211 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
@@ -39,6 +39,7 @@ import
org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInAgent;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeType;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
@@ -405,6 +406,7 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
final List<Boolean> pipeCompletedList = new ArrayList<>();
final List<Long> pipeRemainingEventCountList = new ArrayList<>();
final List<Double> pipeRemainingTimeList = new ArrayList<>();
+ final List<Map<String, Long>> pipeRecentFailureList = new ArrayList<>();
try {
for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) {
pipeMetaBinaryList.add(pipeMeta.serialize());
@@ -438,6 +440,8 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
pipeCompletedList.add(isCompleted);
pipeRemainingEventCountList.add(remainingEventAndTime.getLeft());
pipeRemainingTimeList.add(remainingEventAndTime.getRight());
+ pipeRecentFailureList.add(
+ ((PipeTemporaryMetaInAgent)
pipeMeta.getTemporaryMeta()).getRecentFailures());
logger.ifPresent(
l ->
@@ -457,6 +461,7 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
resp.setPipeCompletedList(pipeCompletedList);
resp.setPipeRemainingEventCountList(pipeRemainingEventCountList);
resp.setPipeRemainingTimeList(pipeRemainingTimeList);
+ resp.setPipeRecentFailureList(pipeRecentFailureList);
PipeInsertionDataNodeListener.getInstance().listenToHeartbeat(true);
}
@@ -486,6 +491,7 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
final List<Boolean> pipeCompletedList = new ArrayList<>();
final List<Long> pipeRemainingEventCountList = new ArrayList<>();
final List<Double> pipeRemainingTimeList = new ArrayList<>();
+ final List<Map<String, Long>> pipeRecentFailureList = new ArrayList<>();
try {
for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) {
pipeMetaBinaryList.add(pipeMeta.serialize());
@@ -519,6 +525,8 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
pipeCompletedList.add(isCompleted);
pipeRemainingEventCountList.add(remainingEventAndTime.getLeft());
pipeRemainingTimeList.add(remainingEventAndTime.getRight());
+ pipeRecentFailureList.add(
+ ((PipeTemporaryMetaInAgent)
pipeMeta.getTemporaryMeta()).getRecentFailures());
logger.ifPresent(
l ->
@@ -538,6 +546,7 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
resp.setPipeCompletedList(pipeCompletedList);
resp.setPipeRemainingEventCountList(pipeRemainingEventCountList);
resp.setPipeRemainingTimeList(pipeRemainingTimeList);
+ resp.setPipeRecentFailureList(pipeRecentFailureList);
PipeInsertionDataNodeListener.getInstance().listenToHeartbeat(true);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
index 193693c5a95..3c4b3d55019 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
@@ -27,6 +27,7 @@ import
org.apache.iotdb.commons.pipe.agent.task.execution.PipeSubtaskScheduler;
import
org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitManager;
import org.apache.iotdb.commons.pipe.agent.task.subtask.PipeReportableSubtask;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.commons.pipe.resource.PipeResourceFailureType;
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
@@ -189,6 +190,7 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
}
decreaseReferenceCountAndReleaseLastEvent(event, shouldReport);
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
+ recordResourceFailure(event, PipeResourceFailureType.MEMORY_TIMEOUT);
PipeLogger.log(
LOGGER::info,
"Temporarily out of memory in pipe event processing, will wait for
the memory to release. Message: %s",
@@ -196,6 +198,7 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
return false;
} catch (final Exception e) {
if (ExceptionUtils.getRootCause(e) instanceof
PipeRuntimeOutOfMemoryCriticalException) {
+ recordResourceFailure(event, PipeResourceFailureType.MEMORY_TIMEOUT);
PipeLogger.log(
LOGGER::info,
"Temporarily out of memory in pipe event processing, will wait for
the memory to release. Message: %s",
@@ -298,4 +301,13 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
protected void report(final EnrichedEvent event, final PipeRuntimeException
exception) {
PipeDataNodeAgent.runtime().report(event, exception);
}
+
+ private void recordResourceFailure(final Event event, final
PipeResourceFailureType failureType) {
+ if (event instanceof EnrichedEvent) {
+ final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
+ PipeDataNodeAgent.task()
+ .recordPipeResourceFailure(
+ enrichedEvent.getPipeName(), enrichedEvent.getCreationTime(),
failureType);
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
index 91a588ce699..aa87f0850c0 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
@@ -27,6 +27,7 @@ import
org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
import
org.apache.iotdb.commons.pipe.agent.task.subtask.PipeAbstractSinkSubtask;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.commons.pipe.resource.PipeResourceFailureType;
import org.apache.iotdb.commons.pipe.sink.protocol.IoTDBSink;
import
org.apache.iotdb.commons.pipe.sink.protocol.PipeConnectorWithEventDiscard;
import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils;
@@ -512,4 +513,11 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
lastExceptionTime = Long.MAX_VALUE;
PipeDataNodeAgent.runtime().report(event, exception);
}
+
+ @Override
+ protected void reportResourceFailure(
+ final EnrichedEvent event, final PipeResourceFailureType failureType) {
+ PipeDataNodeAgent.task()
+ .recordPipeResourceFailure(event.getPipeName(),
event.getCreationTime(), failureType);
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
index 3d5d6af26a0..52d22ac9774 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
@@ -24,11 +24,15 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.ThriftClient;
import
org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfigurableException;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkResourceException;
import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.commons.pipe.resource.PipeResourceFailureType;
+import org.apache.iotdb.commons.pipe.resource.PipeStopStrategy;
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import org.apache.iotdb.commons.pipe.sink.protocol.IoTDBSink;
+import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import
org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionWritePlanEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
@@ -76,6 +80,8 @@ import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
+import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -113,6 +119,9 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
private final BlockingQueue<TsFileInsertionEvent> retryTsFileQueue = new
LinkedBlockingQueue<>();
private final PipeDataRegionEventCounter retryEventQueueEventCounter =
new PipeDataRegionEventCounter();
+ // Guarded by this. Events need identity semantics because the same payload
may compare equal.
+ private final Map<Event, PipeResourceFailureType>
retryEvent2ResourceFailureType =
+ new IdentityHashMap<>();
private IoTDBDataNodeAsyncClientManager clientManager;
private IoTDBDataNodeAsyncClientManager transferTsFileClientManager;
@@ -566,6 +575,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
final Event polledEvent;
if (!retryEventQueue.isEmpty()) {
peekedEvent = retryEventQueue.peek();
+ retryEvent2ResourceFailureType.remove(peekedEvent);
if (peekedEvent instanceof PipeInsertNodeTabletInsertionEvent) {
retryTransfer((PipeInsertNodeTabletInsertionEvent) peekedEvent);
@@ -584,6 +594,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
return;
}
peekedEvent = retryTsFileQueue.peek();
+ retryEvent2ResourceFailureType.remove(peekedEvent);
retryTransfer((PipeTsFileInsertionEvent) peekedEvent);
polledEvent = retryTsFileQueue.poll();
}
@@ -624,6 +635,12 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
+ ", tsfile events: "
+ retryEventQueueEventCounter.getTsFileInsertionEventCount()
+ ").";
+ final PipeResourceFailureType retryQueueResourceFailureType =
+ getRetryQueueResourceFailureType();
+ if (retryQueueResourceFailureType != null) {
+ throw new PipeRuntimeSinkResourceException(
+ message, retryQueueResourceFailureType, true);
+ }
throw isConnectionException
? new PipeConnectionException(message)
: new PipeException(message);
@@ -686,6 +703,13 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
*/
@SuppressWarnings("java:S899")
public void addFailureEventToRetryQueue(final Event event, final Exception
e) {
+ addFailureEventToRetryQueue(event, e, null);
+ }
+
+ private synchronized void addFailureEventToRetryQueue(
+ final Event event, final Exception e, final Set<Pair<String, Long>>
failureRecordedPipes) {
+ final PipeResourceFailureType resourceFailureType =
+ PipeStopStrategy.getResourceFailureType(e, null);
isConnectionException =
e instanceof PipeConnectionException ||
ThriftClient.isConnectionBroken(e);
if (event instanceof EnrichedEvent) {
@@ -706,6 +730,23 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
return;
}
+ if (resourceFailureType != null && event instanceof EnrichedEvent) {
+ final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
+ final Pair<String, Long> pipeKey =
+ new Pair<>(enrichedEvent.getPipeName(),
enrichedEvent.getCreationTime());
+ if (failureRecordedPipes == null || failureRecordedPipes.add(pipeKey)) {
+ PipeDataNodeAgent.task()
+ .recordPipeResourceFailure(
+ enrichedEvent.getPipeName(), enrichedEvent.getCreationTime(),
resourceFailureType);
+ }
+ }
+
+ if (resourceFailureType == null) {
+ retryEvent2ResourceFailureType.remove(event);
+ } else {
+ retryEvent2ResourceFailureType.put(event, resourceFailureType);
+ }
+
if (event instanceof PipeTsFileInsertionEvent) {
retryTsFileQueue.offer((PipeTsFileInsertionEvent) event);
retryEventQueueEventCounter.increaseEventCount(event);
@@ -732,7 +773,17 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
*/
public void addFailureEventsToRetryQueue(
final Iterable<EnrichedEvent> events, final Exception e) {
- events.forEach(event -> addFailureEventToRetryQueue(event, e));
+ final Set<Pair<String, Long>> failureRecordedPipes = new HashSet<>();
+ events.forEach(event -> addFailureEventToRetryQueue(event, e,
failureRecordedPipes));
+ }
+
+ private synchronized PipeResourceFailureType
getRetryQueueResourceFailureType() {
+ for (final PipeResourceFailureType failureType :
PipeResourceFailureType.values()) {
+ if (retryEvent2ResourceFailureType.containsValue(failureType)) {
+ return failureType;
+ }
+ }
+ return null;
}
public boolean isEnableSendTsFileLimit() {
@@ -875,6 +926,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
&& isDroppedPipe((EnrichedEvent) event, committerKey)) {
((EnrichedEvent)
event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
retryEventQueueEventCounter.decreaseEventCount(event);
+ retryEvent2ResourceFailureType.remove(event);
return true;
}
return false;
@@ -886,6 +938,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
&& isDroppedPipe((EnrichedEvent) event, committerKey)) {
((EnrichedEvent)
event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
retryEventQueueEventCounter.decreaseEventCount(event);
+ retryEvent2ResourceFailureType.remove(event);
return true;
}
return false;
@@ -938,10 +991,12 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink {
final Event event =
retryTsFileQueue.isEmpty() ? retryEventQueue.poll() :
retryTsFileQueue.poll();
retryEventQueueEventCounter.decreaseEventCount(event);
+ retryEvent2ResourceFailureType.remove(event);
if (event instanceof EnrichedEvent) {
((EnrichedEvent)
event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
}
}
+ retryEvent2ResourceFailureType.clear();
}
//////////////////////// APIs provided for metric framework
////////////////////////
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/ColumnHeaderConstant.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/ColumnHeaderConstant.java
index 24d9509e0f3..92712daee52 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/ColumnHeaderConstant.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/ColumnHeaderConstant.java
@@ -181,6 +181,7 @@ public class ColumnHeaderConstant {
public static final String EXCEPTION_MESSAGE = "ExceptionMessage";
public static final String REMAINING_EVENT_COUNT = "RemainingEventCount";
public static final String ESTIMATED_REMAINING_SECONDS =
"EstimatedRemainingSeconds";
+ public static final String RECENT_FAILURES = "RecentFailures";
// column names for select into
public static final String SOURCE_DEVICE = "SourceDevice";
@@ -450,7 +451,8 @@ public class ColumnHeaderConstant {
new ColumnHeader(PIPE_CONNECTOR, TSDataType.TEXT),
new ColumnHeader(EXCEPTION_MESSAGE, TSDataType.TEXT),
new ColumnHeader(REMAINING_EVENT_COUNT, TSDataType.TEXT),
- new ColumnHeader(ESTIMATED_REMAINING_SECONDS, TSDataType.TEXT));
+ new ColumnHeader(ESTIMATED_REMAINING_SECONDS, TSDataType.TEXT),
+ new ColumnHeader(RECENT_FAILURES, TSDataType.TEXT));
public static final List<ColumnHeader> showTopicColumnHeaders =
ImmutableList.of(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTask.java
index 16ee005a089..4703c4ebca1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTask.java
@@ -41,6 +41,7 @@ import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.Pair;
import java.util.List;
+import java.util.TreeMap;
import java.util.stream.Collectors;
public class ShowPipeTask implements IConfigTask {
@@ -119,6 +120,14 @@ public class ShowPipeTask implements IConfigTask {
? String.format("%.2f", remainingTime)
: "Unknown",
TSFileConfig.STRING_CHARSET));
+ builder
+ .getColumnBuilder(9)
+ .writeBinary(
+ new Binary(
+ tPipeInfo.isSetRecentFailures()
+ ? new TreeMap<>(tPipeInfo.getRecentFailures()).toString()
+ : "{}",
+ TSFileConfig.STRING_CHARSET));
builder.declarePosition();
}
final DatasetHeader datasetHeader =
DatasetHeaderFactory.getShowPipeHeader();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
index 8960e7fea96..6c5e79ff68b 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskTest.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.pipe.agent.task.subtask.sink;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkCriticalException;
@@ -26,6 +27,7 @@ import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfi
import
org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.commons.pipe.resource.PipeResourceFailureType;
import
org.apache.iotdb.commons.pipe.sink.protocol.PipeConnectorWithEventDiscard;
import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
@@ -328,9 +330,42 @@ public class PipeSinkSubtaskTest {
}
}
+ @Test
+ public void
testClientBorrowFailureRetriesLocallyWithoutReportingCriticalException() {
+ final long originalSleepIntervalInitMs =
+
CommonDescriptor.getInstance().getConfig().getPipeSinkSubtaskSleepIntervalInitMs();
+ final long originalSleepIntervalMaxMs =
+
CommonDescriptor.getInstance().getConfig().getPipeSinkSubtaskSleepIntervalMaxMs();
+
CommonDescriptor.getInstance().getConfig().setPipeSinkSubtaskSleepIntervalInitMs(1);
+
CommonDescriptor.getInstance().getConfig().setPipeSinkSubtaskSleepIntervalMaxMs(2);
+
+ final PipeConnector connector = mock(PipeConnector.class);
+ final UnboundedBlockingPendingQueue<Event> pendingQueue =
+ mock(UnboundedBlockingPendingQueue.class);
+ final CapturingPipeSinkSubtask subtask = new
CapturingPipeSinkSubtask(pendingQueue, connector);
+
+ try {
+ subtask.handleExceptionForTest(
+ mock(EnrichedEvent.class), new ClientManagerException("client pool
exhausted"));
+
+ Assert.assertEquals(
+ PipeResourceFailureType.NETWORK_TIMEOUT,
subtask.getReportedResourceFailureType());
+ Assert.assertNull(subtask.getReportedException());
+ } finally {
+ subtask.close();
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setPipeSinkSubtaskSleepIntervalInitMs(originalSleepIntervalInitMs);
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setPipeSinkSubtaskSleepIntervalMaxMs(originalSleepIntervalMaxMs);
+ }
+ }
+
private static class CapturingPipeSinkSubtask extends PipeSinkSubtask {
private PipeRuntimeException reportedException;
+ private PipeResourceFailureType reportedResourceFailureType;
private CapturingPipeSinkSubtask(
final UnboundedBlockingPendingQueue<Event> pendingQueue, final
PipeConnector connector) {
@@ -353,10 +388,24 @@ public class PipeSinkSubtaskTest {
return reportedException;
}
+ private PipeResourceFailureType getReportedResourceFailureType() {
+ return reportedResourceFailureType;
+ }
+
+ private void handleExceptionForTest(final EnrichedEvent event, final
Exception exception) {
+ handleException(event, exception);
+ }
+
@Override
protected void report(final EnrichedEvent event, final
PipeRuntimeException exception) {
reportedException = exception;
}
+
+ @Override
+ protected void reportResourceFailure(
+ final EnrichedEvent event, final PipeResourceFailureType failureType) {
+ reportedResourceFailureType = failureType;
+ }
}
private static class BlockingHandshakeConnector
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTaskTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTaskTest.java
new file mode 100644
index 00000000000..6bd8676199c
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/pipe/ShowPipeTaskTest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.iotdb.db.queryengine.plan.execution.config.sys.pipe;
+
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo;
+import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
+import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import com.google.common.util.concurrent.SettableFuture;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+public class ShowPipeTaskTest {
+
+ @Test
+ public void testBuildTSBlockWritesRecentFailures() throws Exception {
+ final TShowPipeInfo failedPipe = createPipeInfo("failed_pipe");
+ final Map<String, Long> recentFailures = new HashMap<>();
+ recentFailures.put("network_timeout", 10L);
+ recentFailures.put("memory_timeout", 15L);
+ failedPipe.setRecentFailures(recentFailures);
+ final TShowPipeInfo normalPipe = createPipeInfo("normal_pipe");
+
+ final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ ShowPipeTask.buildTSBlock(Arrays.asList(failedPipe, normalPipe), future);
+
+ final ConfigTaskResult result = future.get();
+ final TsBlock resultSet = result.getResultSet();
+
+ assertEquals(TSStatusCode.SUCCESS_STATUS, result.getStatusCode());
+ assertEquals(
+ ColumnHeaderConstant.RECENT_FAILURES,
result.getResultSetHeader().getRespColumns().get(9));
+ assertEquals(2, resultSet.getPositionCount());
+ assertEquals(
+ "{memory_timeout=15, network_timeout=10}",
resultSet.getColumn(9).getBinary(0).toString());
+ assertEquals("{}", resultSet.getColumn(9).getBinary(1).toString());
+ }
+
+ private TShowPipeInfo createPipeInfo(final String pipeName) {
+ return new TShowPipeInfo(
+ pipeName,
+ 1L,
+ "RUNNING",
+ "{source=iotdb-source}",
+ "{processor=do-nothing-processor}",
+ "{sink=iotdb-thrift-sink}",
+ "");
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeSinkResourceException.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeSinkResourceException.java
new file mode 100644
index 00000000000..efc41d88654
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeSinkResourceException.java
@@ -0,0 +1,51 @@
+/*
+ * 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.iotdb.commons.exception.pipe;
+
+import org.apache.iotdb.commons.pipe.resource.PipeResourceFailureType;
+
+public class PipeRuntimeSinkResourceException
+ extends PipeRuntimeSinkNonReportTimeConfigurableException {
+
+ private final PipeResourceFailureType failureType;
+ private final boolean failureRecorded;
+
+ public PipeRuntimeSinkResourceException(
+ final String message, final PipeResourceFailureType failureType) {
+ this(message, failureType, false);
+ }
+
+ public PipeRuntimeSinkResourceException(
+ final String message,
+ final PipeResourceFailureType failureType,
+ final boolean failureRecorded) {
+ super(message, Long.MAX_VALUE);
+ this.failureType = failureType;
+ this.failureRecorded = failureRecorded;
+ }
+
+ public PipeResourceFailureType getFailureType() {
+ return failureType;
+ }
+
+ public boolean isFailureRecorded() {
+ return failureRecorded;
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
index 1eef328bc7e..21cd8369b01 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/PipeTaskAgent.java
@@ -35,6 +35,7 @@ import
org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInAgent;
import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
import
org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitManager;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.commons.pipe.resource.PipeResourceFailureType;
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import org.apache.iotdb.commons.pipe.sink.limiter.PipeEndPointRateLimiter;
import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
@@ -1248,6 +1249,15 @@ public abstract class PipeTaskAgent {
}
}
+ public void recordPipeResourceFailure(
+ final String pipeName, final long creationTime, final
PipeResourceFailureType failureType) {
+ final PipeMeta pipeMeta = pipeMetaKeeper.getPipeMeta(pipeName);
+ // Avoid recording a failure for a stale pipe before alter.
+ if (Objects.nonNull(pipeMeta) &&
pipeMeta.getStaticMeta().getCreationTime() == creationTime) {
+ ((PipeTemporaryMetaInAgent)
pipeMeta.getTemporaryMeta()).recordResourceFailure(failureType);
+ }
+ }
+
public int getPipeCount() {
return pipeMetaKeeper.getPipeMetaCount();
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaInAgent.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaInAgent.java
index 23914fa8d84..0f7b49225e0 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaInAgent.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaInAgent.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.commons.pipe.agent.task.meta;
import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
+import org.apache.iotdb.commons.pipe.resource.PipeRecentFailureCounter;
+import org.apache.iotdb.commons.pipe.resource.PipeResourceFailureType;
import java.util.Map;
import java.util.Objects;
@@ -30,6 +32,7 @@ public class PipeTemporaryMetaInAgent implements
PipeTemporaryMeta {
// Statistics
private final AtomicLong floatingMemoryUsageInByte = new AtomicLong(0L);
+ private final PipeRecentFailureCounter recentFailureCounter = new
PipeRecentFailureCounter();
// Object pool
private final String pipeNameWithCreationTime;
@@ -53,6 +56,14 @@ public class PipeTemporaryMetaInAgent implements
PipeTemporaryMeta {
return floatingMemoryUsageInByte.get();
}
+ public void recordResourceFailure(final PipeResourceFailureType failureType)
{
+ recentFailureCounter.record(failureType);
+ }
+
+ public Map<String, Long> getRecentFailures() {
+ return recentFailureCounter.getRecentFailures();
+ }
+
public String getPipeNameWithCreationTime() {
return pipeNameWithCreationTime;
}
@@ -87,12 +98,13 @@ public class PipeTemporaryMetaInAgent implements
PipeTemporaryMeta {
final PipeTemporaryMetaInAgent that = (PipeTemporaryMetaInAgent) o;
return Objects.equals(
this.floatingMemoryUsageInByte.get(),
that.floatingMemoryUsageInByte.get())
+ && Objects.equals(this.getRecentFailures(), that.getRecentFailures())
&& Objects.equals(this.regionId2CommitterKeyMap,
that.regionId2CommitterKeyMap);
}
@Override
public int hashCode() {
- return Objects.hash(floatingMemoryUsageInByte, regionId2CommitterKeyMap);
+ return Objects.hash(floatingMemoryUsageInByte, getRecentFailures(),
regionId2CommitterKeyMap);
}
@Override
@@ -100,6 +112,8 @@ public class PipeTemporaryMetaInAgent implements
PipeTemporaryMeta {
return "PipeTemporaryMeta{"
+ "floatingMemoryUsage="
+ floatingMemoryUsageInByte
+ + ", recentFailures="
+ + getRecentFailures()
+ ", regionId2CommitterKeyMap="
+ regionId2CommitterKeyMap
+ '}';
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaInCoordinator.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaInCoordinator.java
index ee127bbae44..2bac38f81bd 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaInCoordinator.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaInCoordinator.java
@@ -19,9 +19,14 @@
package org.apache.iotdb.commons.pipe.agent.task.meta;
+import org.apache.iotdb.commons.pipe.resource.PipeRecentFailureCounter;
+
import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -32,6 +37,8 @@ public class PipeTemporaryMetaInCoordinator implements
PipeTemporaryMeta {
Collections.newSetFromMap(new ConcurrentHashMap<>());
private final ConcurrentMap<Integer, Long> nodeId2RemainingEventMap = new
ConcurrentHashMap<>();
private final ConcurrentMap<Integer, Double> nodeId2RemainingTimeMap = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<Integer, RecentFailureSnapshot>
nodeId2RecentFailuresMap =
+ new ConcurrentHashMap<>();
public void markDataNodeCompleted(final int dataNodeId) {
completedDataNodeIds.add(dataNodeId);
@@ -49,6 +56,27 @@ public class PipeTemporaryMetaInCoordinator implements
PipeTemporaryMeta {
nodeId2RemainingTimeMap.put(dataNodeId, remainingTime);
}
+ public void setRecentFailures(final int dataNodeId, final Map<String, Long>
recentFailures) {
+ if (Objects.isNull(recentFailures) || recentFailures.isEmpty()) {
+ nodeId2RecentFailuresMap.remove(dataNodeId);
+ return;
+ }
+
+ final Map<String, Long> sanitizedFailures = new HashMap<>();
+ recentFailures.forEach(
+ (failureType, count) -> {
+ if (Objects.nonNull(failureType) && Objects.nonNull(count) && count
> 0) {
+ sanitizedFailures.put(failureType, count);
+ }
+ });
+ if (sanitizedFailures.isEmpty()) {
+ nodeId2RecentFailuresMap.remove(dataNodeId);
+ } else {
+ nodeId2RecentFailuresMap.put(
+ dataNodeId, new RecentFailureSnapshot(sanitizedFailures,
System.currentTimeMillis()));
+ }
+ }
+
public Set<Integer> getCompletedDataNodeIds() {
return completedDataNodeIds;
}
@@ -61,6 +89,23 @@ public class PipeTemporaryMetaInCoordinator implements
PipeTemporaryMeta {
return
nodeId2RemainingTimeMap.values().stream().reduce(Math::max).orElse(0d);
}
+ public Map<String, Long> getGlobalRecentFailures() {
+ final long earliestIncludedTime =
+ System.currentTimeMillis() - PipeRecentFailureCounter.WINDOW_MILLIS;
+ nodeId2RecentFailuresMap
+ .entrySet()
+ .removeIf(entry -> entry.getValue().reportTime < earliestIncludedTime);
+
+ final Map<String, Long> result = new TreeMap<>();
+ nodeId2RecentFailuresMap
+ .values()
+ .forEach(
+ snapshot ->
+ snapshot.recentFailures.forEach(
+ (failureType, count) -> result.merge(failureType, count,
Long::sum)));
+ return result;
+ }
+
@Override
public boolean equals(final Object o) {
if (this == o) {
@@ -72,12 +117,17 @@ public class PipeTemporaryMetaInCoordinator implements
PipeTemporaryMeta {
final PipeTemporaryMetaInCoordinator that =
(PipeTemporaryMetaInCoordinator) o;
return Objects.equals(this.completedDataNodeIds, that.completedDataNodeIds)
&& Objects.equals(this.nodeId2RemainingEventMap,
that.nodeId2RemainingEventMap)
- && Objects.equals(this.nodeId2RemainingTimeMap,
that.nodeId2RemainingTimeMap);
+ && Objects.equals(this.nodeId2RemainingTimeMap,
that.nodeId2RemainingTimeMap)
+ && Objects.equals(this.nodeId2RecentFailuresMap,
that.nodeId2RecentFailuresMap);
}
@Override
public int hashCode() {
- return Objects.hash(completedDataNodeIds, nodeId2RemainingEventMap,
nodeId2RemainingTimeMap);
+ return Objects.hash(
+ completedDataNodeIds,
+ nodeId2RemainingEventMap,
+ nodeId2RemainingTimeMap,
+ nodeId2RecentFailuresMap);
}
@Override
@@ -89,6 +139,36 @@ public class PipeTemporaryMetaInCoordinator implements
PipeTemporaryMeta {
+ nodeId2RemainingEventMap
+ ", nodeId2RemainingTimeMap"
+ nodeId2RemainingTimeMap
+ + ", nodeId2RecentFailuresMap="
+ + nodeId2RecentFailuresMap
+ '}';
}
+
+ private static class RecentFailureSnapshot {
+
+ private final Map<String, Long> recentFailures;
+ private final long reportTime;
+
+ private RecentFailureSnapshot(final Map<String, Long> recentFailures,
final long reportTime) {
+ this.recentFailures = recentFailures;
+ this.reportTime = reportTime;
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final RecentFailureSnapshot that = (RecentFailureSnapshot) o;
+ return Objects.equals(recentFailures, that.recentFailures);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(recentFailures);
+ }
+ }
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
index f3c1d72605c..df43914494e 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java
@@ -19,12 +19,13 @@
package org.apache.iotdb.commons.pipe.agent.task.subtask;
-import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkCriticalException;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfigurableException;
import org.apache.iotdb.commons.pipe.agent.task.execution.PipeSubtaskScheduler;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+import org.apache.iotdb.commons.pipe.resource.PipeResourceFailureType;
+import org.apache.iotdb.commons.pipe.resource.PipeStopStrategy;
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils;
import org.apache.iotdb.pipe.api.PipeConnector;
@@ -35,7 +36,6 @@ import org.apache.iotdb.pipe.api.exception.PipeException;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
-import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -321,12 +321,19 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
protected void handleException(final Event event, final Exception e) {
- if (e instanceof PipeRuntimeOutOfMemoryCriticalException
- || ExceptionUtils.getRootCause(e) instanceof
PipeRuntimeOutOfMemoryCriticalException) {
- PipeLogger.log(
- LOGGER::info,
- e,
- "Temporarily out of memory in pipe event transferring, will wait for
the memory to release.");
+ if (!PipeStopStrategy.accept(e, null)) {
+ final PipeResourceFailureType failureType =
PipeStopStrategy.getResourceFailureType(e, null);
+ if (event instanceof EnrichedEvent &&
!PipeStopStrategy.isResourceFailureRecorded(e)) {
+ reportResourceFailure((EnrichedEvent) event, failureType);
+ }
+ if (failureType == PipeResourceFailureType.MEMORY_TIMEOUT) {
+ PipeLogger.log(
+ LOGGER::info,
+ e,
+ "Temporarily out of memory in pipe event transferring, will wait
for the memory to release.");
+ } else {
+ sleep4NonReportException();
+ }
} else if (e instanceof PipeRuntimeSinkNonReportTimeConfigurableException)
{
if (lastExceptionTime == Long.MAX_VALUE) {
lastExceptionTime = System.currentTimeMillis();
@@ -372,4 +379,9 @@ public abstract class PipeAbstractSinkSubtask extends
PipeReportableSubtask {
clearReferenceCountAndReleaseLastEvent(event);
}
}
+
+ protected void reportResourceFailure(
+ final EnrichedEvent event, final PipeResourceFailureType failureType) {
+ // Do nothing by default for subtasks that do not expose resource failure
metrics.
+ }
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/PipeReceiverStatusHandler.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/PipeReceiverStatusHandler.java
index 604ad044dd6..273e04c1c00 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/PipeReceiverStatusHandler.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/PipeReceiverStatusHandler.java
@@ -21,7 +21,10 @@ package org.apache.iotdb.commons.pipe.receiver;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfigurableException;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkResourceException;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.commons.pipe.resource.PipeResourceFailureType;
+import org.apache.iotdb.commons.pipe.resource.PipeStopStrategy;
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.pipe.api.event.Event;
@@ -105,6 +108,14 @@ public class PipeReceiverStatusHandler {
final @Nullable String exceptionMessage,
final String recordMessage,
final boolean log4NoPrivileges) {
+ if (!PipeStopStrategy.accept(null, status)) {
+ PipeLogger.log(
+ LOGGER::info, "Temporary unavailable exception: will retry forever.
status: %s", status);
+ final PipeResourceFailureType failureType =
+ PipeStopStrategy.getResourceFailureType(null, status);
+ throw new PipeRuntimeSinkResourceException(exceptionMessage,
failureType);
+ }
+
switch (status.getCode()) {
case 200: // SUCCESS_STATUS
case 400: // REDIRECTION_RECOMMEND
@@ -118,16 +129,6 @@ public class PipeReceiverStatusHandler {
return;
}
- case 1808: // PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION
- {
- PipeLogger.log(
- LOGGER::info,
- "Temporary unavailable exception: will retry forever. status:
%s",
- status);
- throw new PipeRuntimeSinkNonReportTimeConfigurableException(
- exceptionMessage, Long.MAX_VALUE);
- }
-
case 1810: // PIPE_RECEIVER_USER_CONFLICT_EXCEPTION
case 1815: // PIPE_RECEIVER_PARALLEL_OR_USER_CONFLICT_EXCEPTION
if (!isRetryAllowedWhenConflictOccurs) {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/PipeRecentFailureCounter.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/PipeRecentFailureCounter.java
new file mode 100644
index 00000000000..974e647e02b
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/PipeRecentFailureCounter.java
@@ -0,0 +1,94 @@
+/*
+ * 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.iotdb.commons.pipe.resource;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class PipeRecentFailureCounter {
+
+ public static final long WINDOW_MILLIS = TimeUnit.MINUTES.toMillis(1);
+ private static final long BUCKET_MILLIS = TimeUnit.SECONDS.toMillis(1);
+ // Keep one extra slot for a failure exactly WINDOW_MILLIS old.
+ private static final int BUCKET_COUNT = (int) (WINDOW_MILLIS /
BUCKET_MILLIS) + 1;
+
+ private final Map<PipeResourceFailureType, FailureBucket[]> failureBuckets =
+ new EnumMap<>(PipeResourceFailureType.class);
+
+ public PipeRecentFailureCounter() {
+ for (final PipeResourceFailureType failureType :
PipeResourceFailureType.values()) {
+ final FailureBucket[] buckets = new FailureBucket[BUCKET_COUNT];
+ for (int i = 0; i < BUCKET_COUNT; ++i) {
+ buckets[i] = new FailureBucket();
+ }
+ failureBuckets.put(failureType, buckets);
+ }
+ }
+
+ public void record(final PipeResourceFailureType failureType) {
+ record(failureType, System.currentTimeMillis());
+ }
+
+ @TestOnly
+ synchronized void record(final PipeResourceFailureType failureType, final
long timestamp) {
+ final long bucketStartTime = Math.floorDiv(timestamp, BUCKET_MILLIS) *
BUCKET_MILLIS;
+ final int bucketIndex =
+ (int) Math.floorMod(Math.floorDiv(timestamp, BUCKET_MILLIS),
BUCKET_COUNT);
+ final FailureBucket bucket = failureBuckets.get(failureType)[bucketIndex];
+ if (bucket.startTime != bucketStartTime) {
+ bucket.startTime = bucketStartTime;
+ bucket.count = 0;
+ }
+ ++bucket.count;
+ }
+
+ public Map<String, Long> getRecentFailures() {
+ return getRecentFailures(System.currentTimeMillis());
+ }
+
+ @TestOnly
+ synchronized Map<String, Long> getRecentFailures(final long currentTime) {
+ final Map<String, Long> result = new LinkedHashMap<>();
+ final long earliestIncludedTime = currentTime - WINDOW_MILLIS;
+ for (final PipeResourceFailureType failureType :
PipeResourceFailureType.values()) {
+ long count = 0;
+ for (final FailureBucket bucket : failureBuckets.get(failureType)) {
+ if (bucket.startTime >= earliestIncludedTime && bucket.startTime <=
currentTime) {
+ count += bucket.count;
+ }
+ }
+ if (count > 0) {
+ result.put(failureType.getDisplayName(), count);
+ }
+ }
+ return Collections.unmodifiableMap(result);
+ }
+
+ private static class FailureBucket {
+
+ private long startTime = Long.MIN_VALUE;
+ private long count;
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/PipeResourceFailureType.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/PipeResourceFailureType.java
new file mode 100644
index 00000000000..722da021cbd
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/PipeResourceFailureType.java
@@ -0,0 +1,36 @@
+/*
+ * 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.iotdb.commons.pipe.resource;
+
+public enum PipeResourceFailureType {
+ NETWORK_TIMEOUT("network_timeout"),
+ MEMORY_TIMEOUT("memory_timeout"),
+ RECEIVER_UNAVAILABLE("receiver_unavailable");
+
+ private final String displayName;
+
+ PipeResourceFailureType(final String displayName) {
+ this.displayName = displayName;
+ }
+
+ public String getDisplayName() {
+ return displayName;
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/PipeStopStrategy.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/PipeStopStrategy.java
new file mode 100644
index 00000000000..4535bf57fe5
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/PipeStopStrategy.java
@@ -0,0 +1,76 @@
+/*
+ * 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.iotdb.commons.pipe.resource;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkResourceException;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import javax.annotation.Nullable;
+
+public final class PipeStopStrategy {
+
+ private PipeStopStrategy() {}
+
+ /**
+ * @return {@code true} if the failure may follow the normal stop/report
path, or {@code false} if
+ * it is a transient resource failure that must only be retried locally
+ */
+ public static boolean accept(
+ final @Nullable Exception exception, final @Nullable TSStatus status) {
+ return getResourceFailureType(exception, status) == null;
+ }
+
+ public static PipeResourceFailureType getResourceFailureType(
+ final @Nullable Exception exception, final @Nullable TSStatus status) {
+ Throwable current = exception;
+ while (current != null) {
+ if (current instanceof PipeRuntimeSinkResourceException) {
+ return ((PipeRuntimeSinkResourceException) current).getFailureType();
+ }
+ if (current instanceof PipeRuntimeOutOfMemoryCriticalException) {
+ return PipeResourceFailureType.MEMORY_TIMEOUT;
+ }
+ if (current instanceof ClientManagerException) {
+ return PipeResourceFailureType.NETWORK_TIMEOUT;
+ }
+ current = current.getCause();
+ }
+
+ return status != null
+ && status.getCode()
+ ==
TSStatusCode.PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode()
+ ? PipeResourceFailureType.RECEIVER_UNAVAILABLE
+ : null;
+ }
+
+ public static boolean isResourceFailureRecorded(final @Nullable Exception
exception) {
+ Throwable current = exception;
+ while (current != null) {
+ if (current instanceof PipeRuntimeSinkResourceException) {
+ return ((PipeRuntimeSinkResourceException)
current).isFailureRecorded();
+ }
+ current = current.getCause();
+ }
+ return false;
+ }
+}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaTest.java
new file mode 100644
index 00000000000..6e29ee267a5
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTemporaryMetaTest.java
@@ -0,0 +1,58 @@
+/*
+ * 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.iotdb.commons.pipe.agent.task.meta;
+
+import org.apache.iotdb.commons.pipe.resource.PipeResourceFailureType;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class PipeTemporaryMetaTest {
+
+ @Test
+ public void testRecentFailuresAreRecordedAndAggregated() {
+ final PipeTemporaryMetaInAgent agentMeta = new
PipeTemporaryMetaInAgent("test_pipe", 1L);
+ agentMeta.recordResourceFailure(PipeResourceFailureType.NETWORK_TIMEOUT);
+ agentMeta.recordResourceFailure(PipeResourceFailureType.NETWORK_TIMEOUT);
+ agentMeta.recordResourceFailure(PipeResourceFailureType.MEMORY_TIMEOUT);
+
+ Assert.assertEquals(Long.valueOf(2),
agentMeta.getRecentFailures().get("network_timeout"));
+ Assert.assertEquals(Long.valueOf(1),
agentMeta.getRecentFailures().get("memory_timeout"));
+
+ final PipeTemporaryMetaInCoordinator coordinatorMeta = new
PipeTemporaryMetaInCoordinator();
+ coordinatorMeta.setRecentFailures(1, agentMeta.getRecentFailures());
+ final Map<String, Long> secondNodeFailures = new HashMap<>();
+ secondNodeFailures.put("network_timeout", 3L);
+ coordinatorMeta.setRecentFailures(2, secondNodeFailures);
+
+ Assert.assertEquals(
+ Long.valueOf(5),
coordinatorMeta.getGlobalRecentFailures().get("network_timeout"));
+ Assert.assertEquals(
+ Long.valueOf(1),
coordinatorMeta.getGlobalRecentFailures().get("memory_timeout"));
+
+ coordinatorMeta.setRecentFailures(1, null);
+ Assert.assertEquals(
+ Long.valueOf(3),
coordinatorMeta.getGlobalRecentFailures().get("network_timeout"));
+
Assert.assertFalse(coordinatorMeta.getGlobalRecentFailures().containsKey("memory_timeout"));
+ }
+}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeRecentFailureCounterTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeRecentFailureCounterTest.java
new file mode 100644
index 00000000000..422dcbf2633
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeRecentFailureCounterTest.java
@@ -0,0 +1,48 @@
+/*
+ * 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.iotdb.commons.pipe.resource;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Map;
+
+public class PipeRecentFailureCounterTest {
+
+ @Test
+ public void testCountsOnlyFailuresWithinOneMinute() {
+ final PipeRecentFailureCounter counter = new PipeRecentFailureCounter();
+ final long now = 100_000L;
+
+ counter.record(
+ PipeResourceFailureType.NETWORK_TIMEOUT, now -
PipeRecentFailureCounter.WINDOW_MILLIS - 1);
+ counter.record(
+ PipeResourceFailureType.NETWORK_TIMEOUT, now -
PipeRecentFailureCounter.WINDOW_MILLIS);
+ counter.record(PipeResourceFailureType.NETWORK_TIMEOUT, now);
+ counter.record(PipeResourceFailureType.MEMORY_TIMEOUT, now);
+
+ final Map<String, Long> failures = counter.getRecentFailures(now);
+ Assert.assertEquals(Long.valueOf(2), failures.get("network_timeout"));
+ Assert.assertEquals(Long.valueOf(1), failures.get("memory_timeout"));
+
+ Assert.assertTrue(
+ counter.getRecentFailures(now + PipeRecentFailureCounter.WINDOW_MILLIS
+ 1).isEmpty());
+ }
+}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeStopStrategyTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeStopStrategyTest.java
new file mode 100644
index 00000000000..b68da403a42
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeStopStrategyTest.java
@@ -0,0 +1,83 @@
+/*
+ * 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.iotdb.commons.pipe.resource;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkResourceException;
+import org.apache.iotdb.pipe.api.exception.PipeConnectionException;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+
+public class PipeStopStrategyTest {
+
+ @Test
+ public void testClientBorrowFailureDoesNotStopPipe() {
+ final Exception failure =
+ new PipeConnectionException(
+ "transfer failed",
+ new IOException(new ClientManagerException(new IOException("client
pool exhausted"))));
+
+ Assert.assertFalse(PipeStopStrategy.accept(failure, null));
+ Assert.assertEquals(
+ PipeResourceFailureType.NETWORK_TIMEOUT,
+ PipeStopStrategy.getResourceFailureType(failure, null));
+ }
+
+ @Test
+ public void testMemoryFailuresDoNotStopPipe() {
+ final PipeRuntimeOutOfMemoryCriticalException exception =
+ new PipeRuntimeOutOfMemoryCriticalException("memory unavailable");
+ Assert.assertFalse(PipeStopStrategy.accept(exception, null));
+ Assert.assertEquals(
+ PipeResourceFailureType.MEMORY_TIMEOUT,
+ PipeStopStrategy.getResourceFailureType(exception, null));
+
+ final TSStatus status =
+ new
TSStatus(TSStatusCode.PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode());
+ Assert.assertFalse(PipeStopStrategy.accept(null, status));
+ Assert.assertEquals(
+ PipeResourceFailureType.RECEIVER_UNAVAILABLE,
+ PipeStopStrategy.getResourceFailureType(null, status));
+ }
+
+ @Test
+ public void testOtherFailuresKeepExistingStopPolicy() {
+ Assert.assertTrue(PipeStopStrategy.accept(new IOException("network
disconnected"), null));
+ Assert.assertTrue(
+ PipeStopStrategy.accept(
+ null, new
TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode())));
+ }
+
+ @Test
+ public void testRecordedMarkerIsFoundWhenResourceFailureIsWrapped() {
+ final Exception failure =
+ new IOException(
+ new PipeRuntimeSinkResourceException(
+ "retry queue exhausted",
PipeResourceFailureType.NETWORK_TIMEOUT, true));
+
+ Assert.assertTrue(PipeStopStrategy.isResourceFailureRecorded(failure));
+ }
+}
diff --git a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
index 73aef5c4c59..5dd66be212c 100644
--- a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
+++ b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
@@ -201,6 +201,7 @@ struct TPipeHeartbeatResp {
2: optional list<bool> pipeCompletedList
3: optional list<i64> pipeRemainingEventCountList
4: optional list<double> pipeRemainingTimeList
+ 6: optional list<map<string, i64>> pipeRecentFailureList
}
struct TLicense {
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index d6e0b186aad..b04274945da 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -729,6 +729,7 @@ struct TShowPipeInfo {
7: required string exceptionMessage
8: optional i64 remainingEventCount
9: optional double EstimatedRemainingTime
+ 11: optional map<string, i64> recentFailures
}
struct TGetAllPipeInfoResp {
@@ -1775,4 +1776,3 @@ service IConfigNodeRPCService {
/** Push heartbeat in shutdown */
common.TSStatus pushHeartbeat(i32 dataNodeId, common.TPipeHeartbeatResp resp)
}
-
diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
index 989d624f337..0d1f60d61a8 100644
--- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
+++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
@@ -307,6 +307,7 @@ struct TDataNodeHeartbeatResp {
14: optional list<bool> pipeCompletedList
15: optional list<i64> pipeRemainingEventCountList
16: optional list<double> pipeRemainingTimeList
+ 19: optional list<map<string, i64>> pipeRecentFailureList
}
struct TPipeHeartbeatReq {
@@ -1076,4 +1077,4 @@ service MPPDataExchangeService {
/** Empty rpc, only for connection test */
common.TSStatus testConnectionEmptyRPC()
-}
\ No newline at end of file
+}
