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 2025/07/13 07:44:53 UTC
(pulsar) branch master updated: [improve][broker] Enhance compaction triggering with message check to avoid unnecessary operations (#24449)
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 a40ac3c30e3 [improve][broker] Enhance compaction triggering with message check to avoid unnecessary operations (#24449)
a40ac3c30e3 is described below
commit a40ac3c30e3202ccf992e84be481f6ece588ac62
Author: Cong Zhao <zh...@apache.org>
AuthorDate: Sun Jul 13 15:44:47 2025 +0800
[improve][broker] Enhance compaction triggering with message check to avoid unnecessary operations (#24449)
---
.../broker/admin/impl/PersistentTopicsBase.java | 30 ++++++++---------
.../broker/service/persistent/PersistentTopic.java | 39 +++++++++++++++++-----
.../pulsar/broker/service/PersistentTopicTest.java | 24 ++++++++-----
.../apache/pulsar/compaction/CompactionTest.java | 4 +--
4 files changed, 63 insertions(+), 34 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index a6ce51e01ed..c5d3e928325 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -4263,31 +4263,31 @@ public class PersistentTopicsBase extends AdminResource {
protected void internalTriggerCompactionNonPartitionedTopic(AsyncResponse asyncResponse, boolean authoritative) {
validateTopicOwnershipAsync(topicName, authoritative)
- .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.COMPACT))
- .thenCompose(__ -> getTopicReferenceAsync(topicName))
- .thenAccept(topic -> {
- try {
- ((PersistentTopic) topic).triggerCompaction();
+ .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.COMPACT))
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(topic -> ((PersistentTopic) topic).triggerCompactionWithCheckHasMoreMessages()
+ .whenComplete((result, ex) -> {
+ if (ex == null) {
asyncResponse.resume(Response.noContent().build());
- } catch (AlreadyRunningException e) {
- resumeAsyncResponseExceptionally(asyncResponse,
- new RestException(Status.CONFLICT, e.getMessage()));
return;
- } catch (Exception e) {
+ }
+ ex = FutureUtil.unwrapCompletionException(ex);
+ if (ex instanceof AlreadyRunningException) {
+ resumeAsyncResponseExceptionally(asyncResponse,
+ new RestException(Status.CONFLICT, ex.getMessage()));
+ } else {
log.error("[{}] Failed to trigger compaction on topic {}", clientAppId(),
- topicName, e);
- resumeAsyncResponseExceptionally(asyncResponse, new RestException(e));
- return;
+ topicName, ex);
+ resumeAsyncResponseExceptionally(asyncResponse, new RestException(ex));
}
- }).exceptionally(ex -> {
+ })).exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
if (isNot307And404Exception(ex)) {
log.error("[{}] Failed to trigger compaction for {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
return null;
- }
- );
+ });
}
protected CompletableFuture<LongRunningProcessStatus> internalCompactionStatusAsync(boolean authoritative) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 91d964125bf..fb41bc52ea6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -2137,16 +2137,12 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
if (backlogEstimate > compactionThreshold) {
if (log.isDebugEnabled()) {
log.debug(
- "topic:{} backlogEstimate:{} is bigger than compactionThreshold:{}. Triggering "
- + "compaction", topic, backlogEstimate, compactionThreshold);
- }
- try {
- triggerCompaction();
- } catch (AlreadyRunningException are) {
- log.debug("[{}] Compaction already running, so don't trigger again, "
- + "even though backlog({}) is over threshold({})",
- name, backlogEstimate, compactionThreshold);
+ "topic:{} backlogEstimate:{} is bigger than compactionThreshold:{}. Triggering "
+ + "compaction", topic, backlogEstimate, compactionThreshold);
}
+
+ triggerCompactionWithCheckHasMoreMessages();
+
}
}
} catch (Exception e) {
@@ -4042,6 +4038,31 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
return completableFuture;
}
+ public synchronized CompletableFuture<Void> triggerCompactionWithCheckHasMoreMessages() {
+ return getLastDispatchablePosition().thenCombine(topicCompactionService.getLastCompactedPosition(),
+ (lastDispatchablePosition, lastCompactedPosition) -> {
+ if (lastDispatchablePosition == null) {
+ lastDispatchablePosition = PositionFactory.EARLIEST;
+ }
+ return lastCompactedPosition == null || lastDispatchablePosition.compareTo(lastCompactedPosition) > 0;
+ }).thenAccept(hasMoreMessagesToBeCompacted -> {
+ if (!hasMoreMessagesToBeCompacted) {
+ log.info("[{}] No more messages to compact, skip triggering compaction", topic);
+ return;
+ }
+ try {
+ triggerCompaction();
+ } catch (PulsarServerException | AlreadyRunningException e) {
+ throw new CompletionException(e);
+ }
+ }).whenComplete((__, ex) -> {
+ if (ex != null) {
+ ex = FutureUtil.unwrapCompletionException(ex);
+ log.error("[{}] Trigger Compaction failure.", topic, ex);
+ }
+ });
+ }
+
public synchronized void triggerCompaction()
throws PulsarServerException, AlreadyRunningException {
if (currentCompaction.isDone()) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
index 7db8119efce..ca8f762adc4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
@@ -1813,22 +1813,26 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
.createPolicies(TopicName.get(successTopicName).getNamespaceObject(),
policies);
- PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService);
+ PersistentTopic topic = Mockito.spy(new PersistentTopic(successTopicName, ledgerMock, brokerService));
topic.initialize().get();
+ doAnswer(invocationOnMock -> {
+ topic.triggerCompaction();
+ return CompletableFuture.completedFuture(null);
+ }).when(topic).triggerCompactionWithCheckHasMoreMessages();
topic.checkCompaction();
- verify(compactor, times(0)).compact(anyString());
+ verify(topic, times(0)).triggerCompactionWithCheckHasMoreMessages();
doReturn(10L).when(ledgerMock).getTotalSize();
doReturn(10L).when(ledgerMock).getEstimatedBacklogSize();
topic.checkCompaction();
- verify(compactor, times(1)).compact(anyString());
+ verify(topic, times(1)).triggerCompactionWithCheckHasMoreMessages();
// run a second time, shouldn't run again because already running
topic.checkCompaction();
- verify(compactor, times(1)).compact(anyString());
+ verify(topic, times(1)).triggerCompactionWithCheckHasMoreMessages();
}
@Test
@@ -1849,21 +1853,25 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
.createPolicies(TopicName.get(successTopicName).getNamespaceObject(),
policies);
- PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService);
+ PersistentTopic topic = Mockito.spy(new PersistentTopic(successTopicName, ledgerMock, brokerService));
topic.initialize().get();
+ doAnswer(invocationOnMock -> {
+ topic.triggerCompaction();
+ return CompletableFuture.completedFuture(null);
+ }).when(topic).triggerCompactionWithCheckHasMoreMessages();
topic.checkCompaction();
- verify(compactor, times(0)).compact(anyString());
+ verify(topic, times(0)).triggerCompactionWithCheckHasMoreMessages();
doReturn(10L).when(subCursor).getEstimatedSizeSinceMarkDeletePosition();
topic.checkCompaction();
- verify(compactor, times(1)).compact(anyString());
+ verify(topic, times(1)).triggerCompactionWithCheckHasMoreMessages();
// run a second time, shouldn't run again because already running
topic.checkCompaction();
- verify(compactor, times(1)).compact(anyString());
+ verify(topic, times(1)).triggerCompactionWithCheckHasMoreMessages();
}
@Test
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
index 45e46304b54..4f773b03216 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
@@ -2176,7 +2176,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest {
private void triggerCompactionAndWait(String topicName) throws Exception {
PersistentTopic persistentTopic =
(PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get();
- persistentTopic.triggerCompaction();
+ persistentTopic.triggerCompactionWithCheckHasMoreMessages().join();
CompletableFuture<Long> currentCompaction =
(CompletableFuture<Long>) FieldUtils.readDeclaredField(persistentTopic, "currentCompaction", true);
currentCompaction.get(10, TimeUnit.SECONDS);
@@ -2383,7 +2383,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest {
.topic(topicName).readCompacted(true).subscriptionName("s1")
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
.subscribe();
- persistentTopic.triggerCompaction();
+ persistentTopic.triggerCompactionWithCheckHasMoreMessages().join();
Awaitility.await().untilAsserted(() -> {
assertEquals(persistentTopic.getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 1);
});