kafka TopicMetrics 源码

kafka TopicMetrics 代码

文件路径：/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/TopicMetrics.java

/*
 * 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.kafka.streams.processor.internals.metrics;

import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.metrics.Sensor.RecordingLevel;

import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.TOPIC_LEVEL_GROUP;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.TOTAL_DESCRIPTION;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addTotalCountAndSumMetricsToSensor;

public class TopicMetrics {

    private static final String CONSUMED = "consumed";
    private static final String BYTES_CONSUMED = "bytes-consumed";
    private static final String BYTES_CONSUMED_DESCRIPTION = "bytes consumed from this topic";
    private static final String BYTES_CONSUMED_TOTAL_DESCRIPTION = TOTAL_DESCRIPTION + BYTES_CONSUMED_DESCRIPTION;
    private static final String RECORDS_CONSUMED = "records-consumed";
    private static final String RECORDS_CONSUMED_DESCRIPTION = "records consumed from this topic";
    private static final String RECORDS_CONSUMED_TOTAL_DESCRIPTION = TOTAL_DESCRIPTION + RECORDS_CONSUMED_DESCRIPTION;

    private static final String PRODUCED = "produced";
    private static final String BYTES_PRODUCED = "bytes-produced";
    private static final String BYTES_PRODUCED_DESCRIPTION = "bytes produced to this topic";
    private static final String BYTES_PRODUCED_TOTAL_DESCRIPTION = TOTAL_DESCRIPTION + BYTES_PRODUCED_DESCRIPTION;
    private static final String RECORDS_PRODUCED = "records-produced";
    private static final String RECORDS_PRODUCED_DESCRIPTION = "records produced to this topic";
    private static final String RECORDS_PRODUCED_TOTAL_DESCRIPTION = TOTAL_DESCRIPTION + RECORDS_PRODUCED_DESCRIPTION;

    public static Sensor consumedSensor(final String threadId,
                                        final String taskId,
                                        final String processorNodeId,
                                        final String topic,
                                        final StreamsMetricsImpl streamsMetrics) {
        final Sensor sensor = streamsMetrics.topicLevelSensor(
            threadId,
            taskId,
            processorNodeId,
            topic,
            CONSUMED,
            RecordingLevel.INFO);
        addTotalCountAndSumMetricsToSensor(
            sensor,
            TOPIC_LEVEL_GROUP,
            streamsMetrics.topicLevelTagMap(threadId, taskId, processorNodeId, topic),
            RECORDS_CONSUMED,
            BYTES_CONSUMED,
            RECORDS_CONSUMED_TOTAL_DESCRIPTION,
            BYTES_CONSUMED_TOTAL_DESCRIPTION
        );
        return sensor;
    }

    public static Sensor producedSensor(final String threadId,
                                        final String taskId,
                                        final String processorNodeId,
                                        final String topic,
                                        final StreamsMetricsImpl streamsMetrics) {
        final Sensor sensor = streamsMetrics.topicLevelSensor(
            threadId,
            taskId,
            processorNodeId,
            topic,
            PRODUCED,
            RecordingLevel.INFO);
        addTotalCountAndSumMetricsToSensor(
            sensor,
            TOPIC_LEVEL_GROUP,
            streamsMetrics.topicLevelTagMap(threadId, taskId, processorNodeId, topic),
            RECORDS_PRODUCED,
            BYTES_PRODUCED,
            RECORDS_PRODUCED_TOTAL_DESCRIPTION,
            BYTES_PRODUCED_TOTAL_DESCRIPTION
        );
        return sensor;
    }

}

相关信息

kafka 源码目录

相关文章

kafka ProcessorNodeMetrics 源码

kafka StreamsMetricsImpl 源码

kafka TaskMetrics 源码

kafka ThreadMetrics 源码