If all partition messages are sent, cancel the delayed tasks to avoid…

… full gc or oom (#1749)

Fixes #1739

### Motivation

The request.timeout.ms setting is too large to cause broker full gc or oom.

### Modifications

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

(cherry picked from commit 7994cd6)

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();
             }
         };