This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new d1f362d3ca3 Pipe: Handle transient resource failures locally (#18381)
d1f362d3ca3 is described below
commit d1f362d3ca311b6121c782a423369e7b0998be24
Author: Caideyipi <[email protected]>
AuthorDate: Mon Aug 10 09:28:40 2026 +0800
Pipe: Handle transient resource failures locally (#18381)
---
.../relational/it/schema/IoTDBDatabaseIT.java | 5 +-
.../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 | 29 +++++++
.../runtime/heartbeat/PipeHeartbeatParser.java | 1 +
.../runtime/heartbeat/PipeHeartbeatScheduler.java | 6 +-
.../consensus/response/pipe/PipeTableRespTest.java | 19 +++++
.../runtime/heartbeat/PipeHeartbeatParserTest.java | 52 +++++++++++-
.../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 5 ++
.../subtask/processor/PipeProcessorSubtask.java | 12 +++
.../agent/task/subtask/sink/PipeSinkSubtask.java | 8 ++
.../iotconsensusv2/IoTConsensusV2SyncSink.java | 21 +++--
.../thrift/async/IoTDBDataRegionAsyncSink.java | 57 ++++++++++++-
.../InformationSchemaContentSupplierFactory.java | 6 ++
.../execution/config/sys/pipe/ShowPipeTask.java | 9 +++
.../task/subtask/sink/PipeSinkSubtaskTest.java | 49 ++++++++++++
.../config/sys/pipe/ShowPipeTaskTest.java | 13 ++-
.../pipe/PipeRuntimeSinkResourceException.java | 51 ++++++++++++
...RuntimeSinkRetryTimesConfigurableException.java | 6 ++
.../commons/pipe/agent/task/PipeTaskAgent.java | 10 +++
.../agent/task/meta/PipeTemporaryMetaInAgent.java | 19 ++++-
.../task/meta/PipeTemporaryMetaInCoordinator.java | 81 ++++++++++++++++++-
.../task/subtask/PipeAbstractSinkSubtask.java | 23 ++++--
.../pipe/receiver/PipeReceiverStatusHandler.java | 19 +++--
.../pipe/resource/PipeRecentFailureCounter.java | 93 ++++++++++++++++++++++
.../pipe/resource/PipeResourceFailureType.java | 36 +++++++++
.../commons/pipe/resource/PipeStopStrategy.java | 76 ++++++++++++++++++
.../schema/column/ColumnHeaderConstant.java | 5 +-
.../commons/schema/table/InformationSchema.java | 3 +
.../agent/task/meta/PipeTemporaryMetaTest.java | 32 ++++++++
.../resource/PipeRecentFailureCounterTest.java | 48 +++++++++++
.../pipe/resource/PipeStopStrategyTest.java | 84 +++++++++++++++++++
.../thrift-commons/src/main/thrift/common.thrift | 3 +-
.../src/main/thrift/confignode.thrift | 1 +
.../src/main/thrift/datanode.thrift | 1 +
39 files changed, 874 insertions(+), 36 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java
index e011915de0e..a6124c547a6 100644
---
a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java
@@ -555,7 +555,8 @@ public class IoTDBDatabaseIT {
"exception_message,STRING,ATTRIBUTE,",
"remaining_event_count,INT64,ATTRIBUTE,",
"estimated_remaining_seconds,DOUBLE,ATTRIBUTE,",
- "is_degraded,BOOLEAN,ATTRIBUTE,")));
+ "is_degraded,BOOLEAN,ATTRIBUTE,",
+ "recent_failures,STRING,ATTRIBUTE,")));
TestUtils.assertResultSetEqual(
statement.executeQuery("desc pipe_plugins"),
"ColumnName,DataType,Category,",
@@ -677,7 +678,7 @@ public class IoTDBDatabaseIT {
// Filter out not self-created pipes
TestUtils.assertResultSetEqual(
statement.executeQuery("select * from pipes"),
-
"id,creation_time,state,pipe_source,pipe_processor,pipe_sink,exception_message,remaining_event_count,estimated_remaining_seconds,is_degraded,",
+
"id,creation_time,state,pipe_source,pipe_processor,pipe_sink,exception_message,remaining_event_count,estimated_remaining_seconds,is_degraded,recent_failures,",
Collections.emptySet());
// No auth needed
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 4d9df123520..9c7810dabe2 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
@@ -191,7 +191,8 @@ public class DataNodeHeartbeatHandler implements
AsyncMethodCallback<TDataNodeHe
heartbeatResp.getPipeCompletedList(),
heartbeatResp.getPipeRemainingEventCountList(),
heartbeatResp.getPipeRemainingTimeList(),
- heartbeatResp.getPipeDegradedStatusList());
+ heartbeatResp.getPipeDegradedStatusList(),
+ heartbeatResp.getPipeRecentFailureList());
}
}
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 15167a57ab3..153fae0f9b7 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
@@ -293,6 +293,7 @@ public class PipeTableResp implements DataSet {
if (Objects.nonNull(isDegraded)) {
showPipeInfo.setIsDegraded(isDegraded);
}
+ 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 d646b359e1f..c0276fd2cd5 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
@@ -3338,7 +3338,8 @@ public class ConfigManager implements IManager {
resp.getPipeCompletedList(),
resp.getPipeRemainingEventCountList(),
resp.getPipeRemainingTimeList(),
- resp.getPipeDegradedStatusList());
+ resp.getPipeDegradedStatusList(),
+ 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 854f39c22bd..dbc67befc6b 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.commons.pipe.resource.log.PipeLogger;
import org.apache.iotdb.confignode.i18n.ManagerMessages;
import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent;
@@ -243,6 +244,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 8668debe222..323fdb8ca00 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
@@ -29,6 +29,7 @@ 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.PipeTemporaryMeta;
+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.i18n.ManagerMessages;
@@ -225,6 +226,7 @@ public class PipeConfigNodeTaskAgent extends PipeTaskAgent {
final List<Long> pipeRemainingEventCountList = new ArrayList<>();
final List<Double> pipeRemainingTimeList = new ArrayList<>();
final List<Integer> pipeDegradedStatusList = new ArrayList<>();
+ final List<Map<String, Long>> pipeRecentFailureList = new ArrayList<>();
try {
for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) {
pipeMetaBinaryList.add(pipeMeta.serialize());
@@ -240,6 +242,8 @@ public class PipeConfigNodeTaskAgent extends PipeTaskAgent {
pipeRemainingEventCountList.add(remainingEventCount);
pipeRemainingTimeList.add(estimatedRemainingTime);
pipeDegradedStatusList.add(PipeTemporaryMeta.TS_FILE_EPOCH_DEGRADED_STATUS_UNKNOWN);
+ pipeRecentFailureList.add(
+ ((PipeTemporaryMetaInAgent)
pipeMeta.getTemporaryMeta()).getRecentFailures());
logger.ifPresent(
l ->
@@ -258,6 +262,7 @@ public class PipeConfigNodeTaskAgent extends PipeTaskAgent {
resp.setPipeRemainingEventCountList(pipeRemainingEventCountList);
resp.setPipeRemainingTimeList(pipeRemainingTimeList);
resp.setPipeDegradedStatusList(pipeDegradedStatusList);
+ 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 d9a578c379e..ec00adcd302 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
@@ -29,6 +29,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;
@@ -96,7 +97,8 @@ public class PipeRuntimeCoordinator implements
IClusterStatusSubscriber {
/* @Nullable */ final List<Boolean> pipeCompletedListFromAgent,
/* @Nullable */ final List<Long> pipeRemainingEventCountListFromAgent,
/* @Nullable */ final List<Double> pipeRemainingTimeListFromAgent,
- /* @Nullable */ final List<Integer> pipeDegradedStatusListFromAgent) {
+ /* @Nullable */ final List<Integer> pipeDegradedStatusListFromAgent,
+ /* @Nullable */ final List<Map<String, Long>>
pipeRecentFailureListFromAgent) {
pipeHeartbeatScheduler.parseHeartbeat(
dataNodeId,
new PipeHeartbeat(
@@ -104,6 +106,7 @@ public class PipeRuntimeCoordinator implements
IClusterStatusSubscriber {
pipeCompletedListFromAgent,
pipeRemainingEventCountListFromAgent,
pipeRemainingTimeListFromAgent,
- pipeDegradedStatusListFromAgent));
+ pipeDegradedStatusListFromAgent,
+ 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 31e5020bf89..7aa75b2d78d 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
@@ -24,6 +24,7 @@ import
org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMeta;
import java.nio.ByteBuffer;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -35,6 +36,7 @@ public class PipeHeartbeat {
private final Map<PipeStaticMeta, Long> remainingEventCountMap = new
HashMap<>();
private final Map<PipeStaticMeta, Double> remainingTimeMap = new HashMap<>();
private final Map<PipeStaticMeta, Boolean> isDegradedMap = new HashMap<>();
+ private final Map<PipeStaticMeta, Map<String, Long>> recentFailuresMap = new
HashMap<>();
public PipeHeartbeat(
final List<ByteBuffer> pipeMetaByteBufferListFromAgent,
@@ -42,6 +44,22 @@ public class PipeHeartbeat {
/* @Nullable */ final List<Long> pipeRemainingEventCountListFromAgent,
/* @Nullable */ final List<Double> pipeRemainingTimeListFromAgent,
/* @Nullable */ final List<Integer> pipeDegradedStatusListFromAgent) {
+ this(
+ pipeMetaByteBufferListFromAgent,
+ pipeCompletedListFromAgent,
+ pipeRemainingEventCountListFromAgent,
+ pipeRemainingTimeListFromAgent,
+ pipeDegradedStatusListFromAgent,
+ 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<Integer> pipeDegradedStatusListFromAgent,
+ /* @Nullable */ final List<Map<String, Long>>
pipeRecentFailureListFromAgent) {
// Shall not reach here, just in case
if (Objects.isNull(pipeMetaByteBufferListFromAgent)) {
return;
@@ -77,6 +95,13 @@ public class PipeHeartbeat {
&& i < pipeDegradedStatusListFromAgent.size()
? pipeDegradedStatusListFromAgent.get(i)
: null));
+ recentFailuresMap.put(
+ pipeMeta.getStaticMeta(),
+ Objects.nonNull(pipeRecentFailureListFromAgent)
+ && i < pipeRecentFailureListFromAgent.size()
+ && Objects.nonNull(pipeRecentFailureListFromAgent.get(i))
+ ? new HashMap<>(pipeRecentFailureListFromAgent.get(i))
+ : Collections.emptyMap());
}
}
@@ -104,6 +129,10 @@ public class PipeHeartbeat {
return isDegradedMap.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 c33beed69c0..a8734469d0c 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
@@ -197,6 +197,7 @@ public class PipeHeartbeatParser {
temporaryMeta.setRemainingEvent(nodeId,
pipeHeartbeat.getRemainingEventCount(staticMeta));
temporaryMeta.setRemainingTime(nodeId,
pipeHeartbeat.getRemainingTime(staticMeta));
temporaryMeta.setDegraded(nodeId, pipeHeartbeat.getDegraded(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 00368c0472a..209b08cff15 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
@@ -116,7 +116,8 @@ public class PipeHeartbeatScheduler {
resp.getPipeCompletedList(),
resp.getPipeRemainingEventCountList(),
resp.getPipeRemainingTimeList(),
- resp.getPipeDegradedStatusList())));
+ resp.getPipeDegradedStatusList(),
+ resp.getPipeRecentFailureList())));
// config node heartbeat
try {
@@ -129,7 +130,8 @@ public class PipeHeartbeatScheduler {
null,
configNodeResp.getPipeRemainingEventCountList(),
configNodeResp.getPipeRemainingTimeList(),
- configNodeResp.getPipeDegradedStatusList()));
+ configNodeResp.getPipeDegradedStatusList(),
+ configNodeResp.getPipeRecentFailureList()));
} catch (final Exception e) {
PipeLogger.log(
LOGGER::warn, e,
ManagerMessages.FAILED_TO_COLLECT_PIPE_META_LIST_FROM_CONFIG_NODE_TASK);
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 f4437fd2c34..b3ae595a5ca 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
@@ -169,6 +169,25 @@ public class PipeTableRespTest {
Assert.assertFalse(showPipeResult.get(2).isSetIsDegraded());
}
+ @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 04a898629fd..c7d4e3b5d8d 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
@@ -317,6 +317,49 @@ public class PipeHeartbeatParserTest {
verify(context.procedureManager,
never()).pipeHandleMetaChange(anyBoolean(), anyBoolean());
}
+ @Test
+ public void testParseHeartbeatAggregatesRecentFailuresFromAllDataNodes()
throws Exception {
+
CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false);
+
+ final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo();
+ final PipeMeta pipeMeta = createPipeMeta();
+ pipeTaskInfo.createPipe(
+ new CreatePipePlanV2(pipeMeta.getStaticMeta(),
pipeMeta.getRuntimeMeta()));
+
+ 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, false,
firstNodeFailures));
+ context.parser.parseHeartbeat(2, createPipeHeartbeat(pipeMeta, false,
secondNodeFailures));
+
+ Assert.assertEquals(
+ Long.valueOf(12),
+
getTemporaryMeta(pipeTaskInfo).getGlobalRecentFailures().get("network_timeout"));
+ Assert.assertEquals(
+ Long.valueOf(15),
+
getTemporaryMeta(pipeTaskInfo).getGlobalRecentFailures().get("memory_timeout"));
+ verify(context.procedureManager,
never()).pipeHandleMetaChange(anyBoolean(), anyBoolean());
+ }
+
+ @Test
+ public void testPipeHeartbeatTreatsNullRecentFailureMapAsEmpty() throws
Exception {
+ final PipeMeta pipeMeta = createPipeMeta();
+ final PipeHeartbeat heartbeat =
+ new PipeHeartbeat(
+ Collections.singletonList(pipeMeta.serialize()),
+ Collections.singletonList(false),
+ Collections.singletonList(0L),
+ Collections.singletonList(0d),
+
Collections.singletonList(PipeTemporaryMeta.TS_FILE_EPOCH_DEGRADED_STATUS_UNKNOWN),
+ Collections.singletonList(null));
+
+
Assert.assertTrue(heartbeat.getRecentFailures(pipeMeta.getStaticMeta()).isEmpty());
+ }
+
private ParserTestContext createParserTestContext(final int
registeredDataNodeCount) {
return createParserTestContext(registeredDataNodeCount, new
PipeTaskInfo());
}
@@ -385,12 +428,19 @@ public class PipeHeartbeatParserTest {
private PipeHeartbeat createPipeHeartbeat(final PipeMeta pipeMeta, final
boolean isDegraded)
throws Exception {
+ return createPipeHeartbeat(pipeMeta, isDegraded, Collections.emptyMap());
+ }
+
+ private PipeHeartbeat createPipeHeartbeat(
+ final PipeMeta pipeMeta, final boolean isDegraded, 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(PipeTemporaryMeta.encodeTsFileEpochDegradedStatus(isDegraded)));
+
Collections.singletonList(PipeTemporaryMeta.encodeTsFileEpochDegradedStatus(isDegraded)),
+ Collections.singletonList(recentFailures));
}
private PipeTemporaryMetaInCoordinator getTemporaryMeta(final PipeTaskInfo
pipeTaskInfo) {
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 3adad062db7..1fa7bfb5786 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
@@ -473,6 +473,8 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
PipeTemporaryMeta.encodeTsFileEpochDegradedStatus(
((PipeTemporaryMetaInAgent) pipeMeta.getTemporaryMeta())
.getGlobalTsFileEpochDegraded()));
+ report.pipeRecentFailureList.add(
+ ((PipeTemporaryMetaInAgent)
pipeMeta.getTemporaryMeta()).getRecentFailures());
logger.ifPresent(
l ->
@@ -522,6 +524,7 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
private final List<Long> pipeRemainingEventCountList = new ArrayList<>();
private final List<Double> pipeRemainingTimeList = new ArrayList<>();
private final List<Integer> pipeDegradedStatusList = new ArrayList<>();
+ private final List<Map<String, Long>> pipeRecentFailureList = new
ArrayList<>();
private void setTo(final TDataNodeHeartbeatResp resp) {
resp.setPipeMetaList(pipeMetaBinaryList);
@@ -529,6 +532,7 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
resp.setPipeRemainingEventCountList(pipeRemainingEventCountList);
resp.setPipeRemainingTimeList(pipeRemainingTimeList);
resp.setPipeDegradedStatusList(pipeDegradedStatusList);
+ resp.setPipeRecentFailureList(pipeRecentFailureList);
}
private void setTo(final TPipeHeartbeatResp resp) {
@@ -537,6 +541,7 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
resp.setPipeRemainingEventCountList(pipeRemainingEventCountList);
resp.setPipeRemainingTimeList(pipeRemainingTimeList);
resp.setPipeDegradedStatusList(pipeDegradedStatusList);
+ resp.setPipeRecentFailureList(pipeRecentFailureList);
}
}
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 a95c99ebd52..8c49e03cb06 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
@@ -28,6 +28,7 @@ import
org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
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.i18n.DataNodePipeMessages;
@@ -242,6 +243,7 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
}
decreaseReferenceCountAndReleaseLastEvent(event, shouldReport);
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
+ recordResourceFailure(event, PipeResourceFailureType.MEMORY_TIMEOUT);
PipeLogger.log(
LOGGER::info,
DataNodePipeMessages.TEMPORARILY_OUT_OF_MEMORY_IN_PIPE_EVENT_PROCESSING,
@@ -249,6 +251,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,
DataNodePipeMessages.TEMPORARILY_OUT_OF_MEMORY_IN_PIPE_EVENT_PROCESSING,
@@ -352,4 +355,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 365f211fb51..83de468cba5 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.pipe.sink.protocol.PipeSinkWithSchedulingDelay;
@@ -584,6 +585,13 @@ public class PipeSinkSubtask extends
PipeAbstractSinkSubtask {
PipeDataNodeAgent.runtime().report(event, exception);
}
+ @Override
+ protected void reportResourceFailure(
+ final EnrichedEvent event, final PipeResourceFailureType failureType) {
+ PipeDataNodeAgent.task()
+ .recordPipeResourceFailure(event.getPipeName(),
event.getCreationTime(), failureType);
+ }
+
@Override
public String getDisplayTaskID() {
return generateDisplayTaskID(attributeDisplayString, creationTime,
sinkIndex);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
index 03e589e2f26..2f274189809 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
@@ -156,7 +156,8 @@ public class IoTConsensusV2SyncSink extends IoTDBSink {
.PIPE_EXCEPTION_FAILED_TO_TRANSFER_TABLET_INSERTION_EVENT_S_BECAUSE_S_9710318F,
tabletInsertionEvent,
e.getMessage()),
- Integer.MAX_VALUE);
+ Integer.MAX_VALUE,
+ e);
}
}
@@ -181,7 +182,8 @@ public class IoTConsensusV2SyncSink extends IoTDBSink {
.PIPE_EXCEPTION_FAILED_TO_TRANSFER_TSFILE_INSERTION_EVENT_S_BECAUSE_S_21AD3263,
tsFileInsertionEvent,
e.getMessage()),
- Integer.MAX_VALUE);
+ Integer.MAX_VALUE,
+ e);
}
}
@@ -230,7 +232,8 @@ public class IoTConsensusV2SyncSink extends IoTDBSink {
getFollowerUrl().getPort(),
TABLET_BATCH_SCENARIO,
e.getMessage()),
- Integer.MAX_VALUE);
+ Integer.MAX_VALUE,
+ e);
}
}
@@ -278,7 +281,8 @@ public class IoTConsensusV2SyncSink extends IoTDBSink {
getFollowerUrl().getPort(),
DELETION_SCENARIO,
e.getMessage()),
- Integer.MAX_VALUE);
+ Integer.MAX_VALUE,
+ e);
}
final TSStatus status = resp.getStatus();
@@ -346,7 +350,8 @@ public class IoTConsensusV2SyncSink extends IoTDBSink {
getFollowerUrl().getPort(),
TABLET_INSERTION_NODE_SCENARIO,
e.getMessage()),
- Integer.MAX_VALUE);
+ Integer.MAX_VALUE,
+ e);
}
final TSStatus status = resp.getStatus();
@@ -420,7 +425,8 @@ public class IoTConsensusV2SyncSink extends IoTDBSink {
getFollowerUrl().getPort(),
TSFILE_SCENARIO,
e.getMessage()),
- Integer.MAX_VALUE);
+ Integer.MAX_VALUE,
+ e);
}
final TSStatus status = resp.getStatus();
@@ -487,7 +493,8 @@ public class IoTConsensusV2SyncSink extends IoTDBSink {
.PIPE_EXCEPTION_NETWORK_ERROR_WHEN_TRANSFER_FILE_S_BECAUSE_S_3C673B7A,
file,
e.getMessage()),
- Integer.MAX_VALUE);
+ Integer.MAX_VALUE,
+ e);
}
position += readLength;
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 689a5bb1b7b..422dd70976b 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
@@ -25,13 +25,17 @@ import org.apache.iotdb.commons.audit.UserEntity;
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.commons.pipe.sink.protocol.PipeSinkWithSchedulingDelay;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
+import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
@@ -81,6 +85,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;
@@ -129,6 +135,9 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
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;
@@ -670,6 +679,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
final Event polledEvent;
if (!retryEventQueue.isEmpty()) {
peekedEvent = retryEventQueue.peek();
+ retryEvent2ResourceFailureType.remove(peekedEvent);
if (peekedEvent instanceof PipeInsertNodeTabletInsertionEvent) {
retryTransfer((PipeInsertNodeTabletInsertionEvent) peekedEvent);
@@ -689,6 +699,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
return;
}
peekedEvent = retryTsFileQueue.peek();
+ retryEvent2ResourceFailureType.remove(peekedEvent);
retryTransfer((PipeTsFileInsertionEvent) peekedEvent);
polledEvent = retryTsFileQueue.poll();
}
@@ -726,6 +737,12 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
+ ", 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);
@@ -788,6 +805,13 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
*/
@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) {
@@ -808,6 +832,23 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
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);
@@ -834,7 +875,17 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
*/
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() {
@@ -986,6 +1037,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
&& isDroppedPipe((EnrichedEvent) event, committerKey)) {
((EnrichedEvent)
event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
retryEventQueueEventCounter.decreaseEventCount(event);
+ retryEvent2ResourceFailureType.remove(event);
return true;
}
return false;
@@ -997,6 +1049,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
&& isDroppedPipe((EnrichedEvent) event, committerKey)) {
((EnrichedEvent)
event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName());
retryEventQueueEventCounter.decreaseEventCount(event);
+ retryEvent2ResourceFailureType.remove(event);
return true;
}
return false;
@@ -1049,10 +1102,12 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
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/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java
index 7ca14b38284..4f01b5230ad 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java
@@ -709,6 +709,12 @@ public class InformationSchemaContentSupplierFactory {
} else {
columnBuilders[9].appendNull();
}
+ columnBuilders[10].writeBinary(
+ new Binary(
+ tPipeInfo.isSetRecentFailures()
+ ? new TreeMap<>(tPipeInfo.getRecentFailures()).toString()
+ : "{}",
+ TSFileConfig.STRING_CHARSET));
resultBuilder.declarePosition();
}
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 6698ae88fdc..71a6e6ef4d3 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
@@ -42,6 +42,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 {
@@ -135,6 +136,14 @@ public class ShowPipeTask implements IConfigTask {
} else {
builder.getColumnBuilder(9).appendNull();
}
+ builder
+ .getColumnBuilder(10)
+ .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 673fa3e91e4..3d5427d39b3 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;
@@ -371,9 +373,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) {
@@ -397,10 +432,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
index 3f89e8ee9cc..0822b51f1cb 100644
---
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
@@ -29,6 +29,8 @@ 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;
import static org.junit.Assert.assertFalse;
@@ -37,9 +39,13 @@ import static org.junit.Assert.assertTrue;
public class ShowPipeTaskTest {
@Test
- public void testBuildTSBlockWritesDegradedColumn() throws Exception {
+ public void testBuildTSBlockWritesRuntimeColumns() throws Exception {
final TShowPipeInfo degradedPipe = createPipeInfo("degraded_pipe");
degradedPipe.setIsDegraded(true);
+ final Map<String, Long> recentFailures = new HashMap<>();
+ recentFailures.put("network_timeout", 10L);
+ recentFailures.put("memory_timeout", 15L);
+ degradedPipe.setRecentFailures(recentFailures);
final TShowPipeInfo normalPipe = createPipeInfo("normal_pipe");
normalPipe.setIsDegraded(false);
final TShowPipeInfo unknownPipe = createPipeInfo("unknown_pipe");
@@ -53,10 +59,15 @@ public class ShowPipeTaskTest {
assertEquals(TSStatusCode.SUCCESS_STATUS, result.getStatusCode());
assertEquals(
ColumnHeaderConstant.IS_DEGRADED,
result.getResultSetHeader().getRespColumns().get(9));
+ assertEquals(
+ ColumnHeaderConstant.RECENT_FAILURES,
result.getResultSetHeader().getRespColumns().get(10));
assertEquals(3, resultSet.getPositionCount());
assertTrue(resultSet.getColumn(9).getBoolean(0));
assertFalse(resultSet.getColumn(9).getBoolean(1));
assertTrue(resultSet.getColumn(9).isNull(2));
+ assertEquals(
+ "{memory_timeout=15, network_timeout=10}",
resultSet.getColumn(10).getBinary(0).toString());
+ assertEquals("{}", resultSet.getColumn(10).getBinary(1).toString());
}
private TShowPipeInfo createPipeInfo(final String pipeName) {
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/exception/pipe/PipeRuntimeSinkRetryTimesConfigurableException.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeSinkRetryTimesConfigurableException.java
index aa64e533528..6db893421aa 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeSinkRetryTimesConfigurableException.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeSinkRetryTimesConfigurableException.java
@@ -32,6 +32,12 @@ public class PipeRuntimeSinkRetryTimesConfigurableException
this.retryTimes = retryTimes;
}
+ public PipeRuntimeSinkRetryTimesConfigurableException(
+ final String message, final int retryTimes, final Throwable cause) {
+ super(message, cause);
+ this.retryTimes = retryTimes;
+ }
+
public int getRetryTimes() {
return retryTimes;
}
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 f2f912cbc61..5523717b093 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
@@ -36,6 +36,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;
@@ -1217,6 +1218,15 @@ public abstract class PipeTaskAgent {
}
}
+ public void recordPipeResourceFailure(
+ final String pipeName, final long creationTime, final
PipeResourceFailureType failureType) {
+ final PipeMeta pipeMeta = pipeMetaKeeper.getPipeMeta(pipeName,
creationTime);
+ // To avoid recording a failure for the stale pipe before alter
+ if (Objects.nonNull(pipeMeta) &&
pipeMeta.getStaticMeta().getCreationTime() == creationTime) {
+ ((PipeTemporaryMetaInAgent)
pipeMeta.getTemporaryMeta()).recordResourceFailure(failureType);
+ }
+ }
+
public void setPipeTsFileEpochDegraded(
final String pipeName,
final long creationTime,
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 c28bb51236d..f525c015706 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;
@@ -31,6 +33,7 @@ public class PipeTemporaryMetaInAgent implements
PipeTemporaryMeta {
// Statistics
private final AtomicLong floatingMemoryUsageInByte = new AtomicLong(0L);
+ private final PipeRecentFailureCounter recentFailureCounter = new
PipeRecentFailureCounter();
private final ConcurrentMap<Integer, Boolean>
regionId2TsFileEpochDegradedMap =
new ConcurrentHashMap<>();
@@ -71,6 +74,14 @@ public class PipeTemporaryMetaInAgent implements
PipeTemporaryMeta {
return regionId2TsFileEpochDegradedMap.isEmpty() ? null : false;
}
+ public void recordResourceFailure(final PipeResourceFailureType failureType)
{
+ recentFailureCounter.record(failureType);
+ }
+
+ public Map<String, Long> getRecentFailures() {
+ return recentFailureCounter.getRecentFailures();
+ }
+
public String getPipeNameWithCreationTime() {
return pipeNameWithCreationTime;
}
@@ -107,13 +118,17 @@ public class PipeTemporaryMetaInAgent implements
PipeTemporaryMeta {
this.floatingMemoryUsageInByte.get(),
that.floatingMemoryUsageInByte.get())
&& Objects.equals(
this.regionId2TsFileEpochDegradedMap,
that.regionId2TsFileEpochDegradedMap)
+ && Objects.equals(this.getRecentFailures(), that.getRecentFailures())
&& Objects.equals(this.regionId2CommitterKeyMap,
that.regionId2CommitterKeyMap);
}
@Override
public int hashCode() {
return Objects.hash(
- floatingMemoryUsageInByte.get(), regionId2TsFileEpochDegradedMap,
regionId2CommitterKeyMap);
+ floatingMemoryUsageInByte.get(),
+ regionId2TsFileEpochDegradedMap,
+ getRecentFailures(),
+ regionId2CommitterKeyMap);
}
@Override
@@ -123,6 +138,8 @@ public class PipeTemporaryMetaInAgent implements
PipeTemporaryMeta {
+ floatingMemoryUsageInByte
+ ", regionId2TsFileEpochDegradedMap="
+ regionId2TsFileEpochDegradedMap
+ + ", 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 b44649ff30c..5c3a1ea17eb 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;
@@ -33,6 +38,8 @@ public class PipeTemporaryMetaInCoordinator implements
PipeTemporaryMeta {
private final ConcurrentMap<Integer, Long> nodeId2RemainingEventMap = new
ConcurrentHashMap<>();
private final ConcurrentMap<Integer, Double> nodeId2RemainingTimeMap = new
ConcurrentHashMap<>();
private final ConcurrentMap<Integer, Boolean> nodeId2IsDegradedMap = new
ConcurrentHashMap<>();
+ private final ConcurrentMap<Integer, RecentFailureSnapshot>
nodeId2RecentFailuresMap =
+ new ConcurrentHashMap<>();
public void markDataNodeCompleted(final int dataNodeId) {
completedDataNodeIds.add(dataNodeId);
@@ -58,6 +65,27 @@ public class PipeTemporaryMetaInCoordinator implements
PipeTemporaryMeta {
}
}
+ 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;
}
@@ -77,6 +105,23 @@ public class PipeTemporaryMetaInCoordinator implements
PipeTemporaryMeta {
return nodeId2IsDegradedMap.isEmpty() ? null : false;
}
+ 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) {
@@ -89,7 +134,8 @@ public class PipeTemporaryMetaInCoordinator implements
PipeTemporaryMeta {
return Objects.equals(this.completedDataNodeIds, that.completedDataNodeIds)
&& Objects.equals(this.nodeId2RemainingEventMap,
that.nodeId2RemainingEventMap)
&& Objects.equals(this.nodeId2RemainingTimeMap,
that.nodeId2RemainingTimeMap)
- && Objects.equals(this.nodeId2IsDegradedMap,
that.nodeId2IsDegradedMap);
+ && Objects.equals(this.nodeId2IsDegradedMap, that.nodeId2IsDegradedMap)
+ && Objects.equals(this.nodeId2RecentFailuresMap,
that.nodeId2RecentFailuresMap);
}
@Override
@@ -98,7 +144,8 @@ public class PipeTemporaryMetaInCoordinator implements
PipeTemporaryMeta {
completedDataNodeIds,
nodeId2RemainingEventMap,
nodeId2RemainingTimeMap,
- nodeId2IsDegradedMap);
+ nodeId2IsDegradedMap,
+ nodeId2RecentFailuresMap);
}
@Override
@@ -112,6 +159,36 @@ public class PipeTemporaryMetaInCoordinator implements
PipeTemporaryMeta {
+ nodeId2RemainingTimeMap
+ ", nodeId2IsDegradedMap="
+ nodeId2IsDegradedMap
+ + ", 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 99ec79badab..7a07345c4ae 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,13 +19,14 @@
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.i18n.PipeMessages;
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;
@@ -37,7 +38,6 @@ import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.ListeningScheduledExecutorService;
-import org.apache.tsfile.external.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -353,9 +353,17 @@ 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, PipeMessages.TEMPORARILY_OUT_OF_MEMORY);
+ 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,
PipeMessages.TEMPORARILY_OUT_OF_MEMORY);
+ } else {
+ sleep4NonReportException();
+ }
} else if (e instanceof PipeRuntimeSinkNonReportTimeConfigurableException)
{
if (lastExceptionTime == Long.MAX_VALUE) {
lastExceptionTime = System.currentTimeMillis();
@@ -401,4 +409,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 d5024ddd50f..2cb31c81fd0 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
@@ -22,8 +22,11 @@ package org.apache.iotdb.commons.pipe.receiver;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import
org.apache.iotdb.commons.exception.pipe.IoTConsensusV2RetryWithIncreasingIntervalException;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkNonReportTimeConfigurableException;
+import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkResourceException;
import org.apache.iotdb.commons.i18n.PipeMessages;
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.RetryUtils;
import org.apache.iotdb.commons.utils.TestOnly;
@@ -120,6 +123,14 @@ public class PipeReceiverStatusHandler {
return;
}
+ if (!PipeStopStrategy.accept(null, status)) {
+ PipeLogger.log(
+ LOGGER::info, PipeMessages.TEMPORARY_UNAVAILABLE_RETRY, status,
exceptionMessage);
+ final PipeResourceFailureType failureType =
+ PipeStopStrategy.getResourceFailureType(null, status);
+ throw new PipeRuntimeSinkResourceException(exceptionMessage,
failureType);
+ }
+
switch (status.getCode()) {
case 200: // SUCCESS_STATUS
case 400: // REDIRECTION_RECOMMEND
@@ -133,14 +144,6 @@ public class PipeReceiverStatusHandler {
return;
}
- case 1808: // PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION
- {
- PipeLogger.log(
- LOGGER::info, PipeMessages.TEMPORARY_UNAVAILABLE_RETRY, status,
exceptionMessage);
- 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..ee00946316d
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/PipeRecentFailureCounter.java
@@ -0,0 +1,93 @@
+/*
+ * 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 = 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/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java
index 87c6a01097b..23ccb5c91db 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java
@@ -197,6 +197,7 @@ public class ColumnHeaderConstant {
public static final String REMAINING_EVENT_COUNT = "RemainingEventCount";
public static final String ESTIMATED_REMAINING_SECONDS =
"EstimatedRemainingSeconds";
public static final String IS_DEGRADED = "IsDegraded";
+ public static final String RECENT_FAILURES = "RecentFailures";
// column names for show repair data partition table progress
public static final String REPAIR_DATA_PARTITION_TABLE_STATUS = "Status";
@@ -279,6 +280,7 @@ public class ColumnHeaderConstant {
public static final String ESTIMATED_REMAINING_SECONDS_TABLE_MODEL =
"estimated_remaining_seconds";
public static final String IS_DEGRADED_TABLE_MODEL = "is_degraded";
+ public static final String RECENT_FAILURES_TABLE_MODEL = "recent_failures";
public static final String PLUGIN_NAME_TABLE_MODEL = "plugin_name";
public static final String PLUGIN_TYPE_TABLE_MODEL = "plugin_type";
@@ -616,7 +618,8 @@ public class ColumnHeaderConstant {
new ColumnHeader(EXCEPTION_MESSAGE, TSDataType.TEXT),
new ColumnHeader(REMAINING_EVENT_COUNT, TSDataType.TEXT),
new ColumnHeader(ESTIMATED_REMAINING_SECONDS, TSDataType.TEXT),
- new ColumnHeader(IS_DEGRADED, TSDataType.BOOLEAN));
+ new ColumnHeader(IS_DEGRADED, TSDataType.BOOLEAN),
+ new ColumnHeader(RECENT_FAILURES, TSDataType.TEXT));
public static final List<ColumnHeader>
showRepairDataPartitionTableProgressColumnHeaders =
ImmutableList.of(
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java
index 32f51a173be..645ce6d2b85 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java
@@ -228,6 +228,9 @@ public class InformationSchema {
pipeTable.addColumnSchema(
new AttributeColumnSchema(
ColumnHeaderConstant.IS_DEGRADED_TABLE_MODEL, TSDataType.BOOLEAN));
+ pipeTable.addColumnSchema(
+ new AttributeColumnSchema(
+ ColumnHeaderConstant.RECENT_FAILURES_TABLE_MODEL,
TSDataType.STRING));
schemaTables.put(PIPES, pipeTable);
final TsTable pipePluginTable = new TsTable(PIPE_PLUGINS);
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
index b6091ab76f7..eddd1d29fad 100644
---
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
@@ -19,9 +19,14 @@
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
@@ -89,4 +94,31 @@ public class PipeTemporaryMetaTest {
temporaryMeta.setDegraded(1, null);
Assert.assertNull(temporaryMeta.getGlobalDegraded());
}
+
+ @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..5798ec192ff
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeStopStrategyTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.commons.exception.pipe.PipeRuntimeSinkRetryTimesConfigurableException;
+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 PipeRuntimeSinkRetryTimesConfigurableException(
+ "transfer failed",
+ Integer.MAX_VALUE,
+ 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 bdde4fbea5d..edc824a6b43 100644
--- a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
+++ b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
@@ -203,6 +203,7 @@ struct TPipeHeartbeatResp {
3: optional list<i64> pipeRemainingEventCountList
4: optional list<double> pipeRemainingTimeList
5: optional list<i32> pipeDegradedStatusList
+ 6: optional list<map<string, i64>> pipeRecentFailureList
}
struct TLicense {
@@ -359,4 +360,4 @@ enum FunctionType{
SCALAR=1,
AGGREGATE=2,
TABLE=3
-}
\ No newline at end of file
+}
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index 7aa75731545..1111740f759 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -883,6 +883,7 @@ struct TShowPipeInfo {
8: optional i64 remainingEventCount
9: optional double EstimatedRemainingTime
10: optional bool isDegraded
+ 11: optional map<string, i64> recentFailures
}
struct TGetAllPipeInfoResp {
diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
index ea0f9467a2a..d66ed10ccf9 100644
--- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
+++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
@@ -318,6 +318,7 @@ struct TDataNodeHeartbeatResp {
16: optional list<double> pipeRemainingTimeList
17: optional map<i32, i64> dataRegionRawDataSize
18: optional list<i32> pipeDegradedStatusList
+ 19: optional list<map<string, i64>> pipeRecentFailureList
}
struct TPipeHeartbeatReq {
