From cf72151f1ef0a90a44020375e96f99892afb3b44 Mon Sep 17 00:00:00 2001 From: Qiang Huang Date: Thu, 29 Sep 2022 12:55:20 +0800 Subject: [PATCH] [improve][broker]add active status into cursor stats --- .../org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java | 1 + .../apache/pulsar/broker/service/persistent/PersistentTopic.java | 1 + .../apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java | 1 + .../pulsar/common/policies/data/ManagedLedgerInternalStats.java | 1 + 4 files changed, 4 insertions(+) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java index 254ee767bc7fc..53bccfadbab22 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java @@ -4116,6 +4116,7 @@ public CompletableFuture getManagedLedgerInternalSta cs.individuallyDeletedMessages = cursor.getIndividuallyDeletedMessages(); cs.lastLedgerSwitchTimestamp = DateFormatter.format(cursor.getLastLedgerSwitchTimestamp()); cs.state = cursor.getState(); + cs.active = cursor.isActive(); cs.numberOfEntriesSinceFirstNotAckedMessage = cursor.getNumberOfEntriesSinceFirstNotAckedMessage(); cs.totalNonContiguousDeletedMessagesRange = cursor.getTotalNonContiguousDeletedMessagesRange(); cs.properties = cursor.getProperties(); 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 fdcaaf2ffbdb3..fb1c521bb00b0 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 @@ -2210,6 +2210,7 @@ public CompletableFuture getInternalStats(boolean cs.individuallyDeletedMessages = cursor.getIndividuallyDeletedMessages(); cs.lastLedgerSwitchTimestamp = DateFormatter.format(cursor.getLastLedgerSwitchTimestamp()); cs.state = cursor.getState(); + cs.active = cursor.isActive(); cs.numberOfEntriesSinceFirstNotAckedMessage = cursor.getNumberOfEntriesSinceFirstNotAckedMessage(); cs.totalNonContiguousDeletedMessagesRange = cursor.getTotalNonContiguousDeletedMessagesRange(); cs.properties = cursor.getProperties(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java index 93d1132d812d0..af72a2564bb0c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java @@ -135,6 +135,7 @@ public void testTopicInternalStats() throws Exception { && (cursor.totalNonContiguousDeletedMessagesRange) < numberOfMsgs / 2); assertFalse(cursor.subscriptionHavePendingRead); assertFalse(cursor.subscriptionHavePendingReplayRead); + assertTrue(cursor.active); producer.close(); consumer.close(); log.info("-- Exiting {} test --", methodName); diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ManagedLedgerInternalStats.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ManagedLedgerInternalStats.java index 35bef4cda33c6..ee67f826ee678 100644 --- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ManagedLedgerInternalStats.java +++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ManagedLedgerInternalStats.java @@ -94,6 +94,7 @@ public static class CursorStats { public String individuallyDeletedMessages; public String lastLedgerSwitchTimestamp; public String state; + public boolean active; public long numberOfEntriesSinceFirstNotAckedMessage; public int totalNonContiguousDeletedMessagesRange; public boolean subscriptionHavePendingRead;