(iotdb) branch master updated: Pipe: Fix the problem that the Pipe memory block fails to call the Wait function (#15687)

[justinchen]

This is an automated email from the ASF dual-hosted git repository.

justinchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/master by this push:
     new fb86a969f86 Pipe: Fix the problem that the Pipe memory block fails to 
call the Wait function (#15687)
fb86a969f86 is described below

commit fb86a969f860a5ff0a81c68a808b2fd647cfa3df
Author: Zhenyu Luo <luoluoyuyu2...@gmail.com>
AuthorDate: Tue Jun 10 12:32:30 2025 +0800

    Pipe: Fix the problem that the Pipe memory block fails to call the Wait 
function (#15687)
---
 .../connector/payload/evolvable/batch/PipeTabletEventBatch.java   | 7 +++----
 .../apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java   | 8 ++++----
 2 files changed, 7 insertions(+), 8 deletions(-)

diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTabletEventBatch.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTabletEventBatch.java
index 9a3d9f51785..1cb9f50d92d 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTabletEventBatch.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTabletEventBatch.java
@@ -60,10 +60,9 @@ public abstract class PipeTabletEventBatch implements 
AutoCloseable {
 
     // limit in buffer size
     this.allocatedMemoryBlock =
-        pipeModelFixedMemoryBlock =
-            PipeDataNodeResourceManager.memory()
-                .forceAllocateForModelFixedMemoryBlock(
-                    requestMaxBatchSizeInBytes, PipeMemoryBlockType.BATCH);
+        PipeDataNodeResourceManager.memory()
+            .forceAllocateForModelFixedMemoryBlock(
+                requestMaxBatchSizeInBytes, PipeMemoryBlockType.BATCH);
 
     if (getMaxBatchSizeInBytes() != requestMaxBatchSizeInBytes) {
       LOGGER.info(
diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
index 90e168a8f9b..786a8c63209 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryManager.java
@@ -262,11 +262,11 @@ public class PipeMemoryManager {
       }
     }
 
-    if (getFreeMemorySizeInBytes() < fixedSizeInBytes) {
-      return (PipeModelFixedMemoryBlock) 
forceAllocateWithRetry(getFreeMemorySizeInBytes(), type);
-    }
-
     synchronized (this) {
+      if (getFreeMemorySizeInBytes() < fixedSizeInBytes) {
+        return (PipeModelFixedMemoryBlock) 
forceAllocateWithRetry(getFreeMemorySizeInBytes(), type);
+      }
+
       return (PipeModelFixedMemoryBlock) 
forceAllocateWithRetry(fixedSizeInBytes, type);
     }
   }