You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zh...@apache.org on 2023/08/28 01:42:30 UTC
[pulsar] branch master updated: [fix][broker] Make sure all inflight writes have finished before completion of compaction (#21067)
This is an automated email from the ASF dual-hosted git repository.
zhaocong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new bb9c9b421b0 [fix][broker] Make sure all inflight writes have finished before completion of compaction (#21067)
bb9c9b421b0 is described below
commit bb9c9b421b02c5f88fdd508ede387686065388f5
Author: Cong Zhao <zh...@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)
---
.../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 2efec1fc188..c151940e23b 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
@@ -276,6 +276,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);