You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2025/07/07 12:00:48 UTC
(pulsar) 04/05: [improve] [broker] Part 2 of PIP-370: add metrics "pulsar_replication_disconnected_count" (#23213)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 98561a06abadccc67117673ea3f62aee587766b9
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Mon Aug 26 10:05:21 2024 +0800
[improve] [broker] Part 2 of PIP-370: add metrics "pulsar_replication_disconnected_count" (#23213)
(cherry picked from commit 09a16c26974408de270bcaaf6162b0e2a9a6d203)
---
pip/pip-370.md | 108 ++++++++++++++++++
.../stats/prometheus/AggregatedNamespaceStats.java | 1 +
.../prometheus/AggregatedReplicationStats.java | 3 +
.../stats/prometheus/NamespaceStatsAggregator.java | 8 +-
.../pulsar/broker/stats/prometheus/TopicStats.java | 2 +
.../broker/service/OneWayReplicatorTest.java | 121 +++++++++++++++++++++
.../service/OneWayReplicatorUsingGlobalZKTest.java | 6 +
.../prometheus/AggregatedNamespaceStatsTest.java | 2 +
8 files changed, 250 insertions(+), 1 deletion(-)
diff --git a/pip/pip-370.md b/pip/pip-370.md
new file mode 100644
index 00000000000..a29d5561432
--- /dev/null
+++ b/pip/pip-370.md
@@ -0,0 +1,108 @@
+# PIP-370: configurable remote topic creation in geo-replication
+
+# Background knowledge
+
+**The current topic creation behavior when enabling Geo-Replication**
+Users using Geo-Replication backup data across multiple clusters, as well as Admin APIs related to Geo-Replication and internal replicators of brokers, will trigger topics of auto-creation between clusters.
+- For partitioned topics.
+ - After enabling namespace-level Geo-Replication: the broker will create topics on the remote cluster automatically when calling `pulsar-admin topics create-partitioned-topic`. It does not depend on enabling `allowAutoTopicCreation`.
+ - When enabling topic-level Geo-Replication on a partitioned topic: the broker will create topics on the remote cluster automatically. It does not depend on enabling `allowAutoTopicCreation`.
+ - When calling `pulsar-admin topics update-partitioned-topic -p {partitions}`, the broker will also update partitions on the remote cluster automatically.
+- For non-partitioned topics and partitions of partitioned topics.
+ - The internal Geo-Replicator will trigger topics auto-creation for remote clusters. **(Highlight)** It depends on enabling `allowAutoTopicCreation`. In fact, this behavior is not related to Geo-Replication, it is the behavior of the internal producer of Geo-Replicator,
+
+# Motivation
+
+In the following scenarios, automatic topic creation across clusters is problematic due to race conditions during deployments, and there is no choice that prevents pulsar resource creation affects each other between clusters.
+
+- Users want to maintain pulsar resources manually.
+- Users pulsar resources using `GitOps CD` automated deployment, for which
+ - Clusters are deployed simultaneously without user intervention.
+ - Each cluster is precisely configured from git repo config variables - including the list of all tenants/namespaces/topics to be created in each cluster.
+ - Clusters are configured to be exact clones of each other in terms of pulsar resources.
+
+**Passed solution**: disable `allowAutoTopicCreation`, the APIs `pulsar-admin topics create-partitioned-topic` still create topics on the remote cluster when enabled namespace level replication, the API `enable topic-level replication` still create topics, And the internal replicator will keep printing error logs due to a not found error.
+
+# Goals
+
+- **Phase 1**: Introduce a flag to disable the replicators to automatically trigger topic creation.
+- **Phase 2**: Move all topic creation/expand-partitions behaviors related to Replication to the internal Replicator, pulsar admin API that relates to pulsar topics management does not care about replication anymore.
+ - Move the topic creation operations from `pulsar-admin topics create-partitioned-topic` and `pulsar-admin topics set-replication-clusters` to the component Replicator in the broker internal.
+ - (The same as before)When calling `pulsar-admin topics update-partitioned-topic -p {partitions}`, the broker will also update partitions on the remote cluster automatically.
+
+Note: the proposal will only focus on phase one, and the detailed design for phase two with come up with another proposal.
+
+# Detailed Design
+
+## Configuration
+
+**broker.conf**
+```properties
+# Whether the internal replication of the local cluster will trigger topic auto-creation on the remote cluster.
+# 1. After enabling namespace-level Geo-Replication: whether the local broker will create topics on the remote cluster automatically when calling `pulsar-admin topics create-partitioned-topic`.
+# 2. When enabling topic-level Geo-Replication on a partitioned topic: whether the local broker will create topics on the remote cluster.
+# 3. Whether the internal Geo-Replicator in the local cluster will trigger non-persistent topic auto-creation for remote clusters.
+# It is not a dynamic config, the default value is "true" to preserve backward-compatible behavior.
+createTopicToRemoteClusterForReplication=true
+```
+
+## Design & Implementation Details
+
+### Phase 1: Introduce a flag to disable the replicators to automatically trigger topic creation.
+- If `createTopicToRemoteClusterForReplication` is set to `false`.
+ 1. After enabling namespace-level Geo-Replication: the broker will not create topics on the remote cluster automatically when calling `pulsar-admin topics create-partitioned-topic`.
+ 2. When enabling topic-level Geo-Replication on a partitioned topic: broker will not create topics on the remote cluster automatically.
+ 3. The internal Geo-Replicator will not trigger topic auto-creation for remote clusters, it just keeps retrying to check if the topic exists on the remote cluster, once the topic is created, the replicator starts.
+ 4. It does not change the behavior of creating subscriptions after enabling `enableReplicatedSubscriptions`, the subscription will also be created on the remote cluster after users enable. `enableReplicatedSubscriptions`.
+ 5. The config `allowAutoTopicCreation` still works for the local cluster as before, it will not be affected by the new config `createTopicToRemoteClusterForReplication`.
+- If `createTopicToRemoteClusterForReplication` is set to `true`.
+ a. All components work as before, see details: `Motivation -> The current topic creation behavior when enabling Geo-Replication`
+
+### Phase 2: The replicator will check remote topics' partitioned metadata and update partitions in the remote cluster to the same as the current cluster if needed.
+- If `createTopicToRemoteClusterForReplication` is set to `false`.
+ - The behavior is the same as Phase 1.
+- If `createTopicToRemoteClusterForReplication` is set to `true`.
+ - Pulsar admin API that relates to pulsar topics management does not care about replication anymore.
+ - When a replicator for a topic partition starts, it checks the partitioned metadata in the remote cluster first and updates partitions in the remote cluster to the same as the current cluster if needed. Seem the example as follows:
+
+| `partitions` of local cluster | `partitions` of remote cluster | After `PIP-370 Phase 2` | Before `PIP-370 Phase 2` |
+| --- | --- | --- | --- |
+| `2` | no topic exists | create a partitioned topic with `2` partitions in the remote cluster | the replicator will only trigger partition creation (`{topic}-partition-0` and `{topic}-partition-1`), and will not care about partitioned metadata. |
+| `2` | `1`| **In dispute:** The replicator copies messages from `partition-0` to the remote cluster, does not copy any data for `partition-1` and just prints error logs in the background. | the replicator will only trigger partition creation (`{topic}-partition-0` and `{topic}-partition-1`), and the partitioned metadata in the remote cluster is still `1` |
+| `2` | `2` | modifies nothing. | The same as "After `PIP-370 Phase 2`" |
+| `2` | `>2` | **In dispute:** modifies nothing, the messages will be copied to the same partition in the remote cluster, and no message will be copied to the partition who is larger than `2` in the remote cluster | The same as "After `PIP-370 Phase 2`" |
+| `2` | `0`(non-partitioned topic) | **In dispute:** The replicator does not copy any data and just prints error logs in the background. | the replicator will only trigger partition creation (`{topic}-partition-0` and `{topic}-partition-1`), then users will get `3` non-partitioned topics: `[{tp}, {topic}-partition-0, {topic}-partition-1`. |
+| `0`(non-partitioned topic) | `0`(non-partitioned topic) | Copy data normally | It is the same as before `PIP-370`. |
+| `0`(non-partitioned topic) | no topic exists | create a non-partitioned topic in the remote cluster. | It is the same as before `PIP-370`. |
+| `0`(non-partitioned topic) | `>=1` | **In dispute:** The replicator does not copy any data and just prints error logs in the background. | The replicator will only trigger a non-partitioned topic's creation, then users will get `1` non-partitioned topic and `1` partitioned topic. |
+
+## Metrics
+
+<!--
+For each metric provide:
+* Full name
+* Description
+* Attributes (labels)
+* Unit
+-->
+| Name | Description | Attributes | Units|
+| --- |---------------------------------------------------------------------------------------------|---------------------------| --- |
+| `pulsar_replication_disconnected_count` | Counter. The number of replicators. | cluster, namespace, topic | - |
+
+
+# Monitoring
+- If `pulsar_broker_replication_disconnected_count` keeps larger than `0` for a period of time, it means some replicators do not work, we should push an alert out.
+
+# Backward & Forward Compatibility
+
+## Regarding to Phase-1
+This PIP guarantees full compatibility with default settings(the default value of `createTopicToRemoteClusterForReplication` is `true`). If you want to cherry-pick PIP-370 for another branch in the future, you need to cherry-pick PIP-344 as well. Because the behavior of disables `createTopicToRemoteClusterForReplication` depends on the API `PulsarClient.getPartitionsForTopic(String topic, boolean metadataAutoCreationEnabled)`, which was introduced by [PIP-344](https://github.com/apache/p [...]
+
+## Regarding to Phase-2
+The two scenarios are as follows, the replication will not work as before, which will lead backlog increase, please take care of checking your clusters before upgrading.
+- `local_cluster.topic.partitions = 2` and `remote_cluster.topic.partitions = 0(non-partitioned topic)`: see detail in the section `Design & Implementation Details -> Phase-2`.
+- `local_cluster.topic.partitions = 0(non-partitioned topic)` and and `remote_cluster.topic.partitions >= 1`: see detail in the section `Design & Implementation Details -> Phase-2`.
+
+# Links
+* Mailing List discussion thread: https://lists.apache.org/thread/9fx354cqcy3412w1nx8kwdf9h141omdg
+* Mailing List voting thread: https://lists.apache.org/thread/vph22st5td1rdh1gd68gkrnp9doo6ct2
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java
index 91648c27627..aaaea7b493e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java
@@ -134,6 +134,7 @@ public class AggregatedNamespaceStats {
replStats.replicationBacklog += as.replicationBacklog;
replStats.msgRateExpired += as.msgRateExpired;
replStats.connectedCount += as.connectedCount;
+ replStats.disconnectedCount += as.disconnectedCount;
replStats.replicationDelayInSeconds += as.replicationDelayInSeconds;
});
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedReplicationStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedReplicationStats.java
index 78f33f874e9..82668de6c35 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedReplicationStats.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedReplicationStats.java
@@ -41,6 +41,9 @@ public class AggregatedReplicationStats {
/** The count of replication-subscriber up and running to replicate to remote cluster. */
public long connectedCount;
+ /** The count of replication-subscriber that failed to start to replicate to remote cluster. */
+ public long disconnectedCount;
+
/** Time in seconds from the time a message was produced to the time when it is about to be replicated. */
public long replicationDelayInSeconds;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
index 440666c4ec9..cc12ee175d8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
@@ -304,7 +304,11 @@ public class NamespaceStatsAggregator {
aggReplStats.msgThroughputOut += replStats.msgThroughputOut;
aggReplStats.replicationBacklog += replStats.replicationBacklog;
aggReplStats.msgRateExpired += replStats.msgRateExpired;
- aggReplStats.connectedCount += replStats.connected ? 1 : 0;
+ if (replStats.connected) {
+ aggReplStats.connectedCount += 1;
+ } else {
+ aggReplStats.disconnectedCount += 1;
+ }
aggReplStats.replicationDelayInSeconds += replStats.replicationDelayInSeconds;
});
@@ -513,6 +517,8 @@ public class NamespaceStatsAggregator {
replStats -> replStats.replicationBacklog, cluster, namespace);
writeReplicationStat(stream, "pulsar_replication_connected_count", stats,
replStats -> replStats.connectedCount, cluster, namespace);
+ writeReplicationStat(stream, "pulsar_replication_disconnected_count", stats,
+ replStats -> replStats.disconnectedCount, cluster, namespace);
writeReplicationStat(stream, "pulsar_replication_rate_expired", stats,
replStats -> replStats.msgRateExpired, cluster, namespace);
writeReplicationStat(stream, "pulsar_replication_delay_in_seconds", stats,
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
index 86d478593d9..524d47e7c1b 100644
--- 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
@@ -410,6 +410,8 @@ class TopicStats {
cluster, namespace, topic, remoteCluster, splitTopicAndPartitionIndexLabel);
writeMetric(stream, "pulsar_replication_connected_count", replStats.connectedCount,
cluster, namespace, topic, remoteCluster, splitTopicAndPartitionIndexLabel);
+ writeMetric(stream, "pulsar_replication_disconnected_count", replStats.disconnectedCount,
+ cluster, namespace, topic, remoteCluster, splitTopicAndPartitionIndexLabel);
writeMetric(stream, "pulsar_replication_rate_expired", replStats.msgRateExpired,
cluster, namespace, topic, remoteCluster, splitTopicAndPartitionIndexLabel);
writeMetric(stream, "pulsar_replication_delay_in_seconds", replStats.replicationDelayInSeconds,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
index b2c4643e00c..40f0914b94f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorTest.java
@@ -32,6 +32,7 @@ import static org.testng.Assert.fail;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
+import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import io.netty.util.concurrent.FastThreadLocalThread;
import java.lang.reflect.Field;
@@ -74,6 +75,7 @@ import org.apache.pulsar.broker.resources.ClusterResources;
import org.apache.pulsar.broker.service.persistent.GeoPersistentReplicator;
import org.apache.pulsar.broker.service.persistent.PersistentReplicator;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
@@ -103,6 +105,8 @@ import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.awaitility.Awaitility;
import org.awaitility.reflect.WhiteboxImpl;
+import org.glassfish.jersey.client.JerseyClient;
+import org.glassfish.jersey.client.JerseyClientBuilder;
import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
@@ -1395,4 +1399,121 @@ public class OneWayReplicatorTest extends OneWayReplicatorTestBase {
assertEquals(valueLocal, publishRateAddLocal2);
});
}
+
+ @Test
+ public void testReplicationCountMetrics() throws Exception {
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + nonReplicatedNamespace + "/tp_");
+ // 1.Create topic, does not enable replication now.
+ admin1.topics().createNonPartitionedTopic(topicName);
+ PersistentTopic persistentTopic =
+ (PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get();
+
+ // We inject an error to make the internal producer fail to connect.
+ final AtomicInteger createProducerCounter = new AtomicInteger();
+ final AtomicBoolean failedCreateProducer = new AtomicBoolean(true);
+ Runnable taskToClearInjection = injectMockReplicatorProducerBuilder((producerCnf, originalProducer) -> {
+ if (topicName.equals(producerCnf.getTopicName())) {
+ // There is a switch to determine create producer successfully or not.
+ if (failedCreateProducer.get()) {
+ log.info("Retry create replicator.producer count: {}", createProducerCounter);
+ // Release producer and fail callback.
+ originalProducer.closeAsync();
+ throw new RuntimeException("mock error");
+ }
+ return originalProducer;
+ }
+ return originalProducer;
+ });
+
+ // 2.Enable replication.
+ admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1, cluster2));
+
+ // Verify: metrics.
+ // Cluster level:
+ // - pulsar_replication_connected_count
+ // - pulsar_replication_disconnected_count
+ // Namespace level:
+ // - pulsar_replication_connected_count
+ // - pulsar_replication_disconnected_count
+ // Topic level:
+ // - pulsar_replication_connected_count
+ // - pulsar_replication_disconnected_count
+ JerseyClient httpClient = JerseyClientBuilder.createClient();
+ Awaitility.await().untilAsserted(() -> {
+ int topicConnected = 0;
+ int topicDisconnected = 0;
+
+ String response = httpClient.target(pulsar1.getWebServiceAddress()).path("/metrics/")
+ .request().get(String.class);
+ Multimap<String, PrometheusMetricsClient.Metric> metricMap = PrometheusMetricsClient.parseMetrics(response);
+ if (!metricMap.containsKey("pulsar_replication_disconnected_count")) {
+ fail("Expected 1 disconnected replicator.");
+ }
+ for (PrometheusMetricsClient.Metric metric : metricMap.get("pulsar_replication_connected_count")) {
+ if (cluster1.equals(metric.tags.get("cluster"))
+ && nonReplicatedNamespace.equals(metric.tags.get("namespace"))
+ && topicName.equals(metric.tags.get("topic"))) {
+ topicConnected += Double.valueOf(metric.value).intValue();
+ }
+ }
+ for (PrometheusMetricsClient.Metric metric : metricMap.get("pulsar_replication_disconnected_count")) {
+ if (cluster1.equals(metric.tags.get("cluster"))
+ && nonReplicatedNamespace.equals(metric.tags.get("namespace"))
+ && topicName.equals(metric.tags.get("topic"))) {
+ topicDisconnected += Double.valueOf(metric.value).intValue();
+ }
+ }
+ log.info("{}, {},", topicConnected, topicDisconnected);
+ assertEquals(topicConnected, 0);
+ assertEquals(topicDisconnected, 1);
+ });
+
+ // Let replicator connect successfully.
+ failedCreateProducer.set(false);
+ // Verify: metrics.
+ // Cluster level:
+ // - pulsar_replication_connected_count
+ // - pulsar_replication_disconnected_count
+ // Namespace level:
+ // - pulsar_replication_connected_count
+ // - pulsar_replication_disconnected_count
+ // Topic level:
+ // - pulsar_replication_connected_count
+ // - pulsar_replication_disconnected_count
+ Awaitility.await().atMost(Duration.ofSeconds(130)).untilAsserted(() -> {
+ int topicConnected = 0;
+ int topicDisconnected = 0;
+
+ String response = httpClient.target(pulsar1.getWebServiceAddress()).path("/metrics/")
+ .request().get(String.class);
+ Multimap<String, PrometheusMetricsClient.Metric> metricMap = PrometheusMetricsClient.parseMetrics(response);
+ if (!metricMap.containsKey("pulsar_replication_disconnected_count")) {
+ fail("Expected 1 disconnected replicator.");
+ }
+ for (PrometheusMetricsClient.Metric metric : metricMap.get("pulsar_replication_connected_count")) {
+ if (cluster1.equals(metric.tags.get("cluster"))
+ && nonReplicatedNamespace.equals(metric.tags.get("namespace"))
+ && topicName.equals(metric.tags.get("topic"))) {
+ topicConnected += Double.valueOf(metric.value).intValue();
+ }
+ }
+ for (PrometheusMetricsClient.Metric metric : metricMap.get("pulsar_replication_disconnected_count")) {
+ if (cluster1.equals(metric.tags.get("cluster"))
+ && nonReplicatedNamespace.equals(metric.tags.get("namespace"))
+ && topicName.equals(metric.tags.get("topic"))) {
+ topicDisconnected += Double.valueOf(metric.value).intValue();
+ }
+ }
+ log.info("{}, {}", topicConnected, topicDisconnected);
+ assertEquals(topicConnected, 1);
+ assertEquals(topicDisconnected, 0);
+ });
+
+ // cleanup.
+ taskToClearInjection.run();
+ admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1));
+ waitReplicatorStopped(topicName);
+ admin1.topics().delete(topicName, false);
+ admin2.topics().delete(topicName, false);
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
index ce7e5578ff1..da704379141 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorUsingGlobalZKTest.java
@@ -231,4 +231,10 @@ public class OneWayReplicatorUsingGlobalZKTest extends OneWayReplicatorTest {
public void testTopicPoliciesReplicationRule() throws Exception {
super.testTopicPoliciesReplicationRule();
}
+
+ @Test(enabled = false)
+ @Override
+ public void testReplicationCountMetrics() throws Exception {
+ super.testReplicationCountMetrics();
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStatsTest.java
index 0e12d75f74f..11358eb1e2c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStatsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStatsTest.java
@@ -101,6 +101,7 @@ public class AggregatedNamespaceStatsTest {
replStats2.msgThroughputOut = 1536.0;
replStats2.replicationBacklog = 99;
replStats2.connectedCount = 1;
+ replStats2.disconnectedCount = 2;
replStats2.msgRateExpired = 3.0;
replStats2.replicationDelayInSeconds = 20;
topicStats2.replicationStats.put(namespace, replStats2);
@@ -148,6 +149,7 @@ public class AggregatedNamespaceStatsTest {
assertEquals(nsReplStats.msgThroughputOut, 1792.0);
assertEquals(nsReplStats.replicationBacklog, 100);
assertEquals(nsReplStats.connectedCount, 1);
+ assertEquals(nsReplStats.disconnectedCount, 2);
assertEquals(nsReplStats.msgRateExpired, 6.0);
assertEquals(nsReplStats.replicationDelayInSeconds, 40);