org.apache.flink.connector.kafka.source.metrics

Class KafkaSourceReaderMetrics

java.lang.Object

org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics

@PublicEvolving
public class KafkaSourceReaderMetrics
extends Object

A collection class for handling metrics in KafkaSourceReader.

All metrics of Kafka source reader are registered under group "KafkaSourceReader", which is a child group of OperatorMetricGroup. Metrics related to a specific topic partition will be registered in the group "KafkaSourceReader.topic.{topic_name}.partition.{partition_id}".

For example, current consuming offset of topic "my-topic" and partition 1 will be reported in metric: "{some_parent_groups}.operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset"

and number of successful commits will be reported in metric: "{some_parent_groups}.operator.KafkaSourceReader.commitsSucceeded"

All metrics of Kafka consumer are also registered under group "KafkaSourceReader.KafkaConsumer". For example, Kafka consumer metric "records-consumed-total" can be found at: {some_parent_groups}.operator.KafkaSourceReader.KafkaConsumer.records-consumed-total"

Field Summary

Fields

Modifier and Type	Field and Description
static String	BYTES_CONSUMED_TOTAL
static String	COMMITS_FAILED_METRIC_COUNTER
static String	COMMITS_SUCCEEDED_METRIC_COUNTER
static String	COMMITTED_OFFSET_METRIC_GAUGE
static String	CONSUMER_FETCH_MANAGER_GROUP
static String	CURRENT_OFFSET_METRIC_GAUGE
static long	INITIAL_OFFSET
static String	KAFKA_CONSUMER_METRIC_GROUP
static String	KAFKA_SOURCE_READER_METRIC_GROUP
static String	PARTITION_GROUP
static String	RECORDS_LAG
static String	TOPIC_GROUP

Constructor Summary

Constructors

Constructor and Description
KafkaSourceReaderMetrics(SourceReaderMetricGroup sourceReaderMetricGroup)

Method Summary

Modifier and Type	Method and Description
void	maybeAddRecordsLagMetric(org.apache.kafka.clients.consumer.KafkaConsumer<?,?> consumer, org.apache.kafka.common.TopicPartition tp) Add a partition's records-lag metric to tracking list if this partition never appears before.
void	recordCommittedOffset(org.apache.kafka.common.TopicPartition tp, long offset) Update the latest committed offset of the given TopicPartition.
void	recordCurrentOffset(org.apache.kafka.common.TopicPartition tp, long offset) Update current consuming offset of the given TopicPartition.
void	recordFailedCommit() Mark a failure commit.
void	recordSucceededCommit() Mark a successful commit.
void	registerKafkaConsumerMetrics(org.apache.kafka.clients.consumer.KafkaConsumer<?,?> kafkaConsumer) Register metrics of KafkaConsumer in Kafka metric group.
void	registerNumBytesIn(org.apache.kafka.clients.consumer.KafkaConsumer<?,?> consumer) Register MetricNames.IO_NUM_BYTES_IN.
void	registerTopicPartition(org.apache.kafka.common.TopicPartition tp) Register metric groups for the given TopicPartition.
void	removeRecordsLagMetric(org.apache.kafka.common.TopicPartition tp) Remove a partition's records-lag metric from tracking list.
void	updateNumBytesInCounter() Update MetricNames.IO_NUM_BYTES_IN.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

KAFKA_SOURCE_READER_METRIC_GROUP

public static final String KAFKA_SOURCE_READER_METRIC_GROUP

See Also:

Constant Field Values

TOPIC_GROUP

public static final String TOPIC_GROUP

See Also:

Constant Field Values

PARTITION_GROUP

public static final String PARTITION_GROUP

See Also:

Constant Field Values

CURRENT_OFFSET_METRIC_GAUGE

public static final String CURRENT_OFFSET_METRIC_GAUGE

See Also:

Constant Field Values

COMMITTED_OFFSET_METRIC_GAUGE

public static final String COMMITTED_OFFSET_METRIC_GAUGE

See Also:

Constant Field Values

COMMITS_SUCCEEDED_METRIC_COUNTER

public static final String COMMITS_SUCCEEDED_METRIC_COUNTER

See Also:

Constant Field Values

COMMITS_FAILED_METRIC_COUNTER

public static final String COMMITS_FAILED_METRIC_COUNTER

See Also:

Constant Field Values

KAFKA_CONSUMER_METRIC_GROUP

public static final String KAFKA_CONSUMER_METRIC_GROUP

See Also:

Constant Field Values

CONSUMER_FETCH_MANAGER_GROUP

public static final String CONSUMER_FETCH_MANAGER_GROUP

See Also:

Constant Field Values

BYTES_CONSUMED_TOTAL

public static final String BYTES_CONSUMED_TOTAL

See Also:

Constant Field Values

RECORDS_LAG

public static final String RECORDS_LAG

See Also:

Constant Field Values

INITIAL_OFFSET

public static final long INITIAL_OFFSET

See Also:

Constant Field Values

Constructor Detail

KafkaSourceReaderMetrics

public KafkaSourceReaderMetrics(SourceReaderMetricGroup sourceReaderMetricGroup)

Method Detail

registerKafkaConsumerMetrics

public void registerKafkaConsumerMetrics(org.apache.kafka.clients.consumer.KafkaConsumer<?,?> kafkaConsumer)

Register metrics of KafkaConsumer in Kafka metric group.

Parameters:

kafkaConsumer - Kafka consumer used by partition split reader.

registerTopicPartition

public void registerTopicPartition(org.apache.kafka.common.TopicPartition tp)

Register metric groups for the given TopicPartition.

Parameters:

tp - Registering topic partition

recordCurrentOffset

public void recordCurrentOffset(org.apache.kafka.common.TopicPartition tp,
                                long offset)

Update current consuming offset of the given TopicPartition.

Parameters:

tp - Updating topic partition

offset - Current consuming offset

recordCommittedOffset

public void recordCommittedOffset(org.apache.kafka.common.TopicPartition tp,
                                  long offset)

Update the latest committed offset of the given TopicPartition.

Parameters:

tp - Updating topic partition

offset - Committing offset

recordSucceededCommit

public void recordSucceededCommit()

Mark a successful commit.

recordFailedCommit

public void recordFailedCommit()

Mark a failure commit.

registerNumBytesIn

public void registerNumBytesIn(org.apache.kafka.clients.consumer.KafkaConsumer<?,?> consumer)

Register MetricNames.IO_NUM_BYTES_IN.

Parameters:

consumer - Kafka consumer

maybeAddRecordsLagMetric

public void maybeAddRecordsLagMetric(org.apache.kafka.clients.consumer.KafkaConsumer<?,?> consumer,
                                     org.apache.kafka.common.TopicPartition tp)

Add a partition's records-lag metric to tracking list if this partition never appears before.

This method also lazily register MetricNames.PENDING_RECORDS in SourceReaderMetricGroup

Parameters:

consumer - Kafka consumer

tp - Topic partition

removeRecordsLagMetric

public void removeRecordsLagMetric(org.apache.kafka.common.TopicPartition tp)

Remove a partition's records-lag metric from tracking list.

Parameters:

tp - Unassigned topic partition

updateNumBytesInCounter

public void updateNumBytesInCounter()

Update MetricNames.IO_NUM_BYTES_IN.

Instead of simply setting OperatorIOMetricGroup.getNumBytesInCounter() to the same value as bytes-consumed-total from Kafka consumer, which will screw TaskIOMetricGroup.getNumBytesInCounter() if chained sources exist, we track the increment of bytes-consumed-total and count it towards the counter.