Pipe: Fix NullPointerException in concurrent event access (#16849) (#16872)

* fix

* fix

* fix

* fix

* fix

(cherry picked from commit 5bc4779)

Author: luoluoyuyu

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java

@@ -142,9 +142,9 @@ protected boolean executeOnce() {
         throw e;
       } else {
         LOGGER.info(
-            "{} in pipe transfer, ignored because the connector subtask is dropped.",
+            "{} in pipe transfer, ignored because the connector subtask is dropped.{}",
             e.getClass().getSimpleName(),
-            e);
+            e.getMessage() != null ? " Message: " + e.getMessage() : "");
         clearReferenceCountAndReleaseLastEvent(event);
       }
     } catch (final Exception e) {
@@ -161,7 +161,8 @@ protected boolean executeOnce() {
             e);
       } else {
         LOGGER.info(
-            "Exception in pipe transfer, ignored because the connector subtask is dropped.", e);
+            "Exception in pipe transfer, ignored because the connector subtask is dropped.{}",
+            e.getMessage() != null ? " Message: " + e.getMessage() : "");
         clearReferenceCountAndReleaseLastEvent(event);
       }
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java

@@ -106,13 +106,20 @@ public InsertNode getInsertNode() {
   }
 
   public ByteBuffer getByteBuffer() throws WALPipeException {
-    return insertNode.serializeToByteBuffer();
+    final InsertNode node = insertNode;
+    if (Objects.isNull(node)) {
+      throw new PipeException("InsertNode has been released");
+    }
+    return node.serializeToByteBuffer();
   }
 
   public String getDeviceId() {
-    return Objects.nonNull(insertNode.getDevicePath())
-        ? insertNode.getDevicePath().getFullPath()
-        : null;
+    final InsertNode node = insertNode;
+    if (Objects.isNull(node)) {
+      return null;
+    }
+    final PartialPath targetPath = node.getDevicePath();
+    return Objects.nonNull(targetPath) ? targetPath.getFullPath() : null;
   }
 
   public long getExtractTime() {
@@ -185,13 +192,21 @@ public PipeInsertNodeTabletInsertionEvent shallowCopySelfAndBindPipeTaskMetaForP
       final PipePattern pattern,
       final long startTime,
       final long endTime) {
+    final InsertNode node = insertNode;
+    if (Objects.isNull(node)) {
+      throw new PipeException("InsertNode has been released");
+    }
     return new PipeInsertNodeTabletInsertionEvent(
         insertNode, pipeName, creationTime, pipeTaskMeta, pattern, startTime, endTime);
   }
 
   @Override
   public boolean isGeneratedByPipe() {
-    return insertNode.isGeneratedByPipe();
+    final InsertNode node = insertNode;
+    if (Objects.isNull(node)) {
+      throw new PipeException("InsertNode has been released");
+    }
+    return node.isGeneratedByPipe();
   }
 
   @Override
@@ -328,6 +343,9 @@ private List<TabletInsertionDataContainer> initDataContainers() {
 
       dataContainers = new ArrayList<>();
       final InsertNode node = getInsertNode();
+      if (Objects.isNull(node)) {
+        throw new PipeException("InsertNode has been released");
+      }
       switch (node.getType()) {
         case INSERT_ROW:
         case INSERT_TABLET:
@@ -393,6 +411,7 @@ public List<PipeRawTabletInsertionEvent> toRawTabletInsertionEvents() {
 
   @Override
   public String toString() {
+    final InsertNode insertNode = this.insertNode;
     return String.format(
             "PipeInsertNodeTabletInsertionEvent{progressIndex=%s, isAligned=%s, isGeneratedByPipe=%s, dataContainers=%s}",
             progressIndex,
@@ -405,6 +424,7 @@ public String toString() {
 
   @Override
   public String coreReportMessage() {
+    final InsertNode insertNode = this.insertNode;
     return String.format(
             "PipeInsertNodeTabletInsertionEvent{progressIndex=%s, isAligned=%s, isGeneratedByPipe=%s}",
             progressIndex,