f

Author: Caideyipi

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

@@ -163,16 +163,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() {
-    if (Objects.isNull(insertNode)) {
+    final InsertNode node = insertNode;
+    if (Objects.isNull(node)) {
       return null;
     }
-    return Objects.nonNull(insertNode.getTargetPath())
-        ? insertNode.getTargetPath().getFullPath()
-        : null;
+    final PartialPath targetPath = node.getTargetPath();
+    return Objects.nonNull(targetPath) ? targetPath.getFullPath() : null;
   }
 
   public long getExtractTime() {
@@ -250,10 +254,14 @@ public PipeInsertNodeTabletInsertionEvent shallowCopySelfAndBindPipeTaskMetaForP
       final boolean skipIfNoPrivileges,
       final long startTime,
       final long endTime) {
+    final InsertNode node = insertNode;
+    if (Objects.isNull(node)) {
+      throw new PipeException("InsertNode has been released");
+    }
     return new PipeInsertNodeTabletInsertionEvent(
         getRawIsTableModelEvent(),
         getSourceDatabaseNameFromDataRegion(),
-        insertNode,
+        node,
         pipeName,
         creationTime,
         pipeTaskMeta,
@@ -269,28 +277,37 @@ public PipeInsertNodeTabletInsertionEvent shallowCopySelfAndBindPipeTaskMetaForP
 
   @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
   public void throwIfNoPrivilege() throws Exception {
     if (skipIfNoPrivileges) {
       return;
     }
-    if (Objects.nonNull(insertNode.getTargetPath())) {
+    final InsertNode node = insertNode;
+    if (Objects.isNull(node)) {
+      // Event is released, skip privilege check
+      return;
+    }
+    if (Objects.nonNull(node.getTargetPath())) {
       if (isTableModelEvent()) {
         checkTableName(
-            DeviceIDFactory.getInstance().getDeviceID(insertNode.getTargetPath()).getTableName());
+            DeviceIDFactory.getInstance().getDeviceID(node.getTargetPath()).getTableName());
       } else {
-        checkTreePattern(insertNode.getDeviceID(), insertNode.getMeasurements());
+        checkTreePattern(node.getDeviceID(), node.getMeasurements());
       }
     } else if (insertNode instanceof InsertRowsNode) {
-      for (final InsertNode node : ((InsertRowsNode) insertNode).getInsertRowNodeList()) {
+      for (final InsertNode subNode : ((InsertRowsNode) node).getInsertRowNodeList()) {
         if (isTableModelEvent()) {
           checkTableName(
-              DeviceIDFactory.getInstance().getDeviceID(node.getTargetPath()).getTableName());
+              DeviceIDFactory.getInstance().getDeviceID(subNode.getTargetPath()).getTableName());
         } else {
-          checkTreePattern(node.getDeviceID(), node.getMeasurements());
+          checkTreePattern(subNode.getDeviceID(), subNode.getMeasurements());
         }
       }
     }
@@ -473,6 +490,9 @@ private List<TabletInsertionEventParser> initEventParsers() {
 
       eventParsers = 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:
@@ -569,23 +589,25 @@ public List<PipeRawTabletInsertionEvent> toRawTabletInsertionEvents() {
 
   @Override
   public String toString() {
+    final InsertNode node = insertNode;
     return String.format(
             "PipeInsertNodeTabletInsertionEvent{progressIndex=%s, isAligned=%s, isGeneratedByPipe=%s, eventParsers=%s}",
             progressIndex,
-            Objects.nonNull(insertNode) ? insertNode.isAligned() : null,
-            Objects.nonNull(insertNode) ? insertNode.isGeneratedByPipe() : null,
+            Objects.nonNull(node) ? node.isAligned() : null,
+            Objects.nonNull(node) ? node.isGeneratedByPipe() : null,
             eventParsers)
         + " - "
         + super.toString();
   }
 
   @Override
   public String coreReportMessage() {
+    final InsertNode node = insertNode;
     return String.format(
             "PipeInsertNodeTabletInsertionEvent{progressIndex=%s, isAligned=%s, isGeneratedByPipe=%s}",
             progressIndex,
-            Objects.nonNull(insertNode) ? insertNode.isAligned() : null,
-            Objects.nonNull(insertNode) ? insertNode.isGeneratedByPipe() : null)
+            Objects.nonNull(node) ? node.isAligned() : null,
+            Objects.nonNull(node) ? node.isGeneratedByPipe() : null)
         + " - "
         + super.coreReportMessage();
   }
@@ -610,12 +632,15 @@ public PipeEventResource eventResourceBuilder() {
   // invoked, the event will soon be released.
   @Override
   public long ramBytesUsed() {
-    return bytes > 0
-        ? bytes
-        : (bytes =
-            INSTANCE_SIZE
-                + (Objects.nonNull(insertNode) ? InsertNodeMemoryEstimator.sizeOf(insertNode) : 0)
-                + (Objects.nonNull(progressIndex) ? progressIndex.ramBytesUsed() : 0));
+    if (bytes > 0) {
+      return bytes;
+    }
+    final InsertNode node = insertNode;
+    bytes =
+        INSTANCE_SIZE
+            + (Objects.nonNull(node) ? InsertNodeMemoryEstimator.sizeOf(node) : 0)
+            + (Objects.nonNull(progressIndex) ? progressIndex.ramBytesUsed() : 0);
+    return bytes;
   }
 
   private static class PipeInsertNodeTabletInsertionEventResource extends PipeEventResource {