Monitor if a cursor moves its mark-delete position (apache#8930)

Motivation msgBacklog or storageSize doesn't provide a clear idea if mark-delete position is advanced or not. Add a new metric in SubscriptionStat to monitor if its mark-delete position is advanced or not.
xiaoyaolanchong · Jan 6, 2021 · 2fd878a · 2fd878a
1 parent c12765a
commit 2fd878a
Show file tree

Hide file tree

Showing 8 changed files with 63 additions and 9 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -237,7 +237,7 @@ private CompletableFuture<NamespaceBundle> getFullBundleAsync(NamespaceName fqnn
     /**
      * Return the URL of the broker who's owning a particular service unit in asynchronous way.
      *
-     * If the service unit is not owned, return a CompletableFuture with empty optional
+     * If the service unit is not owned, return a CompletableFuture with empty optional.
      */
     public CompletableFuture<Optional<URL>> getWebServiceUrlAsync(ServiceUnitId suName, LookupOptions options)
             throws Exception {

diff --git a/...main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java b/...main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.broker.service.persistent;
 
+import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.LongAdder;
@@ -134,7 +135,11 @@ public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
     public void findEntryComplete(Position position, Object ctx) {
         if (position != null) {
             log.info("[{}][{}] Expiring all messages until position {}", topicName, subName, position);
+            Position prevMarkDeletePos = cursor.getMarkDeletedPosition();
             cursor.asyncMarkDelete(position, markDeleteCallback, cursor.getNumberOfEntriesInBacklog(false));
+            if (!Objects.equals(cursor.getMarkDeletedPosition(), prevMarkDeletePos) && subscription != null) {
+                subscription.updateLastMarkDeleteAdvancedTimestamp();
+            }
         } else {
             if (log.isDebugEnabled()) {
                 log.debug("[{}][{}] No messages to expire", topicName, subName);

diff --git a/...ker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/...ker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -91,6 +91,7 @@ public class PersistentSubscription implements Subscription {
 
     private long lastExpireTimestamp = 0L;
     private long lastConsumedFlowTimestamp = 0L;
+    private long lastMarkDeleteAdvancedTimestamp = 0L;
 
     // for connected subscriptions, message expiry will be checked if the backlog is greater than this threshold
     private static final int MINIMUM_BACKLOG_FOR_EXPIRY_CHECK = 1000;
@@ -137,6 +138,11 @@ public PersistentSubscription(PersistentTopic topic, String subscriptionName, Ma
         IS_FENCED_UPDATER.set(this, FALSE);
     }
 
+    public void updateLastMarkDeleteAdvancedTimestamp() {
+        this.lastMarkDeleteAdvancedTimestamp =
+            Math.max(this.lastMarkDeleteAdvancedTimestamp, System.currentTimeMillis());
+    }
+
     @Override
     public BrokerInterceptor interceptor() {
         return topic.getBrokerService().getInterceptor();
@@ -333,6 +339,8 @@ public void acknowledgeMessage(List<Position> positions, AckType ackType, Map<St
         }
 
         if (!cursor.getMarkDeletedPosition().equals(previousMarkDeletePosition)) {
+            this.updateLastMarkDeleteAdvancedTimestamp();
+
             // Mark delete position advance
             ReplicatedSubscriptionSnapshotCache snapshotCache  = this.replicatedSubscriptionSnapshotCache;
             if (snapshotCache != null) {
@@ -899,6 +907,7 @@ public SubscriptionStats getStats(Boolean getPreciseBacklog) {
         SubscriptionStats subStats = new SubscriptionStats();
         subStats.lastExpireTimestamp = lastExpireTimestamp;
         subStats.lastConsumedFlowTimestamp = lastConsumedFlowTimestamp;
+        subStats.lastMarkDeleteAdvancedTimestamp = lastMarkDeleteAdvancedTimestamp;
         Dispatcher dispatcher = this.dispatcher;
         if (dispatcher != null) {
             Map<String, List<String>> consumerKeyHashRanges = getType() == SubType.Key_Shared

diff --git a/.../src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java b/.../src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java
@@ -46,6 +46,14 @@ public class AggregatedSubscriptionStats {
 
     long lastExpireTimestamp;
 
+    long lastConsumedFlowTimestamp;
+
+    long lastConsumedTimestamp;
+
+    long lastAckedTimestamp;
+
+    long lastMarkDeleteAdvancedTimestamp;
+
     double msgRateExpired;
 
     long totalMsgExpired;

diff --git a/...ker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java b/...ker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
@@ -140,10 +140,14 @@ private static void getTopicStats(Topic topic, TopicStats stats, boolean include
                     .computeIfAbsent(subName, k -> new AggregatedSubscriptionStats());
             subsStats.msgBacklog = subscriptionStats.msgBacklog;
             subsStats.msgDelayed = subscriptionStats.msgDelayed;
-            subsStats.lastExpireTimestamp = subscriptionStats.lastExpireTimestamp;
             subsStats.msgRateExpired = subscriptionStats.msgRateExpired;
             subsStats.totalMsgExpired = subscriptionStats.totalMsgExpired;
             subsStats.msgBacklogNoDelayed = subsStats.msgBacklog - subsStats.msgDelayed;
+            subsStats.lastExpireTimestamp = subscriptionStats.lastExpireTimestamp;
+            subsStats.lastAckedTimestamp = subscriptionStats.lastAckedTimestamp;
+            subsStats.lastConsumedFlowTimestamp = subscriptionStats.lastConsumedFlowTimestamp;
+            subsStats.lastConsumedTimestamp = subscriptionStats.lastConsumedTimestamp;
+            subsStats.lastMarkDeleteAdvancedTimestamp = subscriptionStats.lastMarkDeleteAdvancedTimestamp;
             subscriptionStats.consumers.forEach(cStats -> {
                 stats.consumersCount++;
                 subsStats.unackedMessages += cStats.unackedMessages;

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java
@@ -186,6 +186,14 @@ static void printTopicStats(SimpleTextOutputStream stream, String cluster, Strin
                     subsStats.msgOutCounter);
             metric(stream, cluster, namespace, topic, n, "pulsar_subscription_last_expire_timestamp",
                     subsStats.lastExpireTimestamp);
+            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_last_acked_timestamp",
+                subsStats.lastAckedTimestamp);
+            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_last_consumed_flow_timestamp",
+                subsStats.lastConsumedFlowTimestamp);
+            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_last_consumed_timestamp",
+                subsStats.lastConsumedTimestamp);
+            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_last_mark_delete_advanced_timestamp",
+                subsStats.lastMarkDeleteAdvancedTimestamp);
             metric(stream, cluster, namespace, topic, n, "pulsar_subscription_msg_rate_expired",
                     subsStats.msgRateExpired);
             metric(stream, cluster, namespace, topic, n, "pulsar_subscription_total_msg_expired",

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java
@@ -2097,17 +2097,29 @@ public void testPersistentTopicsExpireMessages() throws Exception {
         Thread.sleep(1000); // wait for 1 seconds to execute expire message as it is async
 
         topicStats = admin.topics().getStats("persistent://prop-xyz/ns1/ds2");
-        assertEquals(topicStats.subscriptions.get("my-sub1").msgBacklog, 0);
-        assertEquals(topicStats.subscriptions.get("my-sub2").msgBacklog, 10);
-        assertEquals(topicStats.subscriptions.get("my-sub3").msgBacklog, 10);
+        SubscriptionStats subStats1 = topicStats.subscriptions.get("my-sub1");
+        assertEquals(subStats1.msgBacklog, 0);
+        assertTrue(subStats1.lastMarkDeleteAdvancedTimestamp > 0L);
+        SubscriptionStats subStats2 = topicStats.subscriptions.get("my-sub2");
+        assertEquals(subStats2.msgBacklog, 10);
+        assertEquals(subStats2.lastMarkDeleteAdvancedTimestamp, 0L);
+        SubscriptionStats subStats3 = topicStats.subscriptions.get("my-sub3");
+        assertEquals(subStats3.msgBacklog, 10);
+        assertEquals(subStats3.lastMarkDeleteAdvancedTimestamp, 0L);
 
         admin.topics().expireMessagesForAllSubscriptions("persistent://prop-xyz/ns1/ds2", 1);
         Thread.sleep(1000); // wait for 1 seconds to execute expire message as it is async
 
         topicStats = admin.topics().getStats("persistent://prop-xyz/ns1/ds2");
-        assertEquals(topicStats.subscriptions.get("my-sub1").msgBacklog, 0);
-        assertEquals(topicStats.subscriptions.get("my-sub2").msgBacklog, 0);
-        assertEquals(topicStats.subscriptions.get("my-sub3").msgBacklog, 0);
+        SubscriptionStats newSubStats1 = topicStats.subscriptions.get("my-sub1");
+        assertEquals(newSubStats1.msgBacklog, 0);
+        assertEquals(newSubStats1.lastMarkDeleteAdvancedTimestamp, subStats1.lastMarkDeleteAdvancedTimestamp);
+        SubscriptionStats newSubStats2 = topicStats.subscriptions.get("my-sub2");
+        assertEquals(newSubStats2.msgBacklog, 0);
+        assertTrue(newSubStats2.lastMarkDeleteAdvancedTimestamp > subStats2.lastMarkDeleteAdvancedTimestamp);
+        SubscriptionStats newSubStats3 = topicStats.subscriptions.get("my-sub3");
+        assertEquals(newSubStats3.msgBacklog, 0);
+        assertTrue(newSubStats3.lastMarkDeleteAdvancedTimestamp > subStats3.lastMarkDeleteAdvancedTimestamp);
 
         consumer1.close();
         consumer2.close();
@@ -2547,15 +2559,19 @@ public void testBacklogSizeShouldBeZeroWhenConsumerAckedAllMessages() throws Exc
             producer.send(new byte[1024 * i * 5]);
         }
 
+        TopicStats topicStats = admin.topics().getStats(topic);
+        assertEquals(topicStats.subscriptions.get("sub-1").lastMarkDeleteAdvancedTimestamp, 0L);
+
         for (int i = 0; i < messages; i++) {
             consumer.acknowledgeCumulative(consumer.receive());
         }
 
         // Wait ack send
         Thread.sleep(1000);
 
-        TopicStats topicStats = admin.topics().getStats(topic);
+        topicStats = admin.topics().getStats(topic);
         assertEquals(topicStats.backlogSize, 0);
+        assertTrue(topicStats.subscriptions.get("sub-1").lastMarkDeleteAdvancedTimestamp > 0L);
     }
 
     @Test

diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/SubscriptionStats.java
@@ -86,6 +86,9 @@ public class SubscriptionStats {
     /** Last acked message timestamp. */
     public long lastAckedTimestamp;
 
+    /** Last MarkDelete position advanced timesetamp. */
+    public long lastMarkDeleteAdvancedTimestamp;
+
     /** List of connected consumers on this subscription w/ their stats. */
     public List<ConsumerStats> consumers;
 
@@ -121,6 +124,7 @@ public void reset() {
         msgRateExpired = 0;
         totalMsgExpired = 0;
         lastExpireTimestamp = 0L;
+        lastMarkDeleteAdvancedTimestamp = 0L;
         consumers.clear();
         consumersAfterMarkDeletePosition.clear();
         nonContiguousDeletedMessagesRanges = 0;