[fix][broker] Make sure all inflight writes have finished before comp…

…letion of compaction (apache#21067) (cherry picked from commit bb9c9b4)
streamnative · Aug 29, 2023 · 810a2f0 · 810a2f0
1 parent a5d1ca0
commit 810a2f0
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
@@ -279,6 +279,8 @@ private void phaseTwoLoop(RawReader reader, MessageId to, Map<String, MessageId>
                                     }
                                 });
                         if (to.equals(id)) {
+                            // make sure all inflight writes have finished
+                            outstanding.acquire(MAX_OUTSTANDING);
                             addFuture.whenComplete((res, exception2) -> {
                                 if (exception2 == null) {
                                     promise.complete(null);