/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.pulsar.broker.stats.prometheus;

import static org.apache.pulsar.common.naming.TopicName.PARTITIONED_TOPIC_SUFFIX;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import org.apache.bookkeeper.mledger.util.StatsBuckets;
import org.apache.pulsar.common.util.SimpleTextOutputStream;
import org.apache.pulsar.compaction.CompactionRecord;
import org.apache.pulsar.compaction.CompactorMXBean;

class TopicStats {

    int subscriptionsCount;
    int producersCount;
    int consumersCount;
    double rateIn;
    double rateOut;
    double throughputIn;
    double throughputOut;
    long msgInCounter;
    long bytesInCounter;
    long msgOutCounter;
    long bytesOutCounter;
    double averageMsgSize;

    public long msgBacklog;
    long publishRateLimitedTimes;

    long backlogQuotaLimit;
    long backlogQuotaLimitTime;

    ManagedLedgerStats managedLedgerStats = new ManagedLedgerStats();

    Map<String, AggregatedReplicationStats> replicationStats = new HashMap<>();
    Map<String, AggregatedSubscriptionStats> subscriptionStats = new HashMap<>();
    Map<String, AggregatedProducerStats> producerStats = new HashMap<>();

    // Used for tracking duplicate TYPE definitions
    static Map<String, String> metricWithTypeDefinition = new HashMap<>();

    // For compaction
    long compactionRemovedEventCount;
    long compactionSucceedCount;
    long compactionFailedCount;
    long compactionDurationTimeInMills;
    double compactionReadThroughput;
    double compactionWriteThroughput;
    long compactionCompactedEntriesCount;
    long compactionCompactedEntriesSize;
    StatsBuckets compactionLatencyBuckets = new StatsBuckets(CompactionRecord.WRITE_LATENCY_BUCKETS_USEC);
    public int delayedTrackerMemoryUsage;

    public void reset() {
        subscriptionsCount = 0;
        producersCount = 0;
        consumersCount = 0;
        rateIn = 0;
        rateOut = 0;
        throughputIn = 0;
        throughputOut = 0;
        bytesInCounter = 0;
        msgInCounter = 0;
        bytesOutCounter = 0;
        msgOutCounter = 0;

        managedLedgerStats.reset();
        msgBacklog = 0;
        publishRateLimitedTimes = 0L;
        backlogQuotaLimit = 0;
        backlogQuotaLimitTime = -1;

        replicationStats.clear();
        subscriptionStats.clear();
        producerStats.clear();

        compactionRemovedEventCount = 0;
        compactionSucceedCount = 0;
        compactionFailedCount = 0;
        compactionDurationTimeInMills = 0;
        compactionReadThroughput = 0;
        compactionWriteThroughput = 0;
        compactionCompactedEntriesCount = 0;
        compactionCompactedEntriesSize = 0;
        compactionLatencyBuckets.reset();
        delayedTrackerMemoryUsage = 0;
    }

    static void resetTypes() {
        metricWithTypeDefinition.clear();
    }

    static void printTopicStats(SimpleTextOutputStream stream, String cluster, String namespace, String topic,
                                TopicStats stats, Optional<CompactorMXBean> compactorMXBean,
                                boolean splitTopicAndPartitionIndexLabel) {
        metric(stream, cluster, namespace, topic, "pulsar_subscriptions_count", stats.subscriptionsCount,
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_producers_count", stats.producersCount,
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_consumers_count", stats.consumersCount,
                splitTopicAndPartitionIndexLabel);

        metric(stream, cluster, namespace, topic, "pulsar_rate_in", stats.rateIn,
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_rate_out", stats.rateOut,
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_throughput_in", stats.throughputIn,
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_throughput_out", stats.throughputOut,
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_average_msg_size", stats.averageMsgSize,
                splitTopicAndPartitionIndexLabel);

        metric(stream, cluster, namespace, topic, "pulsar_storage_size", stats.managedLedgerStats.storageSize,
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_logical_size",
                stats.managedLedgerStats.storageLogicalSize, splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_msg_backlog", stats.msgBacklog,
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_write_rate",
                stats.managedLedgerStats.storageWriteRate, splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_read_rate", stats.managedLedgerStats.storageReadRate,
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_backlog_size",
                stats.managedLedgerStats.backlogSize, splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_publish_rate_limit_times", stats.publishRateLimitedTimes,
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_offloaded_size", stats.managedLedgerStats
                .offloadedStorageUsed, splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_backlog_quota_limit", stats.backlogQuotaLimit,
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_backlog_quota_limit_time",
                stats.backlogQuotaLimitTime, splitTopicAndPartitionIndexLabel);

        metric(stream, cluster, namespace, topic, "pulsar_delayed_message_index_size_bytes",
                stats.delayedTrackerMemoryUsage, splitTopicAndPartitionIndexLabel);

        long[] latencyBuckets = stats.managedLedgerStats.storageWriteLatencyBuckets.getBuckets();
        metric(stream, cluster, namespace, topic, "pulsar_storage_write_latency_le_0_5", latencyBuckets[0],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_write_latency_le_1", latencyBuckets[1],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_write_latency_le_5", latencyBuckets[2],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_write_latency_le_10", latencyBuckets[3],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_write_latency_le_20", latencyBuckets[4],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_write_latency_le_50", latencyBuckets[5],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_write_latency_le_100", latencyBuckets[6],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_write_latency_le_200", latencyBuckets[7],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_write_latency_le_1000", latencyBuckets[8],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_write_latency_overflow", latencyBuckets[9],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_write_latency_count",
                stats.managedLedgerStats.storageWriteLatencyBuckets.getCount(), splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_write_latency_sum",
                stats.managedLedgerStats.storageWriteLatencyBuckets.getSum(), splitTopicAndPartitionIndexLabel);

        long[] ledgerWriteLatencyBuckets = stats.managedLedgerStats.storageLedgerWriteLatencyBuckets.getBuckets();
        metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_0_5",
                ledgerWriteLatencyBuckets[0], splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_1",
                ledgerWriteLatencyBuckets[1], splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_5",
                ledgerWriteLatencyBuckets[2], splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_10",
                ledgerWriteLatencyBuckets[3], splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_20",
                ledgerWriteLatencyBuckets[4], splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_50",
                ledgerWriteLatencyBuckets[5], splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_100",
                ledgerWriteLatencyBuckets[6], splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_200",
                ledgerWriteLatencyBuckets[7], splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_1000",
                ledgerWriteLatencyBuckets[8], splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_overflow",
                ledgerWriteLatencyBuckets[9], splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_count",
                stats.managedLedgerStats.storageLedgerWriteLatencyBuckets.getCount(),
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_sum",
                stats.managedLedgerStats.storageLedgerWriteLatencyBuckets.getSum(),
                splitTopicAndPartitionIndexLabel);

        long[] entrySizeBuckets = stats.managedLedgerStats.entrySizeBuckets.getBuckets();
        metric(stream, cluster, namespace, topic, "pulsar_entry_size_le_128", entrySizeBuckets[0],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_entry_size_le_512", entrySizeBuckets[1],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_entry_size_le_1_kb", entrySizeBuckets[2],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_entry_size_le_2_kb", entrySizeBuckets[3],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_entry_size_le_4_kb", entrySizeBuckets[4],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_entry_size_le_16_kb", entrySizeBuckets[5],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_entry_size_le_100_kb", entrySizeBuckets[6],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_entry_size_le_1_mb", entrySizeBuckets[7],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_entry_size_le_overflow", entrySizeBuckets[8],
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_entry_size_count",
                stats.managedLedgerStats.entrySizeBuckets.getCount(), splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_entry_size_sum",
                stats.managedLedgerStats.entrySizeBuckets.getSum(), splitTopicAndPartitionIndexLabel);

        stats.producerStats.forEach((p, producerStats) -> {
            metric(stream, cluster, namespace, topic, p, producerStats.producerId, "pulsar_producer_msg_rate_in",
                    producerStats.msgRateIn, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, p, producerStats.producerId, "pulsar_producer_msg_throughput_in",
                    producerStats.msgThroughputIn, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, p, producerStats.producerId, "pulsar_producer_msg_average_Size",
                    producerStats.averageMsgSize, splitTopicAndPartitionIndexLabel);
        });

        stats.subscriptionStats.forEach((n, subsStats) -> {
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_back_log",
                    subsStats.msgBacklog, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_back_log_no_delayed",
                    subsStats.msgBacklogNoDelayed, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_delayed",
                    subsStats.msgDelayed, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_msg_rate_redeliver",
                    subsStats.msgRateRedeliver, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_unacked_messages",
                    subsStats.unackedMessages, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_blocked_on_unacked_messages",
                    subsStats.blockedSubscriptionOnUnackedMsgs ? 1 : 0, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_msg_rate_out",
                    subsStats.msgRateOut, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_msg_ack_rate",
                    subsStats.messageAckRate, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_msg_throughput_out",
                    subsStats.msgThroughputOut, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_out_bytes_total",
                    subsStats.bytesOutCounter, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_out_messages_total",
                    subsStats.msgOutCounter, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_last_expire_timestamp",
                    subsStats.lastExpireTimestamp, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_last_acked_timestamp",
                subsStats.lastAckedTimestamp, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_last_consumed_flow_timestamp",
                subsStats.lastConsumedFlowTimestamp, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_last_consumed_timestamp",
                subsStats.lastConsumedTimestamp, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_last_mark_delete_advanced_timestamp",
                subsStats.lastMarkDeleteAdvancedTimestamp, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_msg_rate_expired",
                    subsStats.msgRateExpired, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_total_msg_expired",
                    subsStats.totalMsgExpired, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_msg_drop_rate",
                    subsStats.msgDropRate, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_consumers_count",
                    subsStats.consumersCount, splitTopicAndPartitionIndexLabel);
            subsStats.consumerStat.forEach((c, consumerStats) -> {
                metric(stream, cluster, namespace, topic, n, c.consumerName(), c.consumerId(),
                        "pulsar_consumer_msg_rate_redeliver", consumerStats.msgRateRedeliver,
                        splitTopicAndPartitionIndexLabel);
                metric(stream, cluster, namespace, topic, n, c.consumerName(), c.consumerId(),
                        "pulsar_consumer_unacked_messages", consumerStats.unackedMessages,
                        splitTopicAndPartitionIndexLabel);
                metric(stream, cluster, namespace, topic, n, c.consumerName(), c.consumerId(),
                        "pulsar_consumer_blocked_on_unacked_messages",
                        consumerStats.blockedSubscriptionOnUnackedMsgs ? 1 : 0,
                        splitTopicAndPartitionIndexLabel);
                metric(stream, cluster, namespace, topic, n, c.consumerName(), c.consumerId(),
                        "pulsar_consumer_msg_rate_out", consumerStats.msgRateOut,
                        splitTopicAndPartitionIndexLabel);

                metric(stream, cluster, namespace, topic, n, c.consumerName(), c.consumerId(),
                        "pulsar_consumer_msg_ack_rate", consumerStats.msgAckRate,
                        splitTopicAndPartitionIndexLabel);

                metric(stream, cluster, namespace, topic, n, c.consumerName(), c.consumerId(),
                        "pulsar_consumer_msg_throughput_out", consumerStats.msgThroughputOut,
                        splitTopicAndPartitionIndexLabel);
                metric(stream, cluster, namespace, topic, n, c.consumerName(), c.consumerId(),
                        "pulsar_consumer_available_permits", consumerStats.availablePermits,
                        splitTopicAndPartitionIndexLabel);
                metric(stream, cluster, namespace, topic, n, c.consumerName(), c.consumerId(),
                        "pulsar_out_bytes_total", consumerStats.bytesOutCounter,
                        splitTopicAndPartitionIndexLabel);
                metric(stream, cluster, namespace, topic, n, c.consumerName(), c.consumerId(),
                        "pulsar_out_messages_total", consumerStats.msgOutCounter,
                        splitTopicAndPartitionIndexLabel);
            });
        });

        if (!stats.replicationStats.isEmpty()) {
            stats.replicationStats.forEach((remoteCluster, replStats) -> {
                metricWithRemoteCluster(stream, cluster, namespace, topic, "pulsar_replication_rate_in", remoteCluster,
                        replStats.msgRateIn, splitTopicAndPartitionIndexLabel);
                metricWithRemoteCluster(stream, cluster, namespace, topic, "pulsar_replication_rate_out", remoteCluster,
                        replStats.msgRateOut, splitTopicAndPartitionIndexLabel);
                metricWithRemoteCluster(stream, cluster, namespace, topic, "pulsar_replication_throughput_in",
                        remoteCluster, replStats.msgThroughputIn, splitTopicAndPartitionIndexLabel);
                metricWithRemoteCluster(stream, cluster, namespace, topic, "pulsar_replication_throughput_out",
                        remoteCluster, replStats.msgThroughputOut, splitTopicAndPartitionIndexLabel);
                metricWithRemoteCluster(stream, cluster, namespace, topic, "pulsar_replication_backlog", remoteCluster,
                        replStats.replicationBacklog, splitTopicAndPartitionIndexLabel);
                metricWithRemoteCluster(stream, cluster, namespace, topic, "pulsar_replication_connected_count",
                        remoteCluster, replStats.connectedCount, splitTopicAndPartitionIndexLabel);
                metricWithRemoteCluster(stream, cluster, namespace, topic, "pulsar_replication_rate_expired",
                        remoteCluster, replStats.msgRateExpired, splitTopicAndPartitionIndexLabel);
                metricWithRemoteCluster(stream, cluster, namespace, topic, "pulsar_replication_delay_in_seconds",
                        remoteCluster, replStats.replicationDelayInSeconds, splitTopicAndPartitionIndexLabel);
            });
        }

        metric(stream, cluster, namespace, topic, "pulsar_in_bytes_total", stats.bytesInCounter,
                splitTopicAndPartitionIndexLabel);
        metric(stream, cluster, namespace, topic, "pulsar_in_messages_total", stats.msgInCounter,
                splitTopicAndPartitionIndexLabel);

        // Compaction
        boolean hasCompaction = compactorMXBean.flatMap(mxBean -> mxBean.getCompactionRecordForTopic(topic))
                .map(__ -> true).orElse(false);
        if (hasCompaction) {
            metric(stream, cluster, namespace, topic, "pulsar_compaction_removed_event_count",
                    stats.compactionRemovedEventCount, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_succeed_count",
                    stats.compactionSucceedCount, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_failed_count",
                    stats.compactionFailedCount, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_duration_time_in_mills",
                    stats.compactionDurationTimeInMills, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_read_throughput",
                    stats.compactionReadThroughput, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_write_throughput",
                    stats.compactionWriteThroughput, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_compacted_entries_count",
                    stats.compactionCompactedEntriesCount, splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_compacted_entries_size",
                    stats.compactionCompactedEntriesSize, splitTopicAndPartitionIndexLabel);
            long[] compactionLatencyBuckets = stats.compactionLatencyBuckets.getBuckets();
            metric(stream, cluster, namespace, topic, "pulsar_compaction_latency_le_0_5",
                    compactionLatencyBuckets[0], splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_latency_le_1",
                    compactionLatencyBuckets[1], splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_latency_le_5",
                    compactionLatencyBuckets[2], splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_latency_le_10",
                    compactionLatencyBuckets[3], splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_latency_le_20",
                    compactionLatencyBuckets[4], splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_latency_le_50",
                    compactionLatencyBuckets[5], splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_latency_le_100",
                    compactionLatencyBuckets[6], splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_latency_le_200",
                    compactionLatencyBuckets[7], splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_latency_le_1000",
                    compactionLatencyBuckets[8], splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_latency_overflow",
                    compactionLatencyBuckets[9], splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_latency_sum",
                    stats.compactionLatencyBuckets.getSum(), splitTopicAndPartitionIndexLabel);
            metric(stream, cluster, namespace, topic, "pulsar_compaction_latency_count",
                    stats.compactionLatencyBuckets.getCount(), splitTopicAndPartitionIndexLabel);
        }
    }

    static void metricType(SimpleTextOutputStream stream, String name) {

        if (!metricWithTypeDefinition.containsKey(name)) {
            metricWithTypeDefinition.put(name, "gauge");
            stream.write("# TYPE ").write(name).write(" gauge\n");
        }

    }

    private static void metric(SimpleTextOutputStream stream, String cluster, String namespace, String topic,
            String name, double value, boolean splitTopicAndPartitionIndexLabel) {
        metricType(stream, name);
        appendRequiredLabels(stream, cluster, namespace, topic, name, splitTopicAndPartitionIndexLabel).write("\"} ");
        stream.write(value);
        appendEndings(stream);
    }

    private static void metric(SimpleTextOutputStream stream, String cluster, String namespace, String topic,
           String subscription, String name, long value, boolean splitTopicAndPartitionIndexLabel) {
        metricType(stream, name);
        appendRequiredLabels(stream, cluster, namespace, topic, name, splitTopicAndPartitionIndexLabel)
                .write("\",subscription=\"").write(subscription).write("\"} ");
        stream.write(value);
        appendEndings(stream);
    }

    private static void metric(SimpleTextOutputStream stream, String cluster, String namespace, String topic,
            String producerName, long produceId, String name, double value, boolean splitTopicAndPartitionIndexLabel) {
        metricType(stream, name);
        appendRequiredLabels(stream, cluster, namespace, topic, name, splitTopicAndPartitionIndexLabel)
                .write("\",producer_name=\"").write(producerName)
                .write("\",producer_id=\"").write(produceId).write("\"} ");
        stream.write(value);
        appendEndings(stream);
    }

    private static void metric(SimpleTextOutputStream stream, String cluster, String namespace, String topic,
            String subscription, String name, double value, boolean splitTopicAndPartitionIndexLabel) {
        metricType(stream, name);
        appendRequiredLabels(stream, cluster, namespace, topic, name, splitTopicAndPartitionIndexLabel)
                .write("\",subscription=\"").write(subscription).write("\"} ");
        stream.write(value);
        appendEndings(stream);
    }

    private static void metric(SimpleTextOutputStream stream, String cluster, String namespace, String topic,
            String subscription, String consumerName, long consumerId, String name, long value,
            boolean splitTopicAndPartitionIndexLabel) {
        metricType(stream, name);
        appendRequiredLabels(stream, cluster, namespace, topic, name, splitTopicAndPartitionIndexLabel)
                .write("\",subscription=\"").write(subscription)
                .write("\",consumer_name=\"").write(consumerName).write("\",consumer_id=\"").write(consumerId)
                .write("\"} ");
        stream.write(value);
        appendEndings(stream);
    }

    private static void metric(SimpleTextOutputStream stream, String cluster, String namespace, String topic,
            String subscription, String consumerName, long consumerId, String name, double value,
            boolean splitTopicAndPartitionIndexLabel) {
        metricType(stream, name);
        appendRequiredLabels(stream, cluster, namespace, topic, name, splitTopicAndPartitionIndexLabel)
                .write("\",subscription=\"").write(subscription)
                .write("\",consumer_name=\"").write(consumerName).write("\",consumer_id=\"")
                .write(consumerId).write("\"} ");
        stream.write(value);
        appendEndings(stream);
    }

    private static void metricWithRemoteCluster(SimpleTextOutputStream stream, String cluster, String namespace,
            String topic, String name, String remoteCluster, double value, boolean splitTopicAndPartitionIndexLabel) {
        metricType(stream, name);
        appendRequiredLabels(stream, cluster, namespace, topic, name, splitTopicAndPartitionIndexLabel)
                .write("\",remote_cluster=\"").write(remoteCluster).write("\"} ");
        stream.write(value);
        appendEndings(stream);
    }

    private static SimpleTextOutputStream appendRequiredLabels(SimpleTextOutputStream stream, String cluster,
            String namespace, String topic, String name, boolean splitTopicAndPartitionIndexLabel) {
        stream.write(name).write("{cluster=\"").write(cluster).write("\",namespace=\"").write(namespace);
        if (splitTopicAndPartitionIndexLabel) {
            int index = topic.indexOf(PARTITIONED_TOPIC_SUFFIX);
            if (index > 0) {
                stream.write("\",topic=\"").write(topic.substring(0, index)).write("\",partition=\"")
                        .write(topic.substring(index + PARTITIONED_TOPIC_SUFFIX.length()));
            } else {
                stream.write("\",topic=\"").write(topic).write("\",partition=\"").write("-1");
            }
        } else {
            stream.write("\",topic=\"").write(topic);
        }
        return stream;
    }

    private static void appendEndings(SimpleTextOutputStream stream) {
        stream.write(' ').write(System.currentTimeMillis()).write('\n');
    }
}