apache · Caideyipi · Aug 3, 2026
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
@@ -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/...org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java b/...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/...src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java b/...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
@@ -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/.../main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java b/.../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/...ain/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeTaskAgent.java b/...ain/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<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 @@ 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/.../org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeRuntimeCoordinator.java b/.../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,14 +97,16 @@ public void parseHeartbeat(
       /* @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(
             pipeMetaByteBufferListFromDataNode,
             pipeCompletedListFromAgent,
             pipeRemainingEventCountListFromAgent,
             pipeRemainingTimeListFromAgent,
-            pipeDegradedStatusListFromAgent));
+            pipeDegradedStatusListFromAgent,
+            pipeRecentFailureListFromAgent));
   }
 }
diff --git a/...org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeat.java b/...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,13 +36,30 @@ 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,
       /* @Nullable */ final List<Boolean> pipeCompletedListFromAgent,
       /* @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 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<String, Long> getRecentFailures(final PipeStaticMeta pipeStaticMeta) {
+    return recentFailuresMap.get(pipeStaticMeta);
+  }
+
   public boolean isEmpty() {
     return pipeMetaMap.isEmpty();
   }

diff --git a/...ache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java b/...ache/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<Integer, PipeTaskMeta> pipeTaskMetaMapFromCoordinator =
           pipeMetaFromCoordinator.getRuntimeMeta().getConsensusGroupId2TaskMetaMap();

diff --git a/...e/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatScheduler.java b/...e/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/.../src/test/java/org/apache/iotdb/confignode/consensus/response/pipe/PipeTableRespTest.java b/.../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<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/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java b/.../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<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 @@ 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<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/...ore/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java b/...ore/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,13 +524,15 @@ private static class PipeMetaReport {
     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);
       resp.setPipeCompletedList(pipeCompletedList);
       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/...main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java b/...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,13 +243,15 @@ 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,
           e.getMessage());
       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);
+    }
+  }
 }