From 93c06be262bcbfb710922d62611853ff7d69fc0a Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Thu, 11 Jul 2024 17:03:21 +0800
Subject: [PATCH] fix(stream): free checkpoint trigger block in case of
 redundant trigger block recved.

---
 source/libs/stream/src/streamCheckpoint.c | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/source/libs/stream/src/streamCheckpoint.c b/source/libs/stream/src/streamCheckpoint.c
index e555db82f6..b490b0e02a 100644
--- a/source/libs/stream/src/streamCheckpoint.c
+++ b/source/libs/stream/src/streamCheckpoint.c
@@ -169,6 +169,8 @@ int32_t streamProcessCheckpointTriggerBlock(SStreamTask* pTask, SStreamDataBlock
             " recv expired checkpoint-trigger block, checkpointId:%" PRId64 " transId:%d, discard",
             id, vgId, pTask->chkInfo.checkpointId, checkpointId, transId);
     taosThreadMutexUnlock(&pTask->lock);
+
+    streamFreeQitem((SStreamQueueItem*)pBlock);
     return TSDB_CODE_SUCCESS;
   }
 
@@ -188,6 +190,8 @@ int32_t streamProcessCheckpointTriggerBlock(SStreamTask* pTask, SStreamDataBlock
 
     streamTaskOpenUpstreamInput(pTask, pBlock->srcTaskId);
     taosThreadMutexUnlock(&pTask->lock);
+
+    streamFreeQitem((SStreamQueueItem*)pBlock);
     return TSDB_CODE_SUCCESS;
   }
 
@@ -197,6 +201,8 @@ int32_t streamProcessCheckpointTriggerBlock(SStreamTask* pTask, SStreamDataBlock
               " discard",
               id, vgId, pActiveInfo->activeId, checkpointId);
       taosThreadMutexUnlock(&pTask->lock);
+
+      streamFreeQitem((SStreamQueueItem*)pBlock);
       return TSDB_CODE_SUCCESS;
     } else {  // checkpointId == pActiveInfo->activeId
       if (pActiveInfo->allUpstreamTriggerRecv == 1) {
@@ -205,6 +211,7 @@ int32_t streamProcessCheckpointTriggerBlock(SStreamTask* pTask, SStreamDataBlock
             "checkpointId:%" PRId64 " transId:%d",
             id, vgId, checkpointId, transId);
         taosThreadMutexUnlock(&pTask->lock);
+        streamFreeQitem((SStreamQueueItem*)pBlock);
         return TSDB_CODE_SUCCESS;
       }
 
@@ -219,6 +226,7 @@ int32_t streamProcessCheckpointTriggerBlock(SStreamTask* pTask, SStreamDataBlock
                    pTask->id.idStr, p->upstreamTaskId, p->upstreamNodeId, p->checkpointId, p->recvTs);
 
             taosThreadMutexUnlock(&pTask->lock);
+            streamFreeQitem((SStreamQueueItem*)pBlock);
             return TSDB_CODE_SUCCESS;
           }
         }