part

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java

diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
index 2ab2061..aec2f63 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
@@ -442,8 +442,11 @@
       }
       pendingQueue.pollLast();
     }
-    if (pendingQueue.peekLast() instanceof ProgressReportEvent) {
-      final ProgressReportEvent oldEvent = (ProgressReportEvent) pendingQueue.peekLast();
+    final Event last = pendingQueue.peekLast();
+    if (last instanceof PipeRealtimeEvent
+        && ((PipeRealtimeEvent) last).getEvent() instanceof ProgressReportEvent) {
+      final ProgressReportEvent oldEvent =
+          (ProgressReportEvent) ((PipeRealtimeEvent) last).getEvent();
       oldEvent.bindProgressIndex(
           oldEvent
               .getProgressIndex()