You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "jeffkbkim (via GitHub)" <gi...@apache.org> on 2023/11/28 00:54:24 UTC
[PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
jeffkbkim opened a new pull request, #14848:
URL: https://github.com/apache/kafka/pull/14848
Move the following metrics from Yammer to Kafka metrics to continue with the migration from yammer to kafka metrics. These are not exposed and newly created metrics so there are no compatibility issues.
* NumConsumerGroups (yammer) -> consumer-groups-size (kafka)
* NumConsumerGroupsEmpty (yammer) -> empty-consumer-groups-size (kafka)
* NumConsumerGroupsAssigning (yammer) -> assigning-consumer-groups-size (kafka)
* NumConsumerGroupsReconciling (yammer) -> reconciling-consumer-groups-size (kafka)
* NumConsumerGroupsStable (yammer) -> stable-consumer-groups-size (kafka)
* NumConsumerGroupsDead (yammer) -> dead-consumer-groups-size (kafka)
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1425608080
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -174,75 +151,91 @@ public GroupCoordinatorMetrics(MetricsRegistry registry, Metrics metrics) {
));
}
- public Long numOffsets() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_OFFSETS)).sum();
+ private Long numOffsets() {
+ return shards.values().stream().mapToLong(GroupCoordinatorMetricsShard::numOffsets).sum();
}
- public Long numGenericGroups() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_GENERIC_GROUPS)).sum();
+ private Long numGenericGroups() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(null)).sum();
}
- public Long numGenericGroupsPreparingRebalanceCount() {
- return NUM_GENERIC_GROUPS_PREPARING_REBALANCE_COUNTER.get();
+ private Long numGenericGroupsPreparingRebalance() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.PREPARING_REBALANCE)).sum();
}
- public Long numGenericGroupsCompletingRebalanceCount() {
- return NUM_GENERIC_GROUPS_COMPLETING_REBALANCE_COUNTER.get();
+ private Long numGenericGroupsCompletingRebalance() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.COMPLETING_REBALANCE)).sum();
}
- public Long numGenericGroupsStableCount() {
- return NUM_GENERIC_GROUPS_STABLE_COUNTER.get();
+ private Long numGenericGroupsStable() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.STABLE)).sum();
}
- public Long numGenericGroupsDeadCount() {
- return NUM_GENERIC_GROUPS_DEAD_COUNTER.get();
+ private Long numGenericGroupsDead() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.DEAD)).sum();
}
- public Long numGenericGroupsEmptyCount() {
- return NUM_GENERIC_GROUPS_EMPTY_COUNTER.get();
+ private Long numGenericGroupsEmpty() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.EMPTY)).sum();
}
- public long numConsumerGroups() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_CONSUMER_GROUPS)).sum();
+ private long numConsumerGroups() {
+ return shards.values().stream().mapToLong(shard -> shard.numConsumerGroups(null)).sum();
}
- public long numConsumerGroupsEmpty() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_CONSUMER_GROUPS_EMPTY)).sum();
+ private long numConsumerGroupsEmpty() {
+ return shards.values().stream().mapToLong(shard -> shard.numConsumerGroups(ConsumerGroup.ConsumerGroupState.EMPTY)).sum();
}
- public long numConsumerGroupsAssigning() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_CONSUMER_GROUPS_ASSIGNING)).sum();
+ private long numConsumerGroupsAssigning() {
+ return shards.values().stream().mapToLong(shard -> shard.numConsumerGroups(ConsumerGroup.ConsumerGroupState.ASSIGNING)).sum();
}
- public long numConsumerGroupsReconciling() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_CONSUMER_GROUPS_RECONCILING)).sum();
+ private long numConsumerGroupsReconciling() {
+ return shards.values().stream().mapToLong(shard -> shard.numConsumerGroups(ConsumerGroup.ConsumerGroupState.RECONCILING)).sum();
}
- public long numConsumerGroupsStable() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_CONSUMER_GROUPS_STABLE)).sum();
+ private long numConsumerGroupsStable() {
+ return shards.values().stream().mapToLong(shard -> shard.numConsumerGroups(ConsumerGroup.ConsumerGroupState.STABLE)).sum();
}
- public long numConsumerGroupsDead() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_CONSUMER_GROUPS_DEAD)).sum();
+ private long numConsumerGroupsDead() {
+ return shards.values().stream().mapToLong(shard -> shard.numConsumerGroups(ConsumerGroup.ConsumerGroupState.DEAD)).sum();
}
@Override
public void close() {
Arrays.asList(
NUM_OFFSETS,
- NUM_GENERIC_GROUPS,
NUM_GENERIC_GROUPS_PREPARING_REBALANCE,
NUM_GENERIC_GROUPS_COMPLETING_REBALANCE,
NUM_GENERIC_GROUPS_STABLE,
NUM_GENERIC_GROUPS_DEAD,
- NUM_GENERIC_GROUPS_EMPTY,
- NUM_CONSUMER_GROUPS,
- NUM_CONSUMER_GROUPS_EMPTY,
- NUM_CONSUMER_GROUPS_ASSIGNING,
- NUM_CONSUMER_GROUPS_RECONCILING,
- NUM_CONSUMER_GROUPS_STABLE,
- NUM_CONSUMER_GROUPS_DEAD
+ NUM_GENERIC_GROUPS_EMPTY
).forEach(registry::removeMetric);
+ metrics.removeMetric(metrics.metricName(CONSUMER_GROUPS_COUNT_METRIC_NAME, METRICS_GROUP));
Review Comment:
Have you considered keeping a reference to the metric names instead of re-creating names here? This may simplify the logic to close them.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1424175673
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -29,44 +28,75 @@
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.ConcurrentHashMap;
/**
* These are the metrics which are managed by the {@link org.apache.kafka.coordinator.group.GroupMetadataManager} class.
* They generally pertain to aspects of group management, such as the number of groups in different states.
*/
public class GroupCoordinatorMetrics extends CoordinatorMetrics implements AutoCloseable {
+
+ enum GenericGroupState {
+ ALL("all"),
+ PREPARING_REBALANCE("PreparingRebalance"),
+ COMPLETING_REBALANCE("CompletingRebalance"),
+ STABLE("Stable"),
+ DEAD("Dead"),
+ EMPTY("Empty");
+
+ private final String name;
+
+ GenericGroupState(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public String toString() {
+ return name;
+ }
+ }
+
+ enum ConsumerGroupState {
+ ALL("all"),
+ EMPTY("empty"),
+ ASSIGNING("assigning"),
+ RECONCILING("reconciling"),
+ STABLE("stable"),
+ DEAD("dead");
+
+ private final String name;
+
+ ConsumerGroupState(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public String toString() {
+ return name;
+ }
+ }
+
public static final String METRICS_GROUP = "group-coordinator-metrics";
- public final static MetricName NUM_OFFSETS = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_OFFSETS = getMetricName(
"GroupMetadataManager", "NumOffsets");
- public final static MetricName NUM_GENERIC_GROUPS = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS = getMetricName(
"GroupMetadataManager", "NumGroups");
- public final static MetricName NUM_GENERIC_GROUPS_PREPARING_REBALANCE = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_PREPARING_REBALANCE = getMetricName(
"GroupMetadataManager", "NumGroupsPreparingRebalance");
- public final static MetricName NUM_GENERIC_GROUPS_COMPLETING_REBALANCE = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_COMPLETING_REBALANCE = getMetricName(
"GroupMetadataManager", "NumGroupsCompletingRebalance");
- public final static MetricName NUM_GENERIC_GROUPS_STABLE = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_STABLE = getMetricName(
"GroupMetadataManager", "NumGroupsStable");
- public final static MetricName NUM_GENERIC_GROUPS_DEAD = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_DEAD = getMetricName(
"GroupMetadataManager", "NumGroupsDead");
- public final static MetricName NUM_GENERIC_GROUPS_EMPTY = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_EMPTY = getMetricName(
"GroupMetadataManager", "NumGroupsEmpty");
- public final static MetricName NUM_CONSUMER_GROUPS = getMetricName(
- "GroupMetadataManager", "NumConsumerGroups");
- public final static MetricName NUM_CONSUMER_GROUPS_EMPTY = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsEmpty");
- public final static MetricName NUM_CONSUMER_GROUPS_ASSIGNING = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsAssigning");
- public final static MetricName NUM_CONSUMER_GROUPS_RECONCILING = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsReconciling");
- public final static MetricName NUM_CONSUMER_GROUPS_STABLE = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsStable");
- public final static MetricName NUM_CONSUMER_GROUPS_DEAD = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsDead");
+
+ public final static String NUM_CONSUMER_GROUPS_METRIC_NAME = "consumer-groups-size";
Review Comment:
should we rename the variables?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1424319049
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -77,29 +70,13 @@ public class GroupCoordinatorMetrics extends CoordinatorMetrics implements AutoC
private final MetricsRegistry registry;
private final Metrics metrics;
- private final Map<TopicPartition, CoordinatorMetricsShard> shards = new HashMap<>();
- private static final AtomicLong NUM_GENERIC_GROUPS_PREPARING_REBALANCE_COUNTER = new AtomicLong(0);
- private static final AtomicLong NUM_GENERIC_GROUPS_COMPLETING_REBALANCE_COUNTER = new AtomicLong(0);
- private static final AtomicLong NUM_GENERIC_GROUPS_STABLE_COUNTER = new AtomicLong(0);
- private static final AtomicLong NUM_GENERIC_GROUPS_DEAD_COUNTER = new AtomicLong(0);
- private static final AtomicLong NUM_GENERIC_GROUPS_EMPTY_COUNTER = new AtomicLong(0);
+ private final Map<TopicPartition, GroupCoordinatorMetricsShard> shards = new ConcurrentHashMap<>();
Review Comment:
we may have concurrent modification when the metric thread scrapes a shard while the shard is removed from `shards` on deactivate.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1426977801
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -281,100 +297,93 @@ public void onUpdateLastCommittedOffset(TopicPartition tp, long offset) {
}
}
- public static MetricName getMetricName(String type, String name) {
+ public static com.yammer.metrics.core.MetricName getMetricName(String type, String name) {
return getMetricName("kafka.coordinator.group", type, name);
}
private void registerGauges() {
- registry.newGauge(NUM_OFFSETS, new Gauge<Long>() {
+ registry.newGauge(NUM_OFFSETS, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
return numOffsets();
}
Review Comment:
Right, that does not work. I believe it's because yammer Gauge is not a functional interface
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1425610115
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -174,75 +151,91 @@ public GroupCoordinatorMetrics(MetricsRegistry registry, Metrics metrics) {
));
}
- public Long numOffsets() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_OFFSETS)).sum();
+ private Long numOffsets() {
+ return shards.values().stream().mapToLong(GroupCoordinatorMetricsShard::numOffsets).sum();
}
- public Long numGenericGroups() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_GENERIC_GROUPS)).sum();
+ private Long numGenericGroups() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(null)).sum();
}
- public Long numGenericGroupsPreparingRebalanceCount() {
- return NUM_GENERIC_GROUPS_PREPARING_REBALANCE_COUNTER.get();
+ private Long numGenericGroupsPreparingRebalance() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.PREPARING_REBALANCE)).sum();
}
- public Long numGenericGroupsCompletingRebalanceCount() {
- return NUM_GENERIC_GROUPS_COMPLETING_REBALANCE_COUNTER.get();
+ private Long numGenericGroupsCompletingRebalance() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.COMPLETING_REBALANCE)).sum();
}
- public Long numGenericGroupsStableCount() {
- return NUM_GENERIC_GROUPS_STABLE_COUNTER.get();
+ private Long numGenericGroupsStable() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.STABLE)).sum();
}
- public Long numGenericGroupsDeadCount() {
- return NUM_GENERIC_GROUPS_DEAD_COUNTER.get();
+ private Long numGenericGroupsDead() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.DEAD)).sum();
}
- public Long numGenericGroupsEmptyCount() {
- return NUM_GENERIC_GROUPS_EMPTY_COUNTER.get();
+ private Long numGenericGroupsEmpty() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.EMPTY)).sum();
}
- public long numConsumerGroups() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_CONSUMER_GROUPS)).sum();
+ private long numConsumerGroups() {
+ return shards.values().stream().mapToLong(shard -> shard.numConsumerGroups(null)).sum();
}
- public long numConsumerGroupsEmpty() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_CONSUMER_GROUPS_EMPTY)).sum();
+ private long numConsumerGroupsEmpty() {
+ return shards.values().stream().mapToLong(shard -> shard.numConsumerGroups(ConsumerGroup.ConsumerGroupState.EMPTY)).sum();
}
Review Comment:
nit: I wonder if we need all of these methods. They all look similar. Have you considered having just one which takes the state as an argument? Then, the metric could just call it with the correct one.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1425483590
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -16,57 +16,50 @@
*/
package org.apache.kafka.coordinator.group.metrics;
-import com.yammer.metrics.core.Gauge;
-import com.yammer.metrics.core.MetricName;
import com.yammer.metrics.core.MetricsRegistry;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.metrics.Gauge;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.metrics.stats.Meter;
import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.Group;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.generic.GenericGroupState;
import org.apache.kafka.server.metrics.KafkaYammerMetrics;
import org.apache.kafka.timeline.SnapshotRegistry;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.ConcurrentHashMap;
/**
* These are the metrics which are managed by the {@link org.apache.kafka.coordinator.group.GroupMetadataManager} class.
* They generally pertain to aspects of group management, such as the number of groups in different states.
*/
public class GroupCoordinatorMetrics extends CoordinatorMetrics implements AutoCloseable {
+
public static final String METRICS_GROUP = "group-coordinator-metrics";
- public final static MetricName NUM_OFFSETS = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_OFFSETS = getMetricName(
"GroupMetadataManager", "NumOffsets");
- public final static MetricName NUM_GENERIC_GROUPS = getMetricName(
- "GroupMetadataManager", "NumGroups");
Review Comment:
Where is this one now? We must keep all the old 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #14848:
URL: https://github.com/apache/kafka/pull/14848
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1422297921
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -29,44 +28,75 @@
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.ConcurrentHashMap;
/**
* These are the metrics which are managed by the {@link org.apache.kafka.coordinator.group.GroupMetadataManager} class.
* They generally pertain to aspects of group management, such as the number of groups in different states.
*/
public class GroupCoordinatorMetrics extends CoordinatorMetrics implements AutoCloseable {
+
+ enum GenericGroupState {
Review Comment:
Those two enums are really confusing because they duplicate existing ones. Could we just reuse the existing ones?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -29,44 +28,75 @@
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.ConcurrentHashMap;
/**
* These are the metrics which are managed by the {@link org.apache.kafka.coordinator.group.GroupMetadataManager} class.
* They generally pertain to aspects of group management, such as the number of groups in different states.
*/
public class GroupCoordinatorMetrics extends CoordinatorMetrics implements AutoCloseable {
+
+ enum GenericGroupState {
+ ALL("all"),
+ PREPARING_REBALANCE("PreparingRebalance"),
+ COMPLETING_REBALANCE("CompletingRebalance"),
+ STABLE("Stable"),
+ DEAD("Dead"),
+ EMPTY("Empty");
+
+ private final String name;
+
+ GenericGroupState(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public String toString() {
+ return name;
+ }
+ }
+
+ enum ConsumerGroupState {
+ ALL("all"),
+ EMPTY("empty"),
+ ASSIGNING("assigning"),
+ RECONCILING("reconciling"),
+ STABLE("stable"),
+ DEAD("dead");
+
+ private final String name;
+
+ ConsumerGroupState(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public String toString() {
+ return name;
+ }
+ }
+
public static final String METRICS_GROUP = "group-coordinator-metrics";
- public final static MetricName NUM_OFFSETS = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_OFFSETS = getMetricName(
"GroupMetadataManager", "NumOffsets");
- public final static MetricName NUM_GENERIC_GROUPS = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS = getMetricName(
"GroupMetadataManager", "NumGroups");
- public final static MetricName NUM_GENERIC_GROUPS_PREPARING_REBALANCE = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_PREPARING_REBALANCE = getMetricName(
"GroupMetadataManager", "NumGroupsPreparingRebalance");
- public final static MetricName NUM_GENERIC_GROUPS_COMPLETING_REBALANCE = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_COMPLETING_REBALANCE = getMetricName(
"GroupMetadataManager", "NumGroupsCompletingRebalance");
- public final static MetricName NUM_GENERIC_GROUPS_STABLE = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_STABLE = getMetricName(
"GroupMetadataManager", "NumGroupsStable");
- public final static MetricName NUM_GENERIC_GROUPS_DEAD = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_DEAD = getMetricName(
"GroupMetadataManager", "NumGroupsDead");
- public final static MetricName NUM_GENERIC_GROUPS_EMPTY = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_EMPTY = getMetricName(
"GroupMetadataManager", "NumGroupsEmpty");
- public final static MetricName NUM_CONSUMER_GROUPS = getMetricName(
- "GroupMetadataManager", "NumConsumerGroups");
- public final static MetricName NUM_CONSUMER_GROUPS_EMPTY = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsEmpty");
- public final static MetricName NUM_CONSUMER_GROUPS_ASSIGNING = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsAssigning");
- public final static MetricName NUM_CONSUMER_GROUPS_RECONCILING = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsReconciling");
- public final static MetricName NUM_CONSUMER_GROUPS_STABLE = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsStable");
- public final static MetricName NUM_CONSUMER_GROUPS_DEAD = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsDead");
+
+ public final static String NUM_CONSUMER_GROUPS_METRIC_NAME = "consumer-groups-size";
Review Comment:
It looks like we also use `count` in other places.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -281,100 +306,98 @@ public void onUpdateLastCommittedOffset(TopicPartition tp, long offset) {
}
}
- public static MetricName getMetricName(String type, String name) {
+ public static com.yammer.metrics.core.MetricName getMetricName(String type, String name) {
return getMetricName("kafka.coordinator.group", type, name);
}
private void registerGauges() {
- registry.newGauge(NUM_OFFSETS, new Gauge<Long>() {
+ registry.newGauge(NUM_OFFSETS, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
return numOffsets();
}
});
- registry.newGauge(NUM_GENERIC_GROUPS, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
return numGenericGroups();
}
});
- registry.newGauge(NUM_GENERIC_GROUPS_PREPARING_REBALANCE, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroupsPreparingRebalanceCount();
- }
- });
-
- registry.newGauge(NUM_GENERIC_GROUPS_COMPLETING_REBALANCE, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroupsCompletingRebalanceCount();
- }
- });
-
- registry.newGauge(NUM_GENERIC_GROUPS_STABLE, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroupsStableCount();
- }
- });
-
- registry.newGauge(NUM_GENERIC_GROUPS_DEAD, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroupsDeadCount();
- }
- });
-
- registry.newGauge(NUM_GENERIC_GROUPS_EMPTY, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_PREPARING_REBALANCE, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numGenericGroupsEmptyCount();
+ return numGenericGroupsPreparingRebalance();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_COMPLETING_REBALANCE, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numConsumerGroups();
+ return numGenericGroupsCompletingRebalance();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS_EMPTY, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_STABLE, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numConsumerGroupsEmpty();
+ return numGenericGroupsStable();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS_ASSIGNING, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_DEAD, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numConsumerGroupsAssigning();
+ return numGenericGroupsDead();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS_RECONCILING, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_EMPTY, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numConsumerGroupsReconciling();
+ return numGenericGroupsEmpty();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS_STABLE, new Gauge<Long>() {
- @Override
- public Long value() {
- return numConsumerGroupsStable();
- }
- });
-
- registry.newGauge(NUM_CONSUMER_GROUPS_DEAD, new Gauge<Long>() {
- @Override
- public Long value() {
- return numConsumerGroupsDead();
- }
- });
+ metrics.addMetric(
+ metrics.metricName(NUM_CONSUMER_GROUPS_METRIC_NAME, METRICS_GROUP),
+ (Gauge<Long>) (config, now) -> numConsumerGroups()
Review Comment:
I wonder if we could have `NUM_GROUPS_METRIC_NAME` with a tag called `type`. This would work for all group types in the future.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -29,44 +28,75 @@
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.ConcurrentHashMap;
/**
* These are the metrics which are managed by the {@link org.apache.kafka.coordinator.group.GroupMetadataManager} class.
* They generally pertain to aspects of group management, such as the number of groups in different states.
*/
public class GroupCoordinatorMetrics extends CoordinatorMetrics implements AutoCloseable {
+
+ enum GenericGroupState {
+ ALL("all"),
+ PREPARING_REBALANCE("PreparingRebalance"),
+ COMPLETING_REBALANCE("CompletingRebalance"),
+ STABLE("Stable"),
+ DEAD("Dead"),
+ EMPTY("Empty");
+
+ private final String name;
+
+ GenericGroupState(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public String toString() {
+ return name;
+ }
+ }
+
+ enum ConsumerGroupState {
+ ALL("all"),
+ EMPTY("empty"),
+ ASSIGNING("assigning"),
+ RECONCILING("reconciling"),
+ STABLE("stable"),
+ DEAD("dead");
+
+ private final String name;
+
+ ConsumerGroupState(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public String toString() {
+ return name;
+ }
+ }
+
public static final String METRICS_GROUP = "group-coordinator-metrics";
- public final static MetricName NUM_OFFSETS = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_OFFSETS = getMetricName(
"GroupMetadataManager", "NumOffsets");
- public final static MetricName NUM_GENERIC_GROUPS = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS = getMetricName(
"GroupMetadataManager", "NumGroups");
- public final static MetricName NUM_GENERIC_GROUPS_PREPARING_REBALANCE = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_PREPARING_REBALANCE = getMetricName(
"GroupMetadataManager", "NumGroupsPreparingRebalance");
- public final static MetricName NUM_GENERIC_GROUPS_COMPLETING_REBALANCE = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_COMPLETING_REBALANCE = getMetricName(
"GroupMetadataManager", "NumGroupsCompletingRebalance");
- public final static MetricName NUM_GENERIC_GROUPS_STABLE = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_STABLE = getMetricName(
"GroupMetadataManager", "NumGroupsStable");
- public final static MetricName NUM_GENERIC_GROUPS_DEAD = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_DEAD = getMetricName(
"GroupMetadataManager", "NumGroupsDead");
- public final static MetricName NUM_GENERIC_GROUPS_EMPTY = getMetricName(
+ public final static com.yammer.metrics.core.MetricName NUM_GENERIC_GROUPS_EMPTY = getMetricName(
"GroupMetadataManager", "NumGroupsEmpty");
- public final static MetricName NUM_CONSUMER_GROUPS = getMetricName(
- "GroupMetadataManager", "NumConsumerGroups");
- public final static MetricName NUM_CONSUMER_GROUPS_EMPTY = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsEmpty");
- public final static MetricName NUM_CONSUMER_GROUPS_ASSIGNING = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsAssigning");
- public final static MetricName NUM_CONSUMER_GROUPS_RECONCILING = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsReconciling");
- public final static MetricName NUM_CONSUMER_GROUPS_STABLE = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsStable");
- public final static MetricName NUM_CONSUMER_GROUPS_DEAD = getMetricName(
- "GroupMetadataManager", "NumConsumerGroupsDead");
+
+ public final static String NUM_CONSUMER_GROUPS_METRIC_NAME = "consumer-groups-size";
Review Comment:
I am debating the name for this one. In the coordinator runtime metrics, we used `num-` instead of `-size`. I wonder whether we should do the same here. What do you think? Have you seen `-size` used more often?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -281,100 +306,98 @@ public void onUpdateLastCommittedOffset(TopicPartition tp, long offset) {
}
}
- public static MetricName getMetricName(String type, String name) {
+ public static com.yammer.metrics.core.MetricName getMetricName(String type, String name) {
return getMetricName("kafka.coordinator.group", type, name);
}
private void registerGauges() {
- registry.newGauge(NUM_OFFSETS, new Gauge<Long>() {
+ registry.newGauge(NUM_OFFSETS, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
return numOffsets();
}
});
- registry.newGauge(NUM_GENERIC_GROUPS, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
return numGenericGroups();
}
});
- registry.newGauge(NUM_GENERIC_GROUPS_PREPARING_REBALANCE, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroupsPreparingRebalanceCount();
- }
- });
-
- registry.newGauge(NUM_GENERIC_GROUPS_COMPLETING_REBALANCE, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroupsCompletingRebalanceCount();
- }
- });
-
- registry.newGauge(NUM_GENERIC_GROUPS_STABLE, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroupsStableCount();
- }
- });
-
- registry.newGauge(NUM_GENERIC_GROUPS_DEAD, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroupsDeadCount();
- }
- });
-
- registry.newGauge(NUM_GENERIC_GROUPS_EMPTY, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_PREPARING_REBALANCE, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numGenericGroupsEmptyCount();
+ return numGenericGroupsPreparingRebalance();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_COMPLETING_REBALANCE, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numConsumerGroups();
+ return numGenericGroupsCompletingRebalance();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS_EMPTY, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_STABLE, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numConsumerGroupsEmpty();
+ return numGenericGroupsStable();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS_ASSIGNING, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_DEAD, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numConsumerGroupsAssigning();
+ return numGenericGroupsDead();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS_RECONCILING, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_EMPTY, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numConsumerGroupsReconciling();
+ return numGenericGroupsEmpty();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS_STABLE, new Gauge<Long>() {
- @Override
- public Long value() {
- return numConsumerGroupsStable();
- }
- });
-
- registry.newGauge(NUM_CONSUMER_GROUPS_DEAD, new Gauge<Long>() {
- @Override
- public Long value() {
- return numConsumerGroupsDead();
- }
- });
+ metrics.addMetric(
+ metrics.metricName(NUM_CONSUMER_GROUPS_METRIC_NAME, METRICS_GROUP),
+ (Gauge<Long>) (config, now) -> numConsumerGroups()
+ );
+
+ metrics.addMetric(
+ metrics.metricName(NUM_CONSUMER_GROUPS_METRIC_NAME, METRICS_GROUP, Utils.mkMap(
+ Utils.mkEntry(NUM_CONSUMER_GROUPS_STATE_TAG, ConsumerGroupState.EMPTY.toString())
+ )),
Review Comment:
nit: Collections.singletonMap()? This also applies to the other ones below.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShard.java:
##########
@@ -95,95 +88,147 @@ public TimelineGaugeCounter(TimelineLong timelineLong, AtomicLong atomicLong) {
public GroupCoordinatorMetricsShard(
SnapshotRegistry snapshotRegistry,
Map<String, Sensor> globalSensors,
- Map<String, AtomicLong> globalGauges,
TopicPartition topicPartition
) {
Objects.requireNonNull(snapshotRegistry);
- TimelineLong numOffsetsTimeline = new TimelineLong(snapshotRegistry);
- TimelineLong numGenericGroupsTimeline = new TimelineLong(snapshotRegistry);
- TimelineLong numConsumerGroupsTimeline = new TimelineLong(snapshotRegistry);
+ numOffsetsTimelineGaugeCounter = new TimelineGaugeCounter(new TimelineLong(snapshotRegistry), new AtomicLong(0));
+ numGenericGroupsTimelineCounter = new TimelineGaugeCounter(new TimelineLong(snapshotRegistry), new AtomicLong(0));
+
+ this.genericGroupGauges = Utils.mkMap(
+ Utils.mkEntry(GroupCoordinatorMetrics.GenericGroupState.PREPARING_REBALANCE, new AtomicLong(0)),
+ Utils.mkEntry(GroupCoordinatorMetrics.GenericGroupState.COMPLETING_REBALANCE, new AtomicLong(0)),
+ Utils.mkEntry(GroupCoordinatorMetrics.GenericGroupState.STABLE, new AtomicLong(0)),
+ Utils.mkEntry(GroupCoordinatorMetrics.GenericGroupState.DEAD, new AtomicLong(0)),
+ Utils.mkEntry(GroupCoordinatorMetrics.GenericGroupState.EMPTY, new AtomicLong(0))
+ );
+
TimelineLong numConsumerGroupsEmptyTimeline = new TimelineLong(snapshotRegistry);
TimelineLong numConsumerGroupsAssigningTimeline = new TimelineLong(snapshotRegistry);
TimelineLong numConsumerGroupsReconcilingTimeline = new TimelineLong(snapshotRegistry);
TimelineLong numConsumerGroupsStableTimeline = new TimelineLong(snapshotRegistry);
TimelineLong numConsumerGroupsDeadTimeline = new TimelineLong(snapshotRegistry);
- this.localGauges = Collections.unmodifiableMap(Utils.mkMap(
- Utils.mkEntry(NUM_OFFSETS.getName(),
- new TimelineGaugeCounter(numOffsetsTimeline, new AtomicLong(0))),
- Utils.mkEntry(NUM_GENERIC_GROUPS.getName(),
- new TimelineGaugeCounter(numGenericGroupsTimeline, new AtomicLong(0))),
- Utils.mkEntry(NUM_CONSUMER_GROUPS.getName(),
- new TimelineGaugeCounter(numConsumerGroupsTimeline, new AtomicLong(0))),
- Utils.mkEntry(NUM_CONSUMER_GROUPS_EMPTY.getName(),
+ this.consumerGroupGauges = Utils.mkMap(
+ Utils.mkEntry(GroupCoordinatorMetrics.ConsumerGroupState.EMPTY,
new TimelineGaugeCounter(numConsumerGroupsEmptyTimeline, new AtomicLong(0))),
- Utils.mkEntry(NUM_CONSUMER_GROUPS_ASSIGNING.getName(),
+ Utils.mkEntry(GroupCoordinatorMetrics.ConsumerGroupState.ASSIGNING,
new TimelineGaugeCounter(numConsumerGroupsAssigningTimeline, new AtomicLong(0))),
- Utils.mkEntry(NUM_CONSUMER_GROUPS_RECONCILING.getName(),
+ Utils.mkEntry(GroupCoordinatorMetrics.ConsumerGroupState.RECONCILING,
new TimelineGaugeCounter(numConsumerGroupsReconcilingTimeline, new AtomicLong(0))),
- Utils.mkEntry(NUM_CONSUMER_GROUPS_STABLE.getName(),
+ Utils.mkEntry(GroupCoordinatorMetrics.ConsumerGroupState.STABLE,
new TimelineGaugeCounter(numConsumerGroupsStableTimeline, new AtomicLong(0))),
- Utils.mkEntry(NUM_CONSUMER_GROUPS_DEAD.getName(),
+ Utils.mkEntry(GroupCoordinatorMetrics.ConsumerGroupState.DEAD,
new TimelineGaugeCounter(numConsumerGroupsDeadTimeline, new AtomicLong(0)))
- ));
+ );
this.globalSensors = Objects.requireNonNull(globalSensors);
- this.globalGauges = Objects.requireNonNull(globalGauges);
this.topicPartition = Objects.requireNonNull(topicPartition);
}
- @Override
- public void incrementGlobalGauge(MetricName metricName) {
- AtomicLong gaugeCounter = globalGauges.get(metricName.getName());
- if (gaugeCounter != null) {
- gaugeCounter.incrementAndGet();
+ public void incrementNumGenericGroups(GroupCoordinatorMetrics.GenericGroupState state) {
+ AtomicLong counter = genericGroupGauges.get(state);
+ if (counter != null) {
+ counter.incrementAndGet();
}
}
- @Override
- public void incrementLocalGauge(MetricName metricName) {
- TimelineGaugeCounter gaugeCounter = localGauges.get(metricName.getName());
+ /**
+ * Increment the number of offsets.
+ */
+ public void incrementNumOffsets() {
+ synchronized (numOffsetsTimelineGaugeCounter.timelineLong) {
+ numOffsetsTimelineGaugeCounter.timelineLong.increment();
+ }
+ }
+
+ /**
+ * Increment the number of consumer groups.
+ *
+ * @param state the consumer group state.
+ */
+ public void incrementNumConsumerGroups(GroupCoordinatorMetrics.ConsumerGroupState state) {
+ TimelineGaugeCounter gaugeCounter = consumerGroupGauges.get(state);
if (gaugeCounter != null) {
synchronized (gaugeCounter.timelineLong) {
gaugeCounter.timelineLong.increment();
}
}
}
- @Override
- public void decrementGlobalGauge(MetricName metricName) {
- AtomicLong gaugeCounter = globalGauges.get(metricName.getName());
- if (gaugeCounter != null) {
- gaugeCounter.decrementAndGet();
+ /**
+ * Decrement the number of offsets.
+ */
+ public void decrementNumOffsets() {
+ synchronized (numOffsetsTimelineGaugeCounter.timelineLong) {
+ numOffsetsTimelineGaugeCounter.timelineLong.decrement();
}
}
- @Override
- public void decrementLocalGauge(MetricName metricName) {
- TimelineGaugeCounter gaugeCounter = localGauges.get(metricName.getName());
+ /**
+ * Decrement the number of consumer groups.
+ *
+ * @param state the consumer group state.
+ */
+ public void decrementNumGenericGroups(GroupCoordinatorMetrics.GenericGroupState state) {
+ AtomicLong counter = genericGroupGauges.get(state);
+ if (counter != null) {
+ counter.decrementAndGet();
+ }
+ }
+
+ /**
+ * Decrement the number of consumer groups.
+ *
+ * @param state the consumer group state.
+ */
+ public void decrementNumConsumerGroups(GroupCoordinatorMetrics.ConsumerGroupState state) {
+ TimelineGaugeCounter gaugeCounter = consumerGroupGauges.get(state);
if (gaugeCounter != null) {
synchronized (gaugeCounter.timelineLong) {
gaugeCounter.timelineLong.decrement();
}
}
}
- @Override
- public long globalGaugeValue(MetricName metricName) {
- AtomicLong gaugeCounter = globalGauges.get(metricName.getName());
- if (gaugeCounter != null) {
- return gaugeCounter.get();
+ /**
+ * Obtain the number of offsets.
+ */
+ public long numOffsets() {
+ return numOffsetsTimelineGaugeCounter.atomicLong.get();
+ }
+
+ /**
+ * Obtain the number of generic groups.
+ */
+ public long numGenericGroups(GroupCoordinatorMetrics.GenericGroupState state) {
+ if (state == GroupCoordinatorMetrics.GenericGroupState.ALL) {
+ return genericGroupGauges.values().stream()
+ .mapToLong(AtomicLong::get).sum();
}
- return 0;
+
+ AtomicLong counter = genericGroupGauges.get(state);
+ if (counter != null) {
+ return counter.get();
+ }
+ return 0L;
}
- @Override
- public long localGaugeValue(MetricName metricName) {
- TimelineGaugeCounter gaugeCounter = localGauges.get(metricName.getName());
+ /**
+ * Obtain the current value of a local consumer group gauge.
+ *
+ * @param state the consumer group state.
+ */
+ public long numConsumerGroups(GroupCoordinatorMetrics.ConsumerGroupState state) {
+ if (state == GroupCoordinatorMetrics.ConsumerGroupState.ALL) {
Review Comment:
Do we actually call it with `ALL`?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14848:
URL: https://github.com/apache/kafka/pull/14848#issuecomment-1857508292
Triggered a new build.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1426193169
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -75,31 +74,23 @@ public class GroupCoordinatorMetrics extends CoordinatorMetrics implements AutoC
public static final String GENERIC_GROUP_REBALANCES_SENSOR_NAME = "GenericGroupRebalances";
public static final String CONSUMER_GROUP_REBALANCES_SENSOR_NAME = "ConsumerGroupRebalances";
+ private final MetricName genericGroupsCountMetricName;
Review Comment:
MetricName also contains the tags so i split it into genericGroupsCount and consumerGroupsCount
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1425599818
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -174,75 +151,91 @@ public GroupCoordinatorMetrics(MetricsRegistry registry, Metrics metrics) {
));
}
- public Long numOffsets() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_OFFSETS)).sum();
+ private Long numOffsets() {
+ return shards.values().stream().mapToLong(GroupCoordinatorMetricsShard::numOffsets).sum();
}
- public Long numGenericGroups() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_GENERIC_GROUPS)).sum();
+ private Long numGenericGroups() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(null)).sum();
Review Comment:
nit: It may be better to add an overload without argument instead of passing `null`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -174,75 +151,91 @@ public GroupCoordinatorMetrics(MetricsRegistry registry, Metrics metrics) {
));
}
- public Long numOffsets() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_OFFSETS)).sum();
+ private Long numOffsets() {
+ return shards.values().stream().mapToLong(GroupCoordinatorMetricsShard::numOffsets).sum();
}
- public Long numGenericGroups() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_GENERIC_GROUPS)).sum();
+ private Long numGenericGroups() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(null)).sum();
}
- public Long numGenericGroupsPreparingRebalanceCount() {
- return NUM_GENERIC_GROUPS_PREPARING_REBALANCE_COUNTER.get();
+ private Long numGenericGroupsPreparingRebalance() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.PREPARING_REBALANCE)).sum();
}
- public Long numGenericGroupsCompletingRebalanceCount() {
- return NUM_GENERIC_GROUPS_COMPLETING_REBALANCE_COUNTER.get();
+ private Long numGenericGroupsCompletingRebalance() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.COMPLETING_REBALANCE)).sum();
}
- public Long numGenericGroupsStableCount() {
- return NUM_GENERIC_GROUPS_STABLE_COUNTER.get();
+ private Long numGenericGroupsStable() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.STABLE)).sum();
}
- public Long numGenericGroupsDeadCount() {
- return NUM_GENERIC_GROUPS_DEAD_COUNTER.get();
+ private Long numGenericGroupsDead() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.DEAD)).sum();
}
- public Long numGenericGroupsEmptyCount() {
- return NUM_GENERIC_GROUPS_EMPTY_COUNTER.get();
+ private Long numGenericGroupsEmpty() {
+ return shards.values().stream().mapToLong(shard -> shard.numGenericGroups(GenericGroupState.EMPTY)).sum();
}
- public long numConsumerGroups() {
- return shards.values().stream().mapToLong(shard -> shard.localGaugeValue(NUM_CONSUMER_GROUPS)).sum();
+ private long numConsumerGroups() {
+ return shards.values().stream().mapToLong(shard -> shard.numConsumerGroups(null)).sum();
Review Comment:
ditto.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1425601951
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShard.java:
##########
@@ -71,21 +55,30 @@ public TimelineGaugeCounter(TimelineLong timelineLong, AtomicLong atomicLong) {
this.atomicLong = atomicLong;
}
}
+ /**
+ * Consumer group size gauge counters keyed by the metric name.
+ */
+ private final Map<GenericGroupState, AtomicLong> genericGroupGauges;
/**
- * Local timeline gauge counters keyed by the metric name.
+ * Consumer group size gauge counters keyed by the metric name.
*/
- private final Map<String, TimelineGaugeCounter> localGauges;
+ private final Map<ConsumerGroup.ConsumerGroupState, TimelineGaugeCounter> consumerGroupGauges;
Review Comment:
nit: Could we import `ConsumerGroupState`?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14848:
URL: https://github.com/apache/kafka/pull/14848#issuecomment-1863996596
Merged to trunk and 3.7.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1426788188
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -281,100 +297,93 @@ public void onUpdateLastCommittedOffset(TopicPartition tp, long offset) {
}
}
- public static MetricName getMetricName(String type, String name) {
+ public static com.yammer.metrics.core.MetricName getMetricName(String type, String name) {
return getMetricName("kafka.coordinator.group", type, name);
}
private void registerGauges() {
- registry.newGauge(NUM_OFFSETS, new Gauge<Long>() {
+ registry.newGauge(NUM_OFFSETS, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
return numOffsets();
}
Review Comment:
just to be clear, i was wondering whether the following would work:
```
registry.newGauge(NUM_OFFSETS, () -> numOffsets());
```
Does it?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1426200690
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -281,100 +297,93 @@ public void onUpdateLastCommittedOffset(TopicPartition tp, long offset) {
}
}
- public static MetricName getMetricName(String type, String name) {
+ public static com.yammer.metrics.core.MetricName getMetricName(String type, String name) {
return getMetricName("kafka.coordinator.group", type, name);
}
private void registerGauges() {
- registry.newGauge(NUM_OFFSETS, new Gauge<Long>() {
+ registry.newGauge(NUM_OFFSETS, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
return numOffsets();
}
Review Comment:
the format is different between yammer metrics and kafka metrics. numOffsets follow the yammer format
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1425605559
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -281,100 +277,99 @@ public void onUpdateLastCommittedOffset(TopicPartition tp, long offset) {
}
}
- public static MetricName getMetricName(String type, String name) {
+ public static com.yammer.metrics.core.MetricName getMetricName(String type, String name) {
return getMetricName("kafka.coordinator.group", type, name);
}
private void registerGauges() {
- registry.newGauge(NUM_OFFSETS, new Gauge<Long>() {
+ registry.newGauge(NUM_OFFSETS, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
return numOffsets();
}
});
- registry.newGauge(NUM_GENERIC_GROUPS, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroups();
- }
- });
-
- registry.newGauge(NUM_GENERIC_GROUPS_PREPARING_REBALANCE, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroupsPreparingRebalanceCount();
- }
- });
-
- registry.newGauge(NUM_GENERIC_GROUPS_COMPLETING_REBALANCE, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroupsCompletingRebalanceCount();
- }
- });
-
- registry.newGauge(NUM_GENERIC_GROUPS_STABLE, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroupsStableCount();
- }
- });
-
- registry.newGauge(NUM_GENERIC_GROUPS_DEAD, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroupsDeadCount();
- }
- });
-
- registry.newGauge(NUM_GENERIC_GROUPS_EMPTY, new Gauge<Long>() {
- @Override
- public Long value() {
- return numGenericGroupsEmptyCount();
- }
- });
-
- registry.newGauge(NUM_CONSUMER_GROUPS, new Gauge<Long>() {
- @Override
- public Long value() {
- return numConsumerGroups();
- }
- });
-
- registry.newGauge(NUM_CONSUMER_GROUPS_EMPTY, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_PREPARING_REBALANCE, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numConsumerGroupsEmpty();
+ return numGenericGroupsPreparingRebalance();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS_ASSIGNING, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_COMPLETING_REBALANCE, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numConsumerGroupsAssigning();
+ return numGenericGroupsCompletingRebalance();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS_RECONCILING, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_STABLE, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numConsumerGroupsReconciling();
+ return numGenericGroupsStable();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS_STABLE, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_DEAD, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numConsumerGroupsStable();
+ return numGenericGroupsDead();
}
});
- registry.newGauge(NUM_CONSUMER_GROUPS_DEAD, new Gauge<Long>() {
+ registry.newGauge(NUM_GENERIC_GROUPS_EMPTY, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
- return numConsumerGroupsDead();
+ return numGenericGroupsEmpty();
}
});
+ metrics.addMetric(
+ metrics.metricName(GROUPS_COUNT_METRIC_NAME, METRICS_GROUP, Collections.singletonMap(
Review Comment:
I wonder if we could add the description to all the new metrics. Have you considered it?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15870: Move new group coordinator metrics from Yammer to Metrics [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14848:
URL: https://github.com/apache/kafka/pull/14848#discussion_r1425827813
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -75,31 +74,23 @@ public class GroupCoordinatorMetrics extends CoordinatorMetrics implements AutoC
public static final String GENERIC_GROUP_REBALANCES_SENSOR_NAME = "GenericGroupRebalances";
public static final String CONSUMER_GROUP_REBALANCES_SENSOR_NAME = "ConsumerGroupRebalances";
+ private final MetricName genericGroupsCountMetricName;
Review Comment:
nit: Should it be called groupsCountMetricName?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -108,6 +99,55 @@ public GroupCoordinatorMetrics(MetricsRegistry registry, Metrics metrics) {
this.registry = Objects.requireNonNull(registry);
this.metrics = Objects.requireNonNull(metrics);
+ genericGroupsCountMetricName = metrics.metricName(
+ GROUPS_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The total number of generic groups.",
Review Comment:
nit: The total number of groups using the generic rebalance protocol.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -281,100 +297,93 @@ public void onUpdateLastCommittedOffset(TopicPartition tp, long offset) {
}
}
- public static MetricName getMetricName(String type, String name) {
+ public static com.yammer.metrics.core.MetricName getMetricName(String type, String name) {
return getMetricName("kafka.coordinator.group", type, name);
}
private void registerGauges() {
- registry.newGauge(NUM_OFFSETS, new Gauge<Long>() {
+ registry.newGauge(NUM_OFFSETS, new com.yammer.metrics.core.Gauge<Long>() {
@Override
public Long value() {
return numOffsets();
}
Review Comment:
nit: Could we also use a lambda for those in order to make the code consistent with the other metrics?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java:
##########
@@ -108,6 +99,55 @@ public GroupCoordinatorMetrics(MetricsRegistry registry, Metrics metrics) {
this.registry = Objects.requireNonNull(registry);
this.metrics = Objects.requireNonNull(metrics);
+ genericGroupsCountMetricName = metrics.metricName(
+ GROUPS_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The total number of generic groups.",
+ Collections.singletonMap(GROUPS_COUNT_TYPE_TAG, Group.GroupType.GENERIC.toString())
+ );
+
+ consumerGroupsCountMetricName = metrics.metricName(
+ GROUPS_COUNT_METRIC_NAME,
+ METRICS_GROUP,
+ "The total number of consumer groups.",
Review Comment:
nit: The total number of groups using the consumer rebalance protocol.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org