public static void generate(PulsarService pulsar, boolean includeTopicMetrics, boolean includeConsumerMetrics,
                                boolean includeProducerMetrics, boolean splitTopicAndPartitionIndexLabel,
                                PrometheusMetricStreams stream) {
        String cluster = pulsar.getConfiguration().getClusterName();
        AggregatedBrokerStats brokerStats = localBrokerStats.get();
        brokerStats.reset();
        AggregatedNamespaceStats namespaceStats = localNamespaceStats.get();
        TopicStats topicStats = localTopicStats.get();
        Optional<CompactorMXBean> compactorMXBean = getCompactorMXBean(pulsar);
        LongAdder topicsCount = new LongAdder();
        Map<String, Long> localNamespaceTopicCount = new HashMap<>();
        pulsar.getBrokerService().getMultiLayerTopicMap().forEach((namespace, bundlesMap) -> {
            namespaceStats.reset();
            topicsCount.reset();

            bundlesMap.forEach((bundle, topicsMap) -> topicsMap.forEach((name, topic) -> {
                getTopicStats(topic, topicStats, includeConsumerMetrics, includeProducerMetrics,
                        pulsar.getConfiguration().isExposePreciseBacklogInPrometheus(),
                        pulsar.getConfiguration().isExposeSubscriptionBacklogSizeInPrometheus(),
                        compactorMXBean
                );

                brokerStats.updateStats(topicStats);

                if (includeTopicMetrics) {
                    topicsCount.add(1);
                    TopicStats.printTopicStats(stream, topicStats, compactorMXBean, cluster, namespace, name,
                            splitTopicAndPartitionIndexLabel);
                } else {
                    namespaceStats.updateStats(topicStats);
                }
            }));

            if (!includeTopicMetrics) {
                // Only include namespace level stats if we don't have the per-topic, otherwise we're going to
                // report the same data twice, and it will make the aggregation difficult
                printNamespaceStats(stream, namespaceStats, cluster, namespace);
            } else {
                localNamespaceTopicCount.put(namespace, topicsCount.sum());
            }
        });

        if (includeTopicMetrics) {
            printTopicsCountStats(stream, localNamespaceTopicCount, cluster);
        }

        printBrokerStats(stream, cluster, brokerStats);
    }
@Test
    public void testGenerateSubscriptionsStats() {
        // given
        final String namespace = "tenant/cluster/ns";

        // prepare multi-layer topic map
        ConcurrentOpenHashMap bundlesMap = ConcurrentOpenHashMap.newBuilder().build();
        ConcurrentOpenHashMap topicsMap = ConcurrentOpenHashMap.newBuilder().build();
        ConcurrentOpenHashMap subscriptionsMaps = ConcurrentOpenHashMap.newBuilder().build();
        bundlesMap.put("my-bundle", topicsMap);
        multiLayerTopicsMap.put(namespace, bundlesMap);

        // Prepare managed ledger
        ManagedLedger ml = Mockito.mock(ManagedLedger.class);
        ManagedLedgerMBeanImpl mlBeanStats = Mockito.mock(ManagedLedgerMBeanImpl.class);
        StatsBuckets statsBuckets = new StatsBuckets(ManagedLedgerMBeanImpl.ENTRY_LATENCY_BUCKETS_USEC);
        when(mlBeanStats.getInternalAddEntryLatencyBuckets()).thenReturn(statsBuckets);
        when(mlBeanStats.getInternalLedgerAddEntryLatencyBuckets()).thenReturn(statsBuckets);
        when(mlBeanStats.getInternalEntrySizeBuckets()).thenReturn(
                new StatsBuckets(ManagedLedgerMBeanImpl.ENTRY_SIZE_BUCKETS_BYTES));
        when(ml.getStats()).thenReturn(mlBeanStats);

        // Prepare topic and subscription
        PersistentTopic topic = Mockito.mock(PersistentTopic.class);
        Subscription subscription = Mockito.mock(Subscription.class);
        Consumer consumer = Mockito.mock(Consumer.class);
        ConsumerStatsImpl consumerStats = new ConsumerStatsImpl();
        when(consumer.getStats()).thenReturn(consumerStats);
        when(subscription.getConsumers()).thenReturn(List.of(consumer));
        subscriptionsMaps.put("my-subscription", subscription);
        SubscriptionStatsImpl subStats = new SubscriptionStatsImpl();
        TopicStatsImpl topicStats = new TopicStatsImpl();
        topicStats.subscriptions.put("my-subscription", subStats);
        when(topic.getStats(false, false, false)).thenReturn(topicStats);
        when(topic.getBrokerService()).thenReturn(broker);
        when(topic.getSubscriptions()).thenReturn(subscriptionsMaps);
        when(topic.getReplicators()).thenReturn(ConcurrentOpenHashMap.<String,Replicator>newBuilder().build());
        when(topic.getManagedLedger()).thenReturn(ml);
        when(topic.getBacklogQuota(Mockito.any())).thenReturn(Mockito.mock(BacklogQuota.class));
        PersistentTopicMetrics persistentTopicMetrics = new PersistentTopicMetrics();
        when(topic.getPersistentTopicMetrics()).thenReturn(persistentTopicMetrics);
        topicsMap.put("my-topic", topic);
        PrometheusMetricStreams metricStreams = Mockito.spy(new PrometheusMetricStreams());

        // Populate subscriptions stats
        subStats.blockedSubscriptionOnUnackedMsgs = true;
        consumerStats.blockedConsumerOnUnackedMsgs = false; // should not affect blockedSubscriptionOnUnackedMsgs
        consumerStats.unackedMessages = 1;
        consumerStats.msgRateRedeliver = 0.7;
        subStats.consumers.add(0, consumerStats);

        // when
        NamespaceStatsAggregator.generate(pulsar, true, true,
                true, true, metricStreams);

        // then
        verifySubscriptionMetric(metricStreams, "pulsar_subscription_blocked_on_unacked_messages", 1);
        verifyConsumerMetric(metricStreams, "pulsar_consumer_blocked_on_unacked_messages", 0);

        verifySubscriptionMetric(metricStreams, "pulsar_subscription_msg_rate_redeliver", 0.7);
        verifySubscriptionMetric(metricStreams, "pulsar_subscription_unacked_messages", 1L);
    }