sijie closed pull request #1910: Add compaction threshold policy
URL: https://github.com/apache/incubator-pulsar/pull/1910
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index bdace5513a..54ac65f4bf 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -630,6 +630,10 @@ public double getExpiredMessageRate() {
return expiryMonitor.getMessageExpiryRate();
}
+ public long estimateBacklogSize() {
+ return cursor.getEstimatedSizeSinceMarkDeletePosition();
+ }
+
public SubscriptionStats getStats() {
SubscriptionStats subStats = new SubscriptionStats();
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 3cfddb0b3e..4768fa8967 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
@@ -970,6 +970,42 @@ public void checkMessageDeduplicationInfo() {
messageDeduplication.purgeInactiveProducers();
}
+ public void checkCompaction() {
+ TopicName name = TopicName.get(topic);
+ try {
+ Policies policies =
brokerService.pulsar().getConfigurationCache().policiesCache()
+ .get(AdminResource.path(POLICIES, name.getNamespace()))
+ .orElseThrow(() -> new KeeperException.NoNodeException());
+
+
+ if (policies.compaction_threshold != 0
+ && currentCompaction.isDone()) {
+
+ long backlogEstimate = 0;
+
+ PersistentSubscription compactionSub =
subscriptions.get(Compactor.COMPACTION_SUBSCRIPTION);
+ if (compactionSub != null) {
+ backlogEstimate = compactionSub.estimateBacklogSize();
+ } else {
+ // compaction has never run, so take full backlog size
+ backlogEstimate = ledger.getEstimatedBacklogSize();
+ }
+
+ if (backlogEstimate > policies.compaction_threshold) {
+ try {
+ triggerCompaction();
+ } catch (AlreadyRunningException are) {
+ log.debug("[{}] Compaction already running, so don't
trigger again, "
+ + "even though backlog({}) is over
threshold({})",
+ name, backlogEstimate,
policies.compaction_threshold);
+ }
+ }
+ }
+ } catch (Exception e) {
+ log.debug("[{}] Error getting policies", topic);
+ }
+ }
+
CompletableFuture<Void> startReplicator(String remoteCluster) {
log.info("[{}] Starting replicator to remote: {}", topic,
remoteCluster);
final CompletableFuture<Void> future = new CompletableFuture<>();
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 17d282e295..db35e74845 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
@@ -31,6 +31,7 @@
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
@@ -79,6 +80,7 @@
import org.apache.pulsar.broker.cache.ConfigurationCacheService;
import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService;
import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.broker.service.Topic.PublishContext;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentReplicator;
import org.apache.pulsar.broker.service.persistent.CompactorSubscription;
import
org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
@@ -115,6 +117,7 @@
import org.testng.annotations.Test;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
@@ -143,6 +146,7 @@ public void setup() throws Exception {
ServiceConfiguration svcConfig = spy(new ServiceConfiguration());
pulsar = spy(new PulsarService(svcConfig));
doReturn(svcConfig).when(pulsar).getConfiguration();
+ doReturn(mock(Compactor.class)).when(pulsar).getCompactor();
mlFactoryMock = mock(ManagedLedgerFactory.class);
doReturn(mlFactoryMock).when(pulsar).getManagedLedgerFactory();
@@ -1264,4 +1268,83 @@ public void testCompactorSubscriptionUpdatedOnInit()
throws Exception {
new CompactorSubscription(topic, compactedTopic,
Compactor.COMPACTION_SUBSCRIPTION, cursorMock);
verify(compactedTopic, Mockito.times(1)).newCompactedLedger(position,
ledgerId);
}
+
+ @Test
+ public void testCompactionTriggeredAfterThresholdFirstInvocation() throws
Exception {
+ CompletableFuture<Long> compactPromise = new CompletableFuture<>();
+ Compactor compactor = pulsar.getCompactor();
+ doReturn(compactPromise).when(compactor).compact(anyString());
+
+ Policies policies = new Policies();
+ policies.compaction_threshold = 1;
+ when(pulsar.getConfigurationCache().policiesCache()
+ .get(AdminResource.path(POLICIES,
TopicName.get(successTopicName).getNamespace())))
+ .thenReturn(Optional.of(policies));
+
+ PersistentTopic topic = new PersistentTopic(successTopicName,
ledgerMock, brokerService);
+
+ topic.checkCompaction();
+
+ verify(compactor, times(0)).compact(anyString());
+
+ doReturn(10L).when(ledgerMock).getEstimatedBacklogSize();
+
+ topic.checkCompaction();
+ verify(compactor, times(1)).compact(anyString());
+
+ // run a second time, shouldn't run again because already running
+ topic.checkCompaction();
+ verify(compactor, times(1)).compact(anyString());
+ }
+
+ @Test
+ public void testCompactionTriggeredAfterThresholdSecondInvocation() throws
Exception {
+ CompletableFuture<Long> compactPromise = new CompletableFuture<>();
+ Compactor compactor = pulsar.getCompactor();
+ doReturn(compactPromise).when(compactor).compact(anyString());
+
+ ManagedCursor subCursor = mock(ManagedCursor.class);
+ doReturn(Lists.newArrayList(subCursor)).when(ledgerMock).getCursors();
+ doReturn(Compactor.COMPACTION_SUBSCRIPTION).when(subCursor).getName();
+
+ Policies policies = new Policies();
+ policies.compaction_threshold = 1;
+ when(pulsar.getConfigurationCache().policiesCache()
+ .get(AdminResource.path(POLICIES,
TopicName.get(successTopicName).getNamespace())))
+ .thenReturn(Optional.of(policies));
+
+ PersistentTopic topic = new PersistentTopic(successTopicName,
ledgerMock, brokerService);
+
+ topic.checkCompaction();
+
+ verify(compactor, times(0)).compact(anyString());
+
+
doReturn(10L).when(subCursor).getEstimatedSizeSinceMarkDeletePosition();
+
+ topic.checkCompaction();
+ verify(compactor, times(1)).compact(anyString());
+
+ // run a second time, shouldn't run again because already running
+ topic.checkCompaction();
+ verify(compactor, times(1)).compact(anyString());
+ }
+
+ @Test
+ public void testCompactionDisabledWithZeroThreshold() throws Exception {
+ CompletableFuture<Long> compactPromise = new CompletableFuture<>();
+ Compactor compactor = pulsar.getCompactor();
+ doReturn(compactPromise).when(compactor).compact(anyString());
+
+ Policies policies = new Policies();
+ policies.compaction_threshold = 0;
+ when(pulsar.getConfigurationCache().policiesCache()
+ .get(AdminResource.path(POLICIES,
TopicName.get(successTopicName).getNamespace())))
+ .thenReturn(Optional.of(policies));
+
+ doReturn(1000L).when(ledgerMock).getEstimatedBacklogSize();
+
+ PersistentTopic topic = new PersistentTopic(successTopicName,
ledgerMock, brokerService);
+ topic.checkCompaction();
+ verify(compactor, times(0)).compact(anyString());
+ }
}
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
index 59c1b3a009..a33119e6b0 100644
---
a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
+++
b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
@@ -57,6 +57,8 @@
public int max_consumers_per_topic = 0;
public int max_consumers_per_subscription = 0;
+ public long compaction_threshold = 0;
+
@Override
public boolean equals(Object obj) {
if (obj instanceof Policies) {
@@ -75,7 +77,8 @@ public boolean equals(Object obj) {
&& Objects.equals(antiAffinityGroup,
other.antiAffinityGroup)
&& max_producers_per_topic == other.max_producers_per_topic
&& max_consumers_per_topic == other.max_consumers_per_topic
- && max_consumers_per_subscription ==
other.max_consumers_per_subscription;
+ && max_consumers_per_subscription ==
other.max_consumers_per_subscription
+ && compaction_threshold == other.compaction_threshold;
}
return false;
@@ -105,6 +108,7 @@ public String toString() {
.add("subscription_auth_mode", subscription_auth_mode)
.add("max_producers_per_topic", max_producers_per_topic)
.add("max_consumers_per_topic", max_consumers_per_topic)
- .add("max_consumers_per_subscription",
max_consumers_per_topic).toString();
+ .add("max_consumers_per_subscription", max_consumers_per_topic)
+ .add("compaction_threshold", compaction_threshold).toString();
}
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
