diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/ReplicaManager.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/ReplicaManager.java
index c977471a08..bc300f80f5 100644
--- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/ReplicaManager.java
+++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/ReplicaManager.java
@@ -163,6 +163,9 @@ public CompletableFuture<Map<TopicPartition, ProduceResponse.PartitionResponse>>
                             return;
                         }
                         if (restTopicPartitionNum == 0) {
+                            // If all tasks are sent, cancel the timer tasks to avoid full gc or oom
+                            producePurgatory.checkAndComplete(new DelayedOperationKey
+                                    .TopicPartitionOperationKey(topicPartition));
                             complete.run();
                         }
                     };