You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/06/09 03:08:09 UTC
[GitHub] [flink] PatrickRen opened a new pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
PatrickRen opened a new pull request #16115:
URL: https://github.com/apache/flink/pull/16115
## What is the purpose of the change
This pull request adds offset metrics for Kafka source reader, and register metrics of Kafka consumer in Flink metric group
## Brief change log
- Add ```KafkaSourceReaderMetrics``` for tracking metrics reported in KafkaSourceReader
- Register current consuming offset and committed offset metric
- Register KafkaConsumer's metrics in operator metric group
- Add a KafkaSourceOption to configure whether to register KafkaConsumer metrics
## Verifying this change
This change added tests and can be verified as follows:
- Added unit test for the new ```KafkaSourceReaderMetrics```
- Added integration test for KafkaSourceReader to verify new metrics are reported correctly
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (docs)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857350673
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809",
"triggerID" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"triggerType" : "PUSH"
}, {
"hash" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 64d70892f99ab436c6bf52012e124874a11cbb18 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809)
* 4c653575cd66258b4cdd0e54b8cf7dfeca474f9b UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857337927
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 64d70892f99ab436c6bf52012e124874a11cbb18 (Wed Jun 09 03:11:18 UTC 2021)
**Warnings:**
* Documentation files were touched, but no `docs/content.zh/` files: Update Chinese documentation or file Jira ticket.
* **This pull request references an unassigned [Jira ticket](https://issues.apache.org/jira/browse/FLINK-22766).** According to the [code contribution guide](https://flink.apache.org/contributing/contribute-code.html), tickets need to be assigned before starting with the implementation work.
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] cuspymd commented on a change in pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
cuspymd commented on a change in pull request #16115:
URL: https://github.com/apache/flink/pull/16115#discussion_r648003881
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetricsListener.java
##########
@@ -0,0 +1,174 @@
+/*
+ * 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.CharacterFilter;
+import org.apache.flink.metrics.Metric;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.groups.OperatorMetricGroup;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.metrics.scope.ScopeFormat;
+import org.apache.flink.runtime.metrics.util.TestingMetricRegistry;
+
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** A utilization class for mocking {@link OperatorMetricGroup} and collecting reported metrics. */
+public class KafkaSourceReaderMetricsListener {
+ public static final String OPERATOR_NAME = "operator_metric_mocker";
+ public static final char DELIMITER = '.';
+
+ Map<String, Metric> kafkaSourceReaderMetrics = new HashMap<>();
+ Map<String, Metric> kafkaConsumerMetrics = new HashMap<>();
+
+ private final OperatorMetricGroup metricGroup;
+
+ public KafkaSourceReaderMetricsListener() {
+ TestingMetricRegistry registry =
+ TestingMetricRegistry.builder()
+ .setDelimiter(DELIMITER)
+ .setRegisterConsumer(
+ (metric, metricName, group) -> {
+ String parentGroupScope =
+ group.getLogicalScope(CharacterFilter.NO_OP_FILTER);
+
+ // We only care about Kafka source reader metrics
+ if (isKafkaSourceReaderMetric(parentGroupScope)) {
+
+ // KafkaConsumer metric
+ if (isKafkaConsumerMetric(parentGroupScope)) {
+ kafkaConsumerMetrics.put(metricName, metric);
+ return;
+ }
+
+ final String topic =
+ group.getAllVariables()
+ .get(
+ ScopeFormat.asVariable(
+ KafkaSourceReaderMetrics
+ .TOPIC_GROUP));
+ final String partition =
+ group.getAllVariables()
+ .get(
+ ScopeFormat.asVariable(
+ KafkaSourceReaderMetrics
+ .PARTITION_GROUP));
+
+ if (topic == null && partition == null) {
+ // General metric
+ kafkaSourceReaderMetrics.put(metricName, metric);
+ } else {
+ // topic partition specific metric
+ kafkaSourceReaderMetrics.put(
+ String.join(
+ String.valueOf(DELIMITER),
+ topic,
+ partition,
+ metricName),
+ metric);
+ }
+ }
+ })
Review comment:
It seems that the code length is too long to express with lambda. Wouldn't it be better to separate it into a function?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857350673
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] PatrickRen commented on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
PatrickRen commented on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-870277421
Thanks @becketqin for reviewing~ I split the patch into two commits:
2b460b5 for adding test util class `MetricListener`
43036001425b for offset and KafkaConsumer metrics
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] PatrickRen commented on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
PatrickRen commented on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-870277421
Thanks @becketqin for reviewing~ I split the patch into two commits:
2b460b5 for adding test util class `MetricListener`
43036001425b for offset and KafkaConsumer metrics
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] becketqin commented on a change in pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
becketqin commented on a change in pull request #16115:
URL: https://github.com/apache/flink/pull/16115#discussion_r659446873
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetrics.java
##########
@@ -0,0 +1,168 @@
+/*
+ * 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>Hierarchy
+ */
+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 = 0;
+
+ // 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 consuming(TopicPartition tp, long offset) {
+ checkTopicPartitionTracked(tp);
+ offsets.get(tp).currentOffset = offset;
+ }
+
+ /**
+ * Update current committing offset of the given {@link TopicPartition}.
+ *
+ * @param tp Updating topic partition
+ * @param offset Committing offset
+ */
+ public void committing(TopicPartition tp, long offset) {
Review comment:
Ditto above
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetrics.java
##########
@@ -0,0 +1,168 @@
+/*
+ * 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>Hierarchy
+ */
+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 = 0;
Review comment:
Should this be set to -1 instead of 0 to avoid confusion?
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetrics.java
##########
@@ -0,0 +1,168 @@
+/*
+ * 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>Hierarchy
+ */
+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 = 0;
+
+ // 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 consuming(TopicPartition tp, long offset) {
Review comment:
Can we make the name more explicit, e.g. `recordConsumingOffset()`
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetrics.java
##########
@@ -0,0 +1,168 @@
+/*
+ * 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>Hierarchy
+ */
+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 = 0;
+
+ // 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 consuming(TopicPartition tp, long offset) {
+ checkTopicPartitionTracked(tp);
+ offsets.get(tp).currentOffset = offset;
+ }
+
+ /**
+ * Update current committing offset of the given {@link TopicPartition}.
+ *
+ * @param tp Updating topic partition
+ * @param offset Committing offset
+ */
+ public void committing(TopicPartition tp, long offset) {
+ checkTopicPartitionTracked(tp);
+ commitsSucceeded.inc();
+ offsets.get(tp).committedOffset = offset;
+ }
+
+ /** Mark a failure commit. */
+ public void failedCommitting() {
Review comment:
Ditto above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] becketqin merged pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
becketqin merged pull request #16115:
URL: https://github.com/apache/flink/pull/16115
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] PatrickRen commented on a change in pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
PatrickRen commented on a change in pull request #16115:
URL: https://github.com/apache/flink/pull/16115#discussion_r648045623
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetricsListener.java
##########
@@ -0,0 +1,174 @@
+/*
+ * 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.CharacterFilter;
+import org.apache.flink.metrics.Metric;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.groups.OperatorMetricGroup;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.metrics.scope.ScopeFormat;
+import org.apache.flink.runtime.metrics.util.TestingMetricRegistry;
+
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** A utilization class for mocking {@link OperatorMetricGroup} and collecting reported metrics. */
+public class KafkaSourceReaderMetricsListener {
+ public static final String OPERATOR_NAME = "operator_metric_mocker";
+ public static final char DELIMITER = '.';
+
+ Map<String, Metric> kafkaSourceReaderMetrics = new HashMap<>();
+ Map<String, Metric> kafkaConsumerMetrics = new HashMap<>();
+
+ private final OperatorMetricGroup metricGroup;
+
+ public KafkaSourceReaderMetricsListener() {
+ TestingMetricRegistry registry =
+ TestingMetricRegistry.builder()
+ .setDelimiter(DELIMITER)
+ .setRegisterConsumer(
+ (metric, metricName, group) -> {
+ String parentGroupScope =
+ group.getLogicalScope(CharacterFilter.NO_OP_FILTER);
+
+ // We only care about Kafka source reader metrics
+ if (isKafkaSourceReaderMetric(parentGroupScope)) {
+
+ // KafkaConsumer metric
+ if (isKafkaConsumerMetric(parentGroupScope)) {
+ kafkaConsumerMetrics.put(metricName, metric);
+ return;
+ }
+
+ final String topic =
+ group.getAllVariables()
+ .get(
+ ScopeFormat.asVariable(
+ KafkaSourceReaderMetrics
+ .TOPIC_GROUP));
+ final String partition =
+ group.getAllVariables()
+ .get(
+ ScopeFormat.asVariable(
+ KafkaSourceReaderMetrics
+ .PARTITION_GROUP));
+
+ if (topic == null && partition == null) {
+ // General metric
+ kafkaSourceReaderMetrics.put(metricName, metric);
+ } else {
+ // topic partition specific metric
+ kafkaSourceReaderMetrics.put(
+ String.join(
+ String.valueOf(DELIMITER),
+ topic,
+ partition,
+ metricName),
+ metric);
+ }
+ }
+ })
Review comment:
Thanks for your review Myoungdo! Fixed in 4c65357.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] PatrickRen commented on a change in pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
PatrickRen commented on a change in pull request #16115:
URL: https://github.com/apache/flink/pull/16115#discussion_r663619727
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetrics.java
##########
@@ -0,0 +1,168 @@
+/*
+ * 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>Hierarchy
+ */
+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 recordConsumingOffset(TopicPartition tp, long offset) {
Review comment:
Thanks for reviewing! Fixed in 258483e
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857350673
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809",
"triggerID" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 64d70892f99ab436c6bf52012e124874a11cbb18 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857350673
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809",
"triggerID" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"triggerType" : "PUSH"
}, {
"hash" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18826",
"triggerID" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4c653575cd66258b4cdd0e54b8cf7dfeca474f9b Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18826)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857350673
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809",
"triggerID" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"triggerType" : "PUSH"
}, {
"hash" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18826",
"triggerID" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"triggerType" : "PUSH"
}, {
"hash" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19659",
"triggerID" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"triggerType" : "PUSH"
}, {
"hash" : "258483e7d333af33187fb591294be801b0d3ab80",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19887",
"triggerID" : "258483e7d333af33187fb591294be801b0d3ab80",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 258483e7d333af33187fb591294be801b0d3ab80 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19887)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] PatrickRen commented on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
PatrickRen commented on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-859303486
Hi @becketqin, could you have a review on this PR? Thank~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857350673
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809",
"triggerID" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"triggerType" : "PUSH"
}, {
"hash" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18826",
"triggerID" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 64d70892f99ab436c6bf52012e124874a11cbb18 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809)
* 4c653575cd66258b4cdd0e54b8cf7dfeca474f9b Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18826)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857350673
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809",
"triggerID" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"triggerType" : "PUSH"
}, {
"hash" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18826",
"triggerID" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"triggerType" : "PUSH"
}, {
"hash" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19659",
"triggerID" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"triggerType" : "PUSH"
}, {
"hash" : "258483e7d333af33187fb591294be801b0d3ab80",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19887",
"triggerID" : "258483e7d333af33187fb591294be801b0d3ab80",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 43036001425b3cf3b995dc9d0bc1e5997922973c Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19659)
* 258483e7d333af33187fb591294be801b0d3ab80 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19887)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857337927
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 258483e7d333af33187fb591294be801b0d3ab80 (Thu Sep 23 17:55:29 UTC 2021)
**Warnings:**
* Documentation files were touched, but no `docs/content.zh/` files: Update Chinese documentation or file Jira ticket.
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857350673
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809",
"triggerID" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"triggerType" : "PUSH"
}, {
"hash" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18826",
"triggerID" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"triggerType" : "PUSH"
}, {
"hash" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19659",
"triggerID" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"triggerType" : "PUSH"
}, {
"hash" : "258483e7d333af33187fb591294be801b0d3ab80",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "258483e7d333af33187fb591294be801b0d3ab80",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 43036001425b3cf3b995dc9d0bc1e5997922973c Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19659)
* 258483e7d333af33187fb591294be801b0d3ab80 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] becketqin commented on a change in pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
becketqin commented on a change in pull request #16115:
URL: https://github.com/apache/flink/pull/16115#discussion_r663611789
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/metrics/KafkaSourceReaderMetrics.java
##########
@@ -0,0 +1,168 @@
+/*
+ * 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>Hierarchy
+ */
+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 recordConsumingOffset(TopicPartition tp, long offset) {
Review comment:
Can we avoid having different naming between the metric name and the method name? For example, it would be good to have `recordCurrentOffset()` rather than `recordConsumingOffset()`. Same applies to `recordCommittingOffset()` which would ideally be `recordCommittedOffset()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] becketqin commented on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
becketqin commented on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-873922951
@PatrickRen Thanks for the patch.
Merged to master: b094a932845db5539fc07b032d49d0bcefd15df2
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857350673
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809",
"triggerID" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"triggerType" : "PUSH"
}, {
"hash" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18826",
"triggerID" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"triggerType" : "PUSH"
}, {
"hash" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4c653575cd66258b4cdd0e54b8cf7dfeca474f9b Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18826)
* 43036001425b3cf3b995dc9d0bc1e5997922973c UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857350673
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809",
"triggerID" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"triggerType" : "PUSH"
}, {
"hash" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18826",
"triggerID" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"triggerType" : "PUSH"
}, {
"hash" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19659",
"triggerID" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"triggerType" : "PUSH"
}, {
"hash" : "258483e7d333af33187fb591294be801b0d3ab80",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19887",
"triggerID" : "258483e7d333af33187fb591294be801b0d3ab80",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 258483e7d333af33187fb591294be801b0d3ab80 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19887)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857350673
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 64d70892f99ab436c6bf52012e124874a11cbb18 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] becketqin merged pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
becketqin merged pull request #16115:
URL: https://github.com/apache/flink/pull/16115
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857350673
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809",
"triggerID" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"triggerType" : "PUSH"
}, {
"hash" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18826",
"triggerID" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"triggerType" : "PUSH"
}, {
"hash" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19659",
"triggerID" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 43036001425b3cf3b995dc9d0bc1e5997922973c Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19659)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-857350673
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18809",
"triggerID" : "64d70892f99ab436c6bf52012e124874a11cbb18",
"triggerType" : "PUSH"
}, {
"hash" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18826",
"triggerID" : "4c653575cd66258b4cdd0e54b8cf7dfeca474f9b",
"triggerType" : "PUSH"
}, {
"hash" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19659",
"triggerID" : "43036001425b3cf3b995dc9d0bc1e5997922973c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4c653575cd66258b4cdd0e54b8cf7dfeca474f9b Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18826)
* 43036001425b3cf3b995dc9d0bc1e5997922973c Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=19659)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] becketqin commented on pull request #16115: [FLINK-22766][connector/kafka] Report offsets and Kafka consumer metrics in Flink metric group
Posted by GitBox <gi...@apache.org>.
becketqin commented on pull request #16115:
URL: https://github.com/apache/flink/pull/16115#issuecomment-873922951
@PatrickRen Thanks for the patch.
Merged to master: b094a932845db5539fc07b032d49d0bcefd15df2
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org