This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new e0fbbcc Add compaction threshold policy (#1910)
e0fbbcc is described below
commit e0fbbcc390561805c0cf681ed7aff894cb7f2f13
Author: Ivan Kelly <[email protected]>
AuthorDate: Tue Jun 5 20:02:18 2018 +0200
Add compaction threshold policy (#1910)
Policy is number of bytes at which compaction should be triggered for
a topic.
This patch contains the policy itself, and the check. Subsequent patches
will include the monitor and rest api.
---
.../service/persistent/PersistentSubscription.java | 4 ++
.../broker/service/persistent/PersistentTopic.java | 36 ++++++++++
.../pulsar/broker/service/PersistentTopicTest.java | 83 ++++++++++++++++++++++
.../pulsar/common/policies/data/Policies.java | 8 ++-
4 files changed, 129 insertions(+), 2 deletions(-)
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 bdace55..54ac65f 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 class PersistentSubscription implements
Subscription {
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 3cfddb0..4768fa8 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 class PersistentTopic implements Topic,
AddEntryCallback {
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 17d282e..db35e74 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.doNothing;
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.admin.AdminResource;
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.BeforeMethod;
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 class PersistentTopicTest {
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 class PersistentTopicTest {
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 59c1b3a..a33119e 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 class Policies {
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 class Policies {
&& 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 class Policies {
.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();
}
}
--
To stop receiving notification emails like this one, please contact
[email protected].
