(iotdb) branch dev/1.3 updated: Pipe: Fix the problem that the Pipe memory block fails to call the Wait function (#15688)

[justinchen]

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

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


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

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

    Pipe: Fix the problem that the Pipe memory block fails to call the Wait 
function (#15688)
---
 .../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 34934bef812..690a7c54083 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
@@ -271,11 +271,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);
     }
   }