You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by jq...@apache.org on 2021/07/13 02:41:13 UTC
[flink] 03/03: [FLINK-22766][connector/kafka] Report offsets and
Kafka consumer metrics in Flink metric group
This is an automated email from the ASF dual-hosted git repository.
jqin pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2c455f324b9ec7ef053253cf4904413b1e5f7a98
Author: Qingsheng Ren <re...@gmail.com>
AuthorDate: Tue Jun 8 11:57:52 2021 +0800
[FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
(cherry picked from commit b094a932845db5539fc07b032d49d0bcefd15df2)
---
docs/content/docs/connectors/datastream/kafka.md | 36 ++++
.../flink/connector/kafka/source/KafkaSource.java | 11 +-
.../connector/kafka/source/KafkaSourceOptions.java | 9 +-
.../source/metrics/KafkaSourceReaderMetrics.java | 183 +++++++++++++++++++++
.../source/reader/KafkaPartitionSplitReader.java | 33 +++-
.../kafka/source/reader/KafkaSourceReader.java | 11 +-
.../metrics/KafkaSourceReaderMetricsTest.java | 166 +++++++++++++++++++
.../reader/KafkaPartitionSplitReaderTest.java | 8 +-
.../kafka/source/reader/KafkaSourceReaderTest.java | 122 +++++++++++++-
.../source/reader/TestingReaderContext.java | 7 +-
10 files changed, 575 insertions(+), 11 deletions(-)
diff --git a/docs/content/docs/connectors/datastream/kafka.md b/docs/content/docs/connectors/datastream/kafka.md
index 1f21e7d..d540004 100644
--- a/docs/content/docs/connectors/datastream/kafka.md
+++ b/docs/content/docs/connectors/datastream/kafka.md
@@ -153,6 +153,8 @@ KafkaSource has following options for configuration:
- ```partition.discovery.interval.ms``` defines the interval im milliseconds for Kafka source
to discover new partitions. See <a href="#dynamic-partition-discovery">Dynamic Partition Discovery</a>
below for more details.
+- ```register.consumer.metrics``` specifies whether to register metrics of KafkaConsumer in Flink
+metric group
For configurations of KafkaConsumer, you can refer to
<a href="http://kafka.apache.org/documentation/#consumerconfigs">Apache Kafka documentation</a>
@@ -210,6 +212,40 @@ the properties of Kafka consumer.
Note that Kafka source does **NOT** rely on committed offsets for fault tolerance. Committing offset
is only for exposing the progress of consumer and consuming group for monitoring.
+### Monitoring
+Kafka source exposes metrics in Flink's metric group for monitoring and diagnosing.
+#### Scope of Metric
+All metrics of Kafka source reader are registered under group ```KafkaSourceReader```, which is a
+child group of operator metric group. 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``` .
+
+#### List of Metrics
+
+| Metric Name | Description | Scope |
+|:----------------:|:-----------------------------------------------:|:-----------------:|
+| currentOffset | Current consuming offset of the topic partition | TopicPartition |
+| committedOffset | Committed offset of the topic partition | TopicPartition |
+| commitsSucceeded | Number of successful commits | KafkaSourceReader |
+| commitsFailed | Number of failed commits | KafkaSourceReader |
+
+#### Kafka Consumer Metrics
+All metrics of Kafka consumer are also registered under group ```KafkaSourceReader.KafkaConsumer```.
+For example, Kafka consumer metric "records-consumed-total" will be reported in metric:
+```<some_parent_groups>.operator.KafkaSourceReader.KafkaConsumer.records-consumed-total``` .
+
+You can configure whether to register Kafka consumer's metric by configuring option
+```register.consumer.metrics```. This option will be set as true by default.
+
+For metrics of Kafka consumer, you can refer to
+<a href="http://kafka.apache.org/documentation/#consumer_monitoring">Apache Kafka Documentation</a>
+for more details.
+
### Behind the Scene
{{< hint info >}}
If you are interested in how Kafka source works under the design of new data source API, you may
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSource.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSource.java
index dd1c3bf..35fd954 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSource.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSource.java
@@ -36,6 +36,7 @@ import org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumStateSe
import org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaSubscriber;
+import org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics;
import org.apache.flink.connector.kafka.source.reader.KafkaPartitionSplitReader;
import org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter;
import org.apache.flink.connector.kafka.source.reader.KafkaSourceReader;
@@ -133,11 +134,16 @@ public class KafkaSource<OUT>
return readerContext.getUserCodeClassLoader();
}
});
+ final KafkaSourceReaderMetrics kafkaSourceReaderMetrics =
+ new KafkaSourceReaderMetrics(readerContext.metricGroup());
Supplier<KafkaPartitionSplitReader<OUT>> splitReaderSupplier =
() ->
new KafkaPartitionSplitReader<>(
- props, deserializationSchema, readerContext.getIndexOfSubtask());
+ props,
+ deserializationSchema,
+ readerContext.getIndexOfSubtask(),
+ kafkaSourceReaderMetrics);
KafkaRecordEmitter<OUT> recordEmitter = new KafkaRecordEmitter<>();
return new KafkaSourceReader<>(
@@ -145,7 +151,8 @@ public class KafkaSource<OUT>
splitReaderSupplier,
recordEmitter,
toConfiguration(props),
- readerContext);
+ readerContext,
+ kafkaSourceReaderMetrics);
}
@Override
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java
index 98c735f..e48804b 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java
@@ -41,9 +41,16 @@ public class KafkaSourceOptions {
"The interval in milliseconds for the Kafka source to discover "
+ "the new partitions. A non-positive value disables the partition discovery.");
+ public static final ConfigOption<Boolean> REGISTER_KAFKA_CONSUMER_METRICS =
+ ConfigOptions.key("register.consumer.metrics")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Whether to register metrics of KafkaConsumer into Flink metric group");
+
@SuppressWarnings("unchecked")
public static <T> T getOption(
- Properties props, ConfigOption configOption, Function<String, T> parser) {
+ Properties props, ConfigOption<?> configOption, Function<String, T> parser) {
String value = props.getProperty(configOption.key());
return (T) (value == null ? configOption.defaultValue() : parser.apply(value));
}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetrics.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetrics.java
new file mode 100644
index 0000000..df1abe5
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetrics.java
@@ -0,0 +1,183 @@
+/*
+ * 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.flink.connector.kafka.source.metrics;
+
+import org.apache.flink.connector.kafka.source.reader.KafkaSourceReader;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.TopicPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A collection class for handling metrics in {@link KafkaSourceReader}.
+ *
+ * <p>All metrics of Kafka source reader are registered under group "KafkaSourceReader", which is a
+ * child group of {@link org.apache.flink.runtime.metrics.groups.OperatorMetricGroup}. Metrics
+ * related to a specific topic partition will be registered in the group
+ * "KafkaSourceReader.topic.{topic_name}.partition.{partition_id}".
+ *
+ * <p>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"
+ *
+ * <p>and number of successful commits will be reported in metric:
+ * "{some_parent_groups}.operator.KafkaSourceReader.commitsSucceeded"
+ *
+ * <p>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"
+ */
+public class KafkaSourceReaderMetrics {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaSourceReaderMetrics.class);
+
+ // Constants
+ public static final String KAFKA_SOURCE_READER_METRIC_GROUP = "KafkaSourceReader";
+ public static final String TOPIC_GROUP = "topic";
+ public static final String PARTITION_GROUP = "partition";
+ public static final String CURRENT_OFFSET_METRIC_GAUGE = "currentOffset";
+ public static final String COMMITTED_OFFSET_METRIC_GAUGE = "committedOffset";
+ public static final String COMMITS_SUCCEEDED_METRIC_COUNTER = "commitsSucceeded";
+ public static final String COMMITS_FAILED_METRIC_COUNTER = "commitsFailed";
+ public static final String KAFKA_CONSUMER_METRIC_GROUP = "KafkaConsumer";
+
+ public static final long INITIAL_OFFSET = -1;
+
+ // Metric group for registering Kafka related metrics
+ private final MetricGroup kafkaSourceReaderMetricGroup;
+
+ // Successful / Failed commits counters
+ private final Counter commitsSucceeded;
+ private final Counter commitsFailed;
+
+ // Map for tracking current consuming / committing offsets
+ private final Map<TopicPartition, Offset> offsets = new HashMap<>();
+
+ public KafkaSourceReaderMetrics(MetricGroup parentMetricGroup) {
+ this.kafkaSourceReaderMetricGroup =
+ parentMetricGroup.addGroup(KAFKA_SOURCE_READER_METRIC_GROUP);
+ this.commitsSucceeded =
+ this.kafkaSourceReaderMetricGroup.counter(COMMITS_SUCCEEDED_METRIC_COUNTER);
+ this.commitsFailed =
+ this.kafkaSourceReaderMetricGroup.counter(COMMITS_FAILED_METRIC_COUNTER);
+ }
+
+ /**
+ * Register metrics of KafkaConsumer in Kafka metric group.
+ *
+ * @param kafkaConsumer Kafka consumer used by partition split reader.
+ */
+ @SuppressWarnings("Convert2MethodRef")
+ public void registerKafkaConsumerMetrics(KafkaConsumer<?, ?> kafkaConsumer) {
+ final Map<MetricName, ? extends Metric> kafkaConsumerMetrics = kafkaConsumer.metrics();
+ if (kafkaConsumerMetrics == null) {
+ LOG.warn("Consumer implementation does not support metrics");
+ return;
+ }
+
+ final MetricGroup kafkaConsumerMetricGroup =
+ kafkaSourceReaderMetricGroup.addGroup(KAFKA_CONSUMER_METRIC_GROUP);
+
+ kafkaConsumerMetrics.forEach(
+ (name, metric) ->
+ kafkaConsumerMetricGroup.gauge(name.name(), () -> metric.metricValue()));
+ }
+
+ /**
+ * Register metric groups for the given {@link TopicPartition}.
+ *
+ * @param tp Registering topic partition
+ */
+ public void registerTopicPartition(TopicPartition tp) {
+ offsets.put(tp, new Offset(INITIAL_OFFSET, INITIAL_OFFSET));
+ registerOffsetMetricsForTopicPartition(tp);
+ }
+
+ /**
+ * Update current consuming offset of the given {@link TopicPartition}.
+ *
+ * @param tp Updating topic partition
+ * @param offset Current consuming offset
+ */
+ public void recordCurrentOffset(TopicPartition tp, long offset) {
+ checkTopicPartitionTracked(tp);
+ offsets.get(tp).currentOffset = offset;
+ }
+
+ /**
+ * Update the latest committed offset of the given {@link TopicPartition}.
+ *
+ * @param tp Updating topic partition
+ * @param offset Committing offset
+ */
+ public void recordCommittedOffset(TopicPartition tp, long offset) {
+ checkTopicPartitionTracked(tp);
+ commitsSucceeded.inc();
+ offsets.get(tp).committedOffset = offset;
+ }
+
+ /** Mark a failure commit. */
+ public void recordFailedCommit() {
+ commitsFailed.inc();
+ }
+
+ // -------- Helper functions --------
+ private void registerOffsetMetricsForTopicPartition(TopicPartition tp) {
+ final MetricGroup topicPartitionGroup =
+ this.kafkaSourceReaderMetricGroup
+ .addGroup(TOPIC_GROUP, tp.topic())
+ .addGroup(PARTITION_GROUP, String.valueOf(tp.partition()));
+ topicPartitionGroup.gauge(
+ CURRENT_OFFSET_METRIC_GAUGE,
+ () ->
+ offsets.getOrDefault(tp, new Offset(INITIAL_OFFSET, INITIAL_OFFSET))
+ .currentOffset);
+ topicPartitionGroup.gauge(
+ COMMITTED_OFFSET_METRIC_GAUGE,
+ () ->
+ offsets.getOrDefault(tp, new Offset(INITIAL_OFFSET, INITIAL_OFFSET))
+ .committedOffset);
+ }
+
+ private void checkTopicPartitionTracked(TopicPartition tp) {
+ if (!offsets.containsKey(tp)) {
+ throw new IllegalArgumentException(
+ String.format("TopicPartition %s is not tracked", tp));
+ }
+ }
+
+ private static class Offset {
+ long currentOffset;
+ long committedOffset;
+
+ Offset(long currentOffset, long committedOffset) {
+ this.currentOffset = currentOffset;
+ this.committedOffset = committedOffset;
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
index ee88629..e5e1425 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java
@@ -24,6 +24,7 @@ import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
import org.apache.flink.connector.kafka.source.KafkaSourceOptions;
+import org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
import org.apache.flink.util.Collector;
@@ -74,12 +75,15 @@ public class KafkaPartitionSplitReader<T>
private final SimpleCollector<T> collector;
private final String groupId;
private final int subtaskId;
+ private final KafkaSourceReaderMetrics kafkaSourceReaderMetrics;
public KafkaPartitionSplitReader(
Properties props,
KafkaRecordDeserializationSchema<T> deserializationSchema,
- int subtaskId) {
+ int subtaskId,
+ KafkaSourceReaderMetrics kafkaSourceReaderMetrics) {
this.subtaskId = subtaskId;
+ this.kafkaSourceReaderMetrics = kafkaSourceReaderMetrics;
Properties consumerProps = new Properties();
consumerProps.putAll(props);
consumerProps.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, createConsumerClientId(props));
@@ -88,6 +92,7 @@ public class KafkaPartitionSplitReader<T>
this.deserializationSchema = deserializationSchema;
this.collector = new SimpleCollector<>();
this.groupId = consumerProps.getProperty(ConsumerConfig.GROUP_ID_CONFIG);
+ maybeRegisterKafkaConsumerMetrics(props, kafkaSourceReaderMetrics, consumer);
}
@Override
@@ -108,7 +113,9 @@ public class KafkaPartitionSplitReader<T>
String splitId = tp.toString();
Collection<Tuple3<T, Long, Long>> recordsForSplit =
recordsBySplits.recordsForSplit(splitId);
- for (ConsumerRecord<byte[], byte[]> consumerRecord : consumerRecords.records(tp)) {
+ final List<ConsumerRecord<byte[], byte[]>> recordsFromPartition =
+ consumerRecords.records(tp);
+ for (ConsumerRecord<byte[], byte[]> consumerRecord : recordsFromPartition) {
// Stop consuming from this partition if the offsets has reached the stopping
// offset.
// Note that there are two cases, either case finishes a split:
@@ -160,6 +167,12 @@ public class KafkaPartitionSplitReader<T>
collector.reset();
}
}
+
+ // Use the last record for updating offset metrics
+ if (recordsFromPartition.size() > 0) {
+ kafkaSourceReaderMetrics.recordCurrentOffset(
+ tp, recordsFromPartition.get(recordsFromPartition.size() - 1).offset());
+ }
}
// Unassign the partitions that has finished.
if (!finishedPartitions.isEmpty()) {
@@ -202,6 +215,8 @@ public class KafkaPartitionSplitReader<T>
partitionsStartingFromSpecifiedOffsets);
parseStoppingOffsets(
s, partitionsStoppingAtLatest, partitionsStoppingAtCommitted);
+ // Track the new topic partition in metrics
+ kafkaSourceReaderMetrics.registerTopicPartition(s.getTopicPartition());
});
// Assign new partitions.
@@ -385,6 +400,20 @@ public class KafkaPartitionSplitReader<T>
return stoppingOffsets.getOrDefault(tp, Long.MAX_VALUE);
}
+ private void maybeRegisterKafkaConsumerMetrics(
+ Properties props,
+ KafkaSourceReaderMetrics kafkaSourceReaderMetrics,
+ KafkaConsumer<?, ?> consumer) {
+ final Boolean needToRegister =
+ KafkaSourceOptions.getOption(
+ props,
+ KafkaSourceOptions.REGISTER_KAFKA_CONSUMER_METRICS,
+ Boolean::parseBoolean);
+ if (needToRegister) {
+ kafkaSourceReaderMetrics.registerKafkaConsumerMetrics(consumer);
+ }
+ }
+
// ---------------- private helper class ------------------------
private static class KafkaPartitionSplitRecords<T> implements RecordsWithSplitIds<T> {
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
index bf3d42e..3a00be5 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
@@ -26,6 +26,7 @@ import org.apache.flink.connector.base.source.reader.RecordEmitter;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.connector.base.source.reader.SingleThreadMultiplexSourceReaderBase;
import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+import org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics;
import org.apache.flink.connector.kafka.source.reader.fetcher.KafkaSourceFetcherManager;
import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplitState;
@@ -54,13 +55,15 @@ public class KafkaSourceReader<T>
// and the split fetcher thread in the callback.
private final SortedMap<Long, Map<TopicPartition, OffsetAndMetadata>> offsetsToCommit;
private final ConcurrentMap<TopicPartition, OffsetAndMetadata> offsetsOfFinishedSplits;
+ private final KafkaSourceReaderMetrics kafkaSourceReaderMetrics;
public KafkaSourceReader(
FutureCompletingBlockingQueue<RecordsWithSplitIds<Tuple3<T, Long, Long>>> elementsQueue,
Supplier<KafkaPartitionSplitReader<T>> splitReaderSupplier,
RecordEmitter<Tuple3<T, Long, Long>, T, KafkaPartitionSplitState> recordEmitter,
Configuration config,
- SourceReaderContext context) {
+ SourceReaderContext context,
+ KafkaSourceReaderMetrics kafkaSourceReaderMetrics) {
super(
elementsQueue,
new KafkaSourceFetcherManager<>(elementsQueue, splitReaderSupplier::get),
@@ -69,6 +72,7 @@ public class KafkaSourceReader<T>
context);
this.offsetsToCommit = Collections.synchronizedSortedMap(new TreeMap<>());
this.offsetsOfFinishedSplits = new ConcurrentHashMap<>();
+ this.kafkaSourceReaderMetrics = kafkaSourceReaderMetrics;
}
@Override
@@ -117,6 +121,7 @@ public class KafkaSourceReader<T>
// The offset commit here is needed by the external monitoring. It won't
// break Flink job's correctness if we fail to commit the offset here.
if (e != null) {
+ kafkaSourceReaderMetrics.recordFailedCommit();
LOG.warn(
"Failed to commit consumer offsets for checkpoint {}",
checkpointId,
@@ -129,6 +134,10 @@ public class KafkaSourceReader<T>
// from the offsets of the finished splits map.
Map<TopicPartition, OffsetAndMetadata> committedPartitions =
offsetsToCommit.get(checkpointId);
+ committedPartitions.forEach(
+ (tp, offset) ->
+ kafkaSourceReaderMetrics.recordCommittedOffset(
+ tp, offset.offset()));
offsetsOfFinishedSplits
.entrySet()
.removeIf(
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetricsTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetricsTest.java
new file mode 100644
index 0000000..c5dfdf3
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetricsTest.java
@@ -0,0 +1,166 @@
+/*
+ * 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.flink.connector.kafka.source.metrics;
+
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.testutils.MetricListener;
+
+import org.apache.kafka.common.TopicPartition;
+import org.junit.Test;
+
+import static org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics.PARTITION_GROUP;
+import static org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics.TOPIC_GROUP;
+import static org.junit.Assert.assertEquals;
+
+/** Unit test for {@link KafkaSourceReaderMetrics}. */
+public class KafkaSourceReaderMetricsTest {
+
+ private static final TopicPartition FOO_0 = new TopicPartition("foo", 0);
+ private static final TopicPartition FOO_1 = new TopicPartition("foo", 1);
+ private static final TopicPartition BAR_0 = new TopicPartition("bar", 0);
+ private static final TopicPartition BAR_1 = new TopicPartition("bar", 1);
+
+ @Test
+ public void testCurrentOffsetTracking() {
+ MetricListener metricListener = new MetricListener();
+
+ final KafkaSourceReaderMetrics kafkaSourceReaderMetrics =
+ new KafkaSourceReaderMetrics(metricListener.getMetricGroup());
+
+ kafkaSourceReaderMetrics.registerTopicPartition(FOO_0);
+ kafkaSourceReaderMetrics.registerTopicPartition(FOO_1);
+ kafkaSourceReaderMetrics.registerTopicPartition(BAR_0);
+ kafkaSourceReaderMetrics.registerTopicPartition(BAR_1);
+
+ kafkaSourceReaderMetrics.recordCurrentOffset(FOO_0, 15213L);
+ kafkaSourceReaderMetrics.recordCurrentOffset(FOO_1, 18213L);
+ kafkaSourceReaderMetrics.recordCurrentOffset(BAR_0, 18613L);
+ kafkaSourceReaderMetrics.recordCurrentOffset(BAR_1, 15513L);
+
+ assertCurrentOffset(FOO_0, 15213L, metricListener);
+ assertCurrentOffset(FOO_1, 18213L, metricListener);
+ assertCurrentOffset(BAR_0, 18613L, metricListener);
+ assertCurrentOffset(BAR_1, 15513L, metricListener);
+ }
+
+ @Test
+ public void testCommitOffsetTracking() {
+ MetricListener metricListener = new MetricListener();
+
+ final KafkaSourceReaderMetrics kafkaSourceReaderMetrics =
+ new KafkaSourceReaderMetrics(metricListener.getMetricGroup());
+
+ kafkaSourceReaderMetrics.registerTopicPartition(FOO_0);
+ kafkaSourceReaderMetrics.registerTopicPartition(FOO_1);
+ kafkaSourceReaderMetrics.registerTopicPartition(BAR_0);
+ kafkaSourceReaderMetrics.registerTopicPartition(BAR_1);
+
+ kafkaSourceReaderMetrics.recordCommittedOffset(FOO_0, 15213L);
+ kafkaSourceReaderMetrics.recordCommittedOffset(FOO_1, 18213L);
+ kafkaSourceReaderMetrics.recordCommittedOffset(BAR_0, 18613L);
+ kafkaSourceReaderMetrics.recordCommittedOffset(BAR_1, 15513L);
+
+ assertCommittedOffset(FOO_0, 15213L, metricListener);
+ assertCommittedOffset(FOO_1, 18213L, metricListener);
+ assertCommittedOffset(BAR_0, 18613L, metricListener);
+ assertCommittedOffset(BAR_1, 15513L, metricListener);
+
+ assertEquals(
+ 4L,
+ metricListener
+ .getCounter(
+ KafkaSourceReaderMetrics.KAFKA_SOURCE_READER_METRIC_GROUP,
+ KafkaSourceReaderMetrics.COMMITS_SUCCEEDED_METRIC_COUNTER)
+ .getCount());
+ }
+
+ @Test
+ public void testNonTrackingTopicPartition() {
+ MetricListener metricListener = new MetricListener();
+ final KafkaSourceReaderMetrics kafkaSourceReaderMetrics =
+ new KafkaSourceReaderMetrics(metricListener.getMetricGroup());
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> kafkaSourceReaderMetrics.recordCurrentOffset(FOO_0, 15213L));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> kafkaSourceReaderMetrics.recordCommittedOffset(FOO_0, 15213L));
+ }
+
+ @Test
+ public void testFailedCommit() {
+ MetricListener metricListener = new MetricListener();
+ final KafkaSourceReaderMetrics kafkaSourceReaderMetrics =
+ new KafkaSourceReaderMetrics(metricListener.getMetricGroup());
+ kafkaSourceReaderMetrics.recordFailedCommit();
+ assertEquals(
+ 1L,
+ metricListener
+ .getCounter(
+ KafkaSourceReaderMetrics.KAFKA_SOURCE_READER_METRIC_GROUP,
+ KafkaSourceReaderMetrics.COMMITS_FAILED_METRIC_COUNTER)
+ .getCount());
+ }
+
+ // ----------- Assertions --------------
+
+ private void assertCurrentOffset(
+ TopicPartition tp, long expectedOffset, MetricListener metricListener) {
+ assertGaugeValueEquals(
+ expectedOffset,
+ metricListener.getGauge(
+ KafkaSourceReaderMetrics.KAFKA_SOURCE_READER_METRIC_GROUP,
+ TOPIC_GROUP,
+ tp.topic(),
+ PARTITION_GROUP,
+ String.valueOf(tp.partition()),
+ KafkaSourceReaderMetrics.CURRENT_OFFSET_METRIC_GAUGE),
+ Long.class);
+ }
+
+ private void assertCommittedOffset(
+ TopicPartition tp, long expectedOffset, MetricListener metricListener) {
+ assertGaugeValueEquals(
+ expectedOffset,
+ metricListener.getGauge(
+ KafkaSourceReaderMetrics.KAFKA_SOURCE_READER_METRIC_GROUP,
+ TOPIC_GROUP,
+ tp.topic(),
+ PARTITION_GROUP,
+ String.valueOf(tp.partition()),
+ KafkaSourceReaderMetrics.COMMITTED_OFFSET_METRIC_GAUGE),
+ Long.class);
+ }
+
+ private <T> void assertGaugeValueEquals(T expected, Gauge<?> gauge, Class<T> type) {
+ final T actual = type.cast(gauge.getValue());
+ assertEquals(expected, actual);
+ }
+
+ private void assertThrows(Class<? extends Throwable> clazz, Runnable runnable) {
+ try {
+ runnable.run();
+ } catch (Throwable e) {
+ if (!clazz.isAssignableFrom(e.getClass())) {
+ throw new IllegalArgumentException(
+ "Unexpected exception thrown: " + e.getClass().getName());
+ }
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
index e0f011d..2a477dd 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
@@ -23,9 +23,11 @@ import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
import org.apache.flink.connector.kafka.source.KafkaSourceTestEnv;
+import org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
import org.apache.flink.connector.testutils.source.deserialization.TestingDeserializationContext;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.TopicPartition;
@@ -178,7 +180,11 @@ public class KafkaPartitionSplitReaderTest {
KafkaRecordDeserializationSchema<Integer> deserializationSchema =
KafkaRecordDeserializationSchema.valueOnly(IntegerDeserializer.class);
deserializationSchema.open(new TestingDeserializationContext());
- return new KafkaPartitionSplitReader<>(props, deserializationSchema, 0);
+ KafkaSourceReaderMetrics kafkaSourceReaderMetrics =
+ new KafkaSourceReaderMetrics(
+ UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup());
+ return new KafkaPartitionSplitReader<>(
+ props, deserializationSchema, 0, kafkaSourceReaderMetrics);
}
private Map<String, KafkaPartitionSplit> assignSplits(
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
index 579bab3..18a024b 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
@@ -21,6 +21,8 @@ package org.apache.flink.connector.kafka.source.reader;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.ReaderOutput;
import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.KafkaSourceBuilder;
import org.apache.flink.connector.kafka.source.KafkaSourceTestEnv;
@@ -31,6 +33,8 @@ import org.apache.flink.connector.testutils.source.reader.SourceReaderTestBase;
import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.testutils.MetricListener;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;
@@ -52,6 +56,14 @@ import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
+import static org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics.COMMITS_SUCCEEDED_METRIC_COUNTER;
+import static org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics.COMMITTED_OFFSET_METRIC_GAUGE;
+import static org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics.CURRENT_OFFSET_METRIC_GAUGE;
+import static org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics.INITIAL_OFFSET;
+import static org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics.KAFKA_CONSUMER_METRIC_GROUP;
+import static org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics.KAFKA_SOURCE_READER_METRIC_GROUP;
+import static org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics.PARTITION_GROUP;
+import static org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics.TOPIC_GROUP;
import static org.apache.flink.core.testutils.CommonTestUtils.waitUtil;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -221,6 +233,69 @@ public class KafkaSourceReaderTest extends SourceReaderTestBase<KafkaPartitionSp
}
}
+ @Test
+ public void testKafkaSourceMetrics() throws Exception {
+ final MetricListener metricListener = new MetricListener();
+ final String groupId = "testKafkaSourceMetrics";
+ final TopicPartition tp = new TopicPartition(TOPIC, 0);
+
+ try (KafkaSourceReader<Integer> reader =
+ (KafkaSourceReader<Integer>)
+ createReader(
+ Boundedness.CONTINUOUS_UNBOUNDED,
+ groupId,
+ metricListener.getMetricGroup())) {
+
+ KafkaPartitionSplit split =
+ new KafkaPartitionSplit(tp, KafkaPartitionSplit.EARLIEST_OFFSET);
+ reader.addSplits(Collections.singletonList(split));
+
+ TestingReaderOutput<Integer> output = new TestingReaderOutput<>();
+ pollUntil(
+ reader,
+ output,
+ () -> output.getEmittedRecords().size() == NUM_RECORDS_PER_SPLIT,
+ String.format(
+ "Failed to poll %d records until timeout", NUM_RECORDS_PER_SPLIT));
+
+ // Metric "records-consumed-total" of KafkaConsumer should be NUM_RECORDS_PER_SPLIT
+ assertEquals(
+ NUM_RECORDS_PER_SPLIT,
+ getKafkaConsumerMetric("records-consumed-total", metricListener));
+
+ // Current consuming offset should be NUM_RECORD_PER_SPLIT - 1
+ assertEquals(NUM_RECORDS_PER_SPLIT - 1, getCurrentOffsetMetric(tp, metricListener));
+
+ // No offset is committed till now
+ assertEquals(INITIAL_OFFSET, getCommittedOffsetMetric(tp, metricListener));
+
+ // Trigger offset commit
+ reader.snapshotState(15213L);
+ reader.notifyCheckpointComplete(15213L);
+ waitUtil(
+ () -> reader.getOffsetsToCommit().isEmpty(),
+ Duration.ofSeconds(60),
+ String.format(
+ "Offsets are not committed successfully. Dangling offsets: %s",
+ reader.getOffsetsToCommit()));
+
+ // Metric "commit-total" of KafkaConsumer should be 1
+ assertEquals(1, getKafkaConsumerMetric("commit-total", metricListener));
+
+ // Committed offset should be NUM_RECORD_PER_SPLIT
+ assertEquals(NUM_RECORDS_PER_SPLIT, getCommittedOffsetMetric(tp, metricListener));
+
+ // Number of successful commits should be 1
+ assertEquals(
+ 1L,
+ metricListener
+ .getCounter(
+ KAFKA_SOURCE_READER_METRIC_GROUP,
+ COMMITS_SUCCEEDED_METRIC_COUNTER)
+ .getCount());
+ }
+ }
+
// ------------------------------------------
@Override
@@ -256,6 +331,17 @@ public class KafkaSourceReaderTest extends SourceReaderTestBase<KafkaPartitionSp
private SourceReader<Integer, KafkaPartitionSplit> createReader(
Boundedness boundedness, String groupId) throws Exception {
+ return createReader(boundedness, groupId, new TestingReaderContext());
+ }
+
+ private SourceReader<Integer, KafkaPartitionSplit> createReader(
+ Boundedness boundedness, String groupId, MetricGroup metricGroup) throws Exception {
+ return createReader(
+ boundedness, groupId, new TestingReaderContext(new Configuration(), metricGroup));
+ }
+
+ private SourceReader<Integer, KafkaPartitionSplit> createReader(
+ Boundedness boundedness, String groupId, SourceReaderContext context) throws Exception {
KafkaSourceBuilder<Integer> builder =
KafkaSource.<Integer>builder()
.setClientIdPrefix("KafkaSourceReaderTest")
@@ -273,7 +359,7 @@ public class KafkaSourceReaderTest extends SourceReaderTestBase<KafkaPartitionSp
builder.setBounded(OffsetsInitializer.latest());
}
- return builder.build().createReader(new TestingReaderContext());
+ return builder.build().createReader(context);
}
private void pollUntil(
@@ -298,6 +384,40 @@ public class KafkaSourceReaderTest extends SourceReaderTestBase<KafkaPartitionSp
errorMessage);
}
+ private long getKafkaConsumerMetric(String name, MetricListener listener) {
+ return ((Double)
+ listener.getGauge(
+ KAFKA_SOURCE_READER_METRIC_GROUP,
+ KAFKA_CONSUMER_METRIC_GROUP,
+ name)
+ .getValue())
+ .longValue();
+ }
+
+ private long getCurrentOffsetMetric(TopicPartition tp, MetricListener listener) {
+ return (long)
+ listener.getGauge(
+ KAFKA_SOURCE_READER_METRIC_GROUP,
+ TOPIC_GROUP,
+ tp.topic(),
+ PARTITION_GROUP,
+ String.valueOf(tp.partition()),
+ CURRENT_OFFSET_METRIC_GAUGE)
+ .getValue();
+ }
+
+ private long getCommittedOffsetMetric(TopicPartition tp, MetricListener listener) {
+ return (long)
+ listener.getGauge(
+ KAFKA_SOURCE_READER_METRIC_GROUP,
+ TOPIC_GROUP,
+ tp.topic(),
+ PARTITION_GROUP,
+ String.valueOf(tp.partition()),
+ COMMITTED_OFFSET_METRIC_GAUGE)
+ .getValue();
+ }
+
// ---------------------
private static List<ProducerRecord<String, Integer>> getRecords() {
diff --git a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java
index 50b48b2..eaca2e9 100644
--- a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java
+++ b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/source/reader/TestingReaderContext.java
@@ -32,7 +32,7 @@ import java.util.List;
/** A testing implementation of the {@link SourceReaderContext}. */
public class TestingReaderContext implements SourceReaderContext {
- private final UnregisteredMetricsGroup metrics = new UnregisteredMetricsGroup();
+ private final MetricGroup metrics;
private final Configuration config;
@@ -41,11 +41,12 @@ public class TestingReaderContext implements SourceReaderContext {
private int numSplitRequests;
public TestingReaderContext() {
- this(new Configuration());
+ this(new Configuration(), new UnregisteredMetricsGroup());
}
- public TestingReaderContext(Configuration config) {
+ public TestingReaderContext(Configuration config, MetricGroup metricGroup) {
this.config = config;
+ this.metrics = metricGroup;
}
// ------------------------------------------------------------------------