If all partition messages are sent, cancel the delayed tasks to avoid full gc or oom

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/ReplicaManager.java

@@ -143,6 +143,8 @@ 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();
             }
         };