From 4229932f8d4f1c5580df58401f5521c9c2d82602 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 3 Aug 2026 09:50:59 +0800 Subject: [PATCH] Pipe: Handle transient resource failures locally --- .../relational/it/schema/IoTDBDatabaseIT.java | 5 +- .../heartbeat/DataNodeHeartbeatHandler.java | 3 +- .../response/pipe/task/PipeTableResp.java | 1 + .../confignode/manager/ConfigManager.java | 3 +- .../agent/task/PipeConfigNodeSubtask.java | 8 ++ .../agent/task/PipeConfigNodeTaskAgent.java | 5 + .../runtime/PipeRuntimeCoordinator.java | 7 +- .../runtime/heartbeat/PipeHeartbeat.java | 29 ++++++ .../heartbeat/PipeHeartbeatParser.java | 1 + .../heartbeat/PipeHeartbeatScheduler.java | 6 +- .../response/pipe/PipeTableRespTest.java | 19 ++++ .../heartbeat/PipeHeartbeatParserTest.java | 52 ++++++++++- .../agent/task/PipeDataNodeTaskAgent.java | 5 + .../processor/PipeProcessorSubtask.java | 12 +++ .../task/subtask/sink/PipeSinkSubtask.java | 8 ++ .../IoTConsensusV2SyncSink.java | 21 +++-- .../async/IoTDBDataRegionAsyncSink.java | 57 +++++++++++- ...formationSchemaContentSupplierFactory.java | 6 ++ .../config/sys/pipe/ShowPipeTask.java | 9 ++ .../subtask/sink/PipeSinkSubtaskTest.java | 49 ++++++++++ .../config/sys/pipe/ShowPipeTaskTest.java | 13 ++- .../PipeRuntimeSinkResourceException.java | 51 ++++++++++ ...meSinkRetryTimesConfigurableException.java | 6 ++ .../pipe/agent/task/PipeTaskAgent.java | 10 ++ .../task/meta/PipeTemporaryMetaInAgent.java | 19 +++- .../meta/PipeTemporaryMetaInCoordinator.java | 81 +++++++++++++++- .../task/subtask/PipeAbstractSinkSubtask.java | 23 ++++- .../receiver/PipeReceiverStatusHandler.java | 19 ++-- .../resource/PipeRecentFailureCounter.java | 93 +++++++++++++++++++ .../resource/PipeResourceFailureType.java | 36 +++++++ .../pipe/resource/PipeStopStrategy.java | 76 +++++++++++++++ .../schema/column/ColumnHeaderConstant.java | 5 +- .../schema/table/InformationSchema.java | 3 + .../task/meta/PipeTemporaryMetaTest.java | 32 +++++++ .../PipeRecentFailureCounterTest.java | 48 ++++++++++ .../pipe/resource/PipeStopStrategyTest.java | 84 +++++++++++++++++ .../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(-) create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/pipe/PipeRuntimeSinkResourceException.java create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/PipeRecentFailureCounter.java create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/PipeResourceFailureType.java create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/PipeStopStrategy.java create mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeRecentFailureCounterTest.java create mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/PipeStopStrategyTest.java 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 e011915de0e58..a6124c547a6f6 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 void testInformationSchema() throws SQLException { "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 void testInformationSchema() throws SQLException { // 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 4d9df1235209b..9c7810dabe2d5 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 @@ private void cachePipeHeartbeat(TDataNodeHeartbeatResp heartbeatResp) { 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 15167a57ab336..153fae0f9b72c 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 TShowPipeResp convertToTShowPipeResp() { 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 d646b359e1f35..c0276fd2cd540 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 TSStatus pushHeartbeat(final int dataNodeId, final TPipeHeartbeatResp res 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 854f39c22bd1d..dbc67befc6b7a 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.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 @@ protected void report(final EnrichedEvent event, final PipeRuntimeException exce 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 8668debe22223..323fdb8ca0045 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.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 @@ protected void collectPipeMetaListInternal( final List pipeRemainingEventCountList = new ArrayList<>(); final List pipeRemainingTimeList = new ArrayList<>(); final List pipeDegradedStatusList = new ArrayList<>(); + final List> pipeRecentFailureList = new ArrayList<>(); try { for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) { pipeMetaBinaryList.add(pipeMeta.serialize()); @@ -240,6 +242,8 @@ protected void collectPipeMetaListInternal( 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 @@ protected void collectPipeMetaListInternal( 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 d9a578c379ec0..ec00adcd30224 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 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 void parseHeartbeat( /* @Nullable */ final List pipeCompletedListFromAgent, /* @Nullable */ final List pipeRemainingEventCountListFromAgent, /* @Nullable */ final List pipeRemainingTimeListFromAgent, - /* @Nullable */ final List pipeDegradedStatusListFromAgent) { + /* @Nullable */ final List pipeDegradedStatusListFromAgent, + /* @Nullable */ final List> pipeRecentFailureListFromAgent) { pipeHeartbeatScheduler.parseHeartbeat( dataNodeId, new PipeHeartbeat( @@ -104,6 +106,7 @@ public void parseHeartbeat( 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 31e5020bf8913..7aa75b2d78d79 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.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 remainingEventCountMap = new HashMap<>(); private final Map remainingTimeMap = new HashMap<>(); private final Map isDegradedMap = new HashMap<>(); + private final Map> recentFailuresMap = new HashMap<>(); public PipeHeartbeat( final List pipeMetaByteBufferListFromAgent, @@ -42,6 +44,22 @@ public PipeHeartbeat( /* @Nullable */ final List pipeRemainingEventCountListFromAgent, /* @Nullable */ final List pipeRemainingTimeListFromAgent, /* @Nullable */ final List pipeDegradedStatusListFromAgent) { + this( + pipeMetaByteBufferListFromAgent, + pipeCompletedListFromAgent, + pipeRemainingEventCountListFromAgent, + pipeRemainingTimeListFromAgent, + pipeDegradedStatusListFromAgent, + null); + } + + public PipeHeartbeat( + final List pipeMetaByteBufferListFromAgent, + /* @Nullable */ final List pipeCompletedListFromAgent, + /* @Nullable */ final List pipeRemainingEventCountListFromAgent, + /* @Nullable */ final List pipeRemainingTimeListFromAgent, + /* @Nullable */ final List pipeDegradedStatusListFromAgent, + /* @Nullable */ final List> pipeRecentFailureListFromAgent) { // Shall not reach here, just in case if (Objects.isNull(pipeMetaByteBufferListFromAgent)) { return; @@ -77,6 +95,13 @@ public 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 Boolean getDegraded(final PipeStaticMeta pipeStaticMeta) { return isDegradedMap.get(pipeStaticMeta); } + public Map 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 c33beed69c062..a8734469d0c8d 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 @@ private void parseHeartbeatAndSaveMetaChangeLocally( 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 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 00368c0472a4a..209b08cff1557 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 @@ private synchronized void heartbeat() { resp.getPipeCompletedList(), resp.getPipeRemainingEventCountList(), resp.getPipeRemainingTimeList(), - resp.getPipeDegradedStatusList()))); + resp.getPipeDegradedStatusList(), + resp.getPipeRecentFailureList()))); // config node heartbeat try { @@ -129,7 +130,8 @@ private synchronized void heartbeat() { 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 f4437fd2c34e1..b3ae595a5caa6 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 void testConvertToTShowPipeRespIncludesDegradedStatus() { 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 firstNodeFailures = new HashMap<>(); + firstNodeFailures.put("network_timeout", 10L); + firstNodeFailures.put("memory_timeout", 15L); + temporaryMeta.setRecentFailures(1, firstNodeFailures); + final Map 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 04a898629fd81..c7d4e3b5d8d2d 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 void testParseHeartbeatTreatsMissingPipeDegradedStatusAsUnknown() throws 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 firstNodeFailures = new HashMap<>(); + firstNodeFailures.put("network_timeout", 10L); + final Map 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 @@ private void createPipe( 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 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 3adad062db7dd..1fa7bfb578683 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 @@ private PipeMetaReport collectPipeMetaReport( PipeTemporaryMeta.encodeTsFileEpochDegradedStatus( ((PipeTemporaryMetaInAgent) pipeMeta.getTemporaryMeta()) .getGlobalTsFileEpochDegraded())); + report.pipeRecentFailureList.add( + ((PipeTemporaryMetaInAgent) pipeMeta.getTemporaryMeta()).getRecentFailures()); logger.ifPresent( l -> @@ -522,6 +524,7 @@ private static class PipeMetaReport { private final List pipeRemainingEventCountList = new ArrayList<>(); private final List pipeRemainingTimeList = new ArrayList<>(); private final List pipeDegradedStatusList = new ArrayList<>(); + private final List> pipeRecentFailureList = new ArrayList<>(); private void setTo(final TDataNodeHeartbeatResp resp) { resp.setPipeMetaList(pipeMetaBinaryList); @@ -529,6 +532,7 @@ private void setTo(final TDataNodeHeartbeatResp resp) { resp.setPipeRemainingEventCountList(pipeRemainingEventCountList); resp.setPipeRemainingTimeList(pipeRemainingTimeList); resp.setPipeDegradedStatusList(pipeDegradedStatusList); + resp.setPipeRecentFailureList(pipeRecentFailureList); } private void setTo(final TPipeHeartbeatResp resp) { @@ -537,6 +541,7 @@ private void setTo(final TPipeHeartbeatResp resp) { 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 a95c99ebd52e9..8c49e03cb0605 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.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 @@ protected boolean executeOnce() throws Exception { } 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 @@ protected boolean executeOnce() throws Exception { 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 @@ protected String getRootCause(final Throwable throwable) { 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 365f211fb51ff..83de468cba54d 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.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 @@ protected void report(final EnrichedEvent event, final PipeRuntimeException exce 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 03e589e2f2699..2f2741898094d 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 void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exc .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 void transfer(final TsFileInsertionEvent tsFileInsertionEvent) throws Exc .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 @@ private void doTransfer() { getFollowerUrl().getPort(), TABLET_BATCH_SCENARIO, e.getMessage()), - Integer.MAX_VALUE); + Integer.MAX_VALUE, + e); } } @@ -278,7 +281,8 @@ private void doTransfer(final PipeDeleteDataNodeEvent pipeDeleteDataNodeEvent) getFollowerUrl().getPort(), DELETION_SCENARIO, e.getMessage()), - Integer.MAX_VALUE); + Integer.MAX_VALUE, + e); } final TSStatus status = resp.getStatus(); @@ -346,7 +350,8 @@ private void doTransfer(PipeInsertNodeTabletInsertionEvent pipeInsertNodeTabletI getFollowerUrl().getPort(), TABLET_INSERTION_NODE_SCENARIO, e.getMessage()), - Integer.MAX_VALUE); + Integer.MAX_VALUE, + e); } final TSStatus status = resp.getStatus(); @@ -420,7 +425,8 @@ private void doTransfer(final PipeTsFileInsertionEvent pipeTsFileInsertionEvent) getFollowerUrl().getPort(), TSFILE_SCENARIO, e.getMessage()), - Integer.MAX_VALUE); + Integer.MAX_VALUE, + e); } final TSStatus status = resp.getStatus(); @@ -487,7 +493,8 @@ protected void transferFilePieces( .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 480ade1654617..c7be3f7ea8c79 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.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; @@ -82,6 +86,8 @@ 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; @@ -130,6 +136,9 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink implements PipeSinkWithS private final BlockingQueue 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 retryEvent2ResourceFailureType = + new IdentityHashMap<>(); private IoTDBDataNodeAsyncClientManager clientManager; private IoTDBDataNodeAsyncClientManager transferTsFileClientManager; @@ -671,6 +680,7 @@ private void transferQueuedEventsIfNecessary(final boolean forced) { final Event polledEvent; if (!retryEventQueue.isEmpty()) { peekedEvent = retryEventQueue.peek(); + retryEvent2ResourceFailureType.remove(peekedEvent); if (peekedEvent instanceof PipeInsertNodeTabletInsertionEvent) { retryTransfer((PipeInsertNodeTabletInsertionEvent) peekedEvent); @@ -690,6 +700,7 @@ private void transferQueuedEventsIfNecessary(final boolean forced) { return; } peekedEvent = retryTsFileQueue.peek(); + retryEvent2ResourceFailureType.remove(peekedEvent); retryTransfer((PipeTsFileInsertionEvent) peekedEvent); polledEvent = retryTsFileQueue.poll(); } @@ -727,6 +738,12 @@ private void transferQueuedEventsIfNecessary(final boolean forced) { + ", 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); @@ -789,6 +806,13 @@ private void retryTransfer(final PipeTsFileInsertionEvent tsFileInsertionEvent) */ @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> failureRecordedPipes) { + final PipeResourceFailureType resourceFailureType = + PipeStopStrategy.getResourceFailureType(e, null); isConnectionException = e instanceof PipeConnectionException || ThriftClient.isConnectionBroken(e); if (event instanceof EnrichedEvent) { @@ -809,6 +833,23 @@ public void addFailureEventToRetryQueue(final Event event, final Exception e) { return; } + if (resourceFailureType != null && event instanceof EnrichedEvent) { + final EnrichedEvent enrichedEvent = (EnrichedEvent) event; + final Pair 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); @@ -835,7 +876,17 @@ public void addFailureEventToRetryQueue(final Event event, final Exception e) { */ public void addFailureEventsToRetryQueue( final Iterable events, final Exception e) { - events.forEach(event -> addFailureEventToRetryQueue(event, e)); + final Set> 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() { @@ -987,6 +1038,7 @@ public synchronized void discardEventsOfPipe(final CommitterKey committerKey) { && isDroppedPipe((EnrichedEvent) event, committerKey)) { ((EnrichedEvent) event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName()); retryEventQueueEventCounter.decreaseEventCount(event); + retryEvent2ResourceFailureType.remove(event); return true; } return false; @@ -998,6 +1050,7 @@ && isDroppedPipe((EnrichedEvent) event, committerKey)) { && isDroppedPipe((EnrichedEvent) event, committerKey)) { ((EnrichedEvent) event).clearReferenceCount(IoTDBDataRegionAsyncSink.class.getName()); retryEventQueueEventCounter.decreaseEventCount(event); + retryEvent2ResourceFailureType.remove(event); return true; } return false; @@ -1050,10 +1103,12 @@ public synchronized void clearRetryEventsReferenceCount() { 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 7ca14b38284ef..4f01b5230ad9b 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 @@ protected void constructLine() { } 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 6698ae88fdc0c..71a6e6ef4d327 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.Pair; import java.util.List; +import java.util.TreeMap; import java.util.stream.Collectors; public class ShowPipeTask implements IConfigTask { @@ -135,6 +136,14 @@ public static void buildTSBlock( } 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 673fa3e91e4a9..3d5427d39b38e 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.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 void testHeartbeatPreservesReceiverProbeDelayException() throws Exception } } + @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 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 pendingQueue, final PipeConnector connector) { @@ -397,10 +432,24 @@ private PipeRuntimeException getReportedException() { 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 3f89e8ee9cce7..0822b51f1cbeb 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.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 @@ 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 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 void testBuildTSBlockWritesDegradedColumn() throws Exception { 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 0000000000000..efc41d886547f --- /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 aa64e533528fd..6db893421aac6 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 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 f2f912cbc61a4..5523717b0932a 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.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 void decreaseFloatingMemoryUsageInByte( } } + 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 c28bb51236d8c..f525c01570650 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 regionId2TsFileEpochDegradedMap = new ConcurrentHashMap<>(); @@ -71,6 +74,14 @@ public Boolean getGlobalTsFileEpochDegraded() { return regionId2TsFileEpochDegradedMap.isEmpty() ? null : false; } + public void recordResourceFailure(final PipeResourceFailureType failureType) { + recentFailureCounter.record(failureType); + } + + public Map getRecentFailures() { + return recentFailureCounter.getRecentFailures(); + } + public String getPipeNameWithCreationTime() { return pipeNameWithCreationTime; } @@ -107,13 +118,17 @@ public boolean equals(final Object o) { 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 String toString() { + 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 b44649ff30cea..5c3a1ea17eb24 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 nodeId2RemainingEventMap = new ConcurrentHashMap<>(); private final ConcurrentMap nodeId2RemainingTimeMap = new ConcurrentHashMap<>(); private final ConcurrentMap nodeId2IsDegradedMap = new ConcurrentHashMap<>(); + private final ConcurrentMap nodeId2RecentFailuresMap = + new ConcurrentHashMap<>(); public void markDataNodeCompleted(final int dataNodeId) { completedDataNodeIds.add(dataNodeId); @@ -58,6 +65,27 @@ public void setDegraded(final int dataNodeId, final Boolean isDegraded) { } } + public void setRecentFailures(final int dataNodeId, final Map recentFailures) { + if (Objects.isNull(recentFailures) || recentFailures.isEmpty()) { + nodeId2RecentFailuresMap.remove(dataNodeId); + return; + } + + final Map 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 getCompletedDataNodeIds() { return completedDataNodeIds; } @@ -77,6 +105,23 @@ public Boolean getGlobalDegraded() { return nodeId2IsDegradedMap.isEmpty() ? null : false; } + public Map getGlobalRecentFailures() { + final long earliestIncludedTime = + System.currentTimeMillis() - PipeRecentFailureCounter.WINDOW_MILLIS; + nodeId2RecentFailuresMap + .entrySet() + .removeIf(entry -> entry.getValue().reportTime < earliestIncludedTime); + + final Map 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 boolean equals(final Object o) { 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 int hashCode() { completedDataNodeIds, nodeId2RemainingEventMap, nodeId2RemainingTimeMap, - nodeId2IsDegradedMap); + nodeId2IsDegradedMap, + nodeId2RecentFailuresMap); } @Override @@ -112,6 +159,36 @@ public String toString() { + nodeId2RemainingTimeMap + ", nodeId2IsDegradedMap=" + nodeId2IsDegradedMap + + ", nodeId2RecentFailuresMap=" + + nodeId2RecentFailuresMap + '}'; } + + private static class RecentFailureSnapshot { + + private final Map recentFailures; + private final long reportTime; + + private RecentFailureSnapshot(final Map 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 99ec79badab20..7a07345c4ae2e 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.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 void sleep4NonReportException() { @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 @@ protected void handlePipeException(final Event event, final PipeException e) { 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 d5024ddd50ff6..2cb31c81fd05d 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 @@ 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 void handle( 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 void handle( 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 0000000000000..ee00946316deb --- /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 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 getRecentFailures() { + return getRecentFailures(System.currentTimeMillis()); + } + + @TestOnly + synchronized Map getRecentFailures(final long currentTime) { + final Map 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 0000000000000..722da021cbd9f --- /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 0000000000000..4535bf57fe5a9 --- /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 18653a52c5708..c36825d528d1d 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 @@ private 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 @@ private 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 @@ private 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 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 32f51a173bec4..645ce6d2b85cb 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 b6091ab76f7e6..eddd1d29fad41 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 void testCoordinatorAggregatesNullableDegradedStatus() { 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 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 0000000000000..422dcbf263377 --- /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 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 0000000000000..5798ec192ff72 --- /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 d52dc08813fbd..4c5506a751423 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 pipeRemainingEventCountList 4: optional list pipeRemainingTimeList 5: optional list pipeDegradedStatusList + 6: optional list> pipeRecentFailureList } struct TLicense { @@ -355,4 +356,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 7aa75731545e5..1111740f759ab 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 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 cc5e0b7dc2511..5bd61971f84e3 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 pipeRemainingTimeList 17: optional map dataRegionRawDataSize 18: optional list pipeDegradedStatusList + 19: optional list> pipeRecentFailureList } struct TPipeHeartbeatReq {