Add if to avoid assertTrue error (#7271)

apache · 29 days ago · 35541f457afa7e3075893c633bdb1ea8765a399c · 35541f4
1 parent 6f10996
commit 35541f457afa7e3075893c633bdb1ea8765a399c
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SinkHandle.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SinkHandle.java
    return serde.serialize(pair.left);
  }

  void acknowledgeTsBlock(int startSequenceId, int endSequenceId) {
    long freedBytes = 0L;
    synchronized (this) {
      if (aborted || closed) {
        return;
      }
      Iterator<Entry<Integer, Pair<TsBlock, Long>>> iterator =
          sequenceIdToTsBlock.entrySet().iterator();
      while (iterator.hasNext()) {
        Entry<Integer, Pair<TsBlock, Long>> entry = iterator.next();
        if (entry.getKey() < startSequenceId) {
          continue;
        }
        if (entry.getKey() >= endSequenceId) {
          break;
        }

        freedBytes += entry.getValue().right;
        bufferRetainedSizeInBytes -= entry.getValue().right;
        iterator.remove();
        logger.info("ack TsBlock {}.", entry.getKey());
      }
    }
     if (isFinished()) {
       sinkHandleListener.onFinish(this);
     }
-    localMemoryManager.getQueryPool().free(localFragmentInstanceId.getQueryId(), freedBytes);
+    // there may exist duplicate ack message in network caused by caller retrying, if so duplicate
+    // ack message's freedBytes may be zero
+    if (freedBytes > 0) {
+      localMemoryManager.getQueryPool().free(localFragmentInstanceId.getQueryId(), freedBytes);
+    }
   }
 
   public TEndPoint getRemoteEndpoint() {