Monitor thread to check if topics need compaction (apache#1915)

Similar to other monitors, runs every 60 seconds by default, and
checks whether any persistent topics need to be compacted.

conf/broker.conf

@@ -190,6 +190,9 @@ maxConsumersPerTopic=0
 # Using a value of 0, is disabling maxConsumersPerSubscription-limit check.
 maxConsumersPerSubscription=0
 
+# Interval between checks to see if topics with compaction policies need to be compacted
+brokerServiceCompactionMonitorIntervalInSeconds=60
+
 ### --- Authentication --- ###
 # Role names that are treated as "proxy roles". If the broker sees a request with
 #role as proxyRoles - it will demand to see a valid original principal.

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -450,6 +450,9 @@ public class ServiceConfiguration implements PulsarConfiguration {
     @FieldContext(dynamic = true)
     private boolean preferLaterVersions = false;
 
+    // Interval between checks to see if topics with compaction policies need to be compacted
+    private int brokerServiceCompactionMonitorIntervalInSeconds = 60;
+
     private String schemaRegistryStorageClassName = "org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorageFactory";
     private Set<String> schemaRegistryCompatibilityCheckers = Sets.newHashSet(
         "org.apache.pulsar.broker.service.schema.JsonSchemaCompatibilityCheck"
@@ -1719,4 +1722,11 @@ public int getS3ManagedLedgerOffloadReadBufferSizeInBytes() {
         return this.s3ManagedLedgerOffloadReadBufferSizeInBytes;
     }
 
+    public void setBrokerServiceCompactionMonitorIntervalInSeconds(int interval) {
+        this.brokerServiceCompactionMonitorIntervalInSeconds = interval;
+    }
+
+    public int getBrokerServiceCompactionMonitorIntervalInSeconds() {
+        return this.brokerServiceCompactionMonitorIntervalInSeconds;
+    }
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -165,6 +165,7 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
 
     private final ScheduledExecutorService inactivityMonitor;
     private final ScheduledExecutorService messageExpiryMonitor;
+    private final ScheduledExecutorService compactionMonitor;
 
     private DistributedIdGenerator producerNameGenerator;
 
@@ -227,6 +228,9 @@ public BrokerService(PulsarService pulsar) throws Exception {
                 .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-inactivity-monitor"));
         this.messageExpiryMonitor = Executors
                 .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-msg-expiry-monitor"));
+        this.compactionMonitor =
+            Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-compaction-monitor"));
+
         this.backlogQuotaManager = new BacklogQuotaManager(pulsar);
         this.backlogQuotaChecker = Executors
                 .newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-backlog-quota-checker"));
@@ -309,6 +313,7 @@ public void start() throws Exception {
         this.startStatsUpdater();
         this.startInactivityMonitor();
         this.startMessageExpiryMonitor();
+        this.startCompactionMonitor();
         this.startBacklogQuotaChecker();
         // register listener to capture zk-latency
         ClientCnxnAspect.addListener(zkStatsListener);
@@ -350,6 +355,14 @@ void startMessageExpiryMonitor() {
                 TimeUnit.MINUTES);
     }
 
+    void startCompactionMonitor() {
+        int interval = pulsar().getConfiguration().getBrokerServiceCompactionMonitorIntervalInSeconds();
+        if (interval > 0) {
+            compactionMonitor.scheduleAtFixedRate(safeRun(() -> checkCompaction()),
+                                                  interval, interval, TimeUnit.SECONDS);
+        }
+    }
+
     void startBacklogQuotaChecker() {
         if (pulsar().getConfiguration().isBacklogQuotaCheckEnabled()) {
             final int interval = pulsar().getConfiguration().getBacklogQuotaCheckIntervalInSeconds();
@@ -387,6 +400,7 @@ public void close() throws IOException {
         statsUpdater.shutdown();
         inactivityMonitor.shutdown();
         messageExpiryMonitor.shutdown();
+        compactionMonitor.shutdown();
         backlogQuotaChecker.shutdown();
         authenticationService.close();
         pulsarStats.close();
@@ -838,6 +852,14 @@ public void checkMessageExpiry() {
         forEachTopic(Topic::checkMessageExpiry);
     }
 
+    public void checkCompaction() {
+        forEachTopic((t) -> {
+                if (t instanceof PersistentTopic) {
+                    ((PersistentTopic) t).checkCompaction();
+                }
+            });
+    }
+
     public void checkMessageDeduplicationInfo() {
         forEachTopic(Topic::checkMessageDeduplicationInfo);
     }

site/_data/config/broker.yaml

@@ -93,6 +93,9 @@ configs:
 - name: messageExpiryCheckIntervalInMinutes
   default: '5'
   description: How frequently to proactively check and purge expired messages
+- name: brokerServiceCompactionMonitorIntervalInSeconds
+  default: '60'
+  description: Interval between checks to see if topics with compaction policies need to be compacted
 - name: activeConsumerFailoverDelayTimeMillis
   default: '1000'
   description: How long to delay rewinding cursor and dispatching messages when active consumer is changed.