Add stats 'subscriptionIsolationLevel' in SubscriptionStatsImpl

apache · Oct 7, 2023 · a4c4db2 · a4c4db2
1 parent 05a5c46
commit a4c4db2
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 4 deletions.
diff --git a/...c/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java b/...c/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
@@ -472,6 +472,7 @@ public NonPersistentSubscriptionStatsImpl getStats() {
 
         subStats.type = getTypeString();
         subStats.msgDropRate = dispatcher.getMessageDropRate().getValueRate();
+        subStats.subscriptionIsolationLevel = this.isolationLevel.toString();
 
         KeySharedMode keySharedMode = this.keySharedMode;
         if (getType() == SubType.Key_Shared && keySharedMode != null) {

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
@@ -128,7 +128,7 @@ public class PersistentSubscription extends AbstractSubscription implements Subs
     private volatile ReplicatedSubscriptionSnapshotCache replicatedSubscriptionSnapshotCache;
     private final PendingAckHandle pendingAckHandle;
     private volatile Map<String, String> subscriptionProperties;
-    private final IsolationLevel subscriptionIsolationLevel;
+    private final IsolationLevel isolationLevel;
     private volatile CompletableFuture<Void> fenceFuture;
 
     static Map<String, Long> getBaseCursorProperties(boolean isReplicated) {
@@ -162,7 +162,7 @@ public PersistentSubscription(PersistentTopic topic, String subscriptionName, Ma
             this.pendingAckHandle = new PendingAckHandleDisabled();
         }
         IS_FENCED_UPDATER.set(this, FALSE);
-        this.subscriptionIsolationLevel = fetchIsolationLevelFromProperties(subscriptionProperties);
+        this.isolationLevel = fetchIsolationLevelFromProperties(subscriptionProperties);
     }
 
     public void updateLastMarkDeleteAdvancedTimestamp() {
@@ -511,7 +511,7 @@ public String getTypeString() {
 
     @Override
     public IsolationLevel getIsolationLevel() {
-        return subscriptionIsolationLevel;
+        return isolationLevel;
     }
 
     @Override
@@ -1195,6 +1195,7 @@ public SubscriptionStatsImpl getStats(Boolean getPreciseBacklog, boolean subscri
         subStats.isReplicated = isReplicated();
         subStats.subscriptionProperties = subscriptionProperties;
         subStats.isDurable = cursor.isDurable();
+        subStats.subscriptionIsolationLevel = this.isolationLevel.toString();
         if (getType() == SubType.Key_Shared && dispatcher instanceof PersistentStickyKeyDispatcherMultipleConsumers) {
             PersistentStickyKeyDispatcherMultipleConsumers keySharedDispatcher =
                     (PersistentStickyKeyDispatcherMultipleConsumers) dispatcher;

diff --git a/...mon/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java b/...mon/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java
@@ -138,7 +138,9 @@ public class SubscriptionStatsImpl implements SubscriptionStats {
     @JsonIgnore
     public Map<String, TopicMetricBean> bucketDelayedIndexStats;
 
-    /** SubscriptionProperties (key/value strings) associated with this subscribe. */
+    /**
+     * SubscriptionProperties (key/value strings) associated with this subscribe.
+     */
     public Map<String, String> subscriptionProperties;
 
     public long filterProcessedMsgCount;
@@ -149,6 +151,11 @@ public class SubscriptionStatsImpl implements SubscriptionStats {
 
     public long filterRescheduledMsgCount;
 
+    /**
+     * Whether the subscription's transaction isolationLevel is READ_COMMITTED or READ_UNCOMMITTED.
+     */
+    public String subscriptionIsolationLevel;
+
     public SubscriptionStatsImpl() {
         this.consumers = new ArrayList<>();
         this.consumersAfterMarkDeletePosition = new LinkedHashMap<>();