Fix the regression for streamnative#1749 (streamnative#2024)

### Motivation streamnative#1923 removes the completion for the produce purgatory in streamnative#1749. ### Modification Added back the logic to check and complete the produce purgatory.
BewareMyPower · Oct 31, 2023 · aa2f171 · aa2f171
1 parent db4a9ad
commit aa2f171
Showing 1 changed file with 3 additions and 0 deletions.
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
@@ -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();
                         }
                     };