[pulsar] branch branch-2.11 updated: [fix][broker] Make sure all inflight writes have finished before completion of compaction (#21067)

[zhaocong]

This is an automated email from the ASF dual-hosted git repository.

zhaocong pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.11 by this push:
     new 640f6ccf62d [fix][broker] Make sure all inflight writes have finished  
before completion of compaction (#21067)
640f6ccf62d is described below

commit 640f6ccf62d0e3e6c67b37f2a7a0b2ccd0042c40
Author: Cong Zhao <zhaoc...@apache.org>
AuthorDate: Mon Aug 28 09:42:22 2023 +0800

    [fix][broker] Make sure all inflight writes have finished  before 
completion of compaction (#21067)
    
    (cherry picked from commit bb9c9b421b02c5f88fdd508ede387686065388f5)
---
 .../src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java   | 2 ++
 1 file changed, 2 insertions(+)

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
index b1e24cce7c7..5d9ae3cf9a1 100644
--- 
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 @@ public class TwoPhaseCompactor extends Compactor {
                                     }
                                 });
                         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);