You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/07/21 10:17:12 UTC
(kafka) branch trunk updated: KAFKA-17153 KafkaMetricsGroup#newGauge should accept functional interface instead of `com.yammer.metrics.core.Gague` (#16618)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new e9a8c3c455f KAFKA-17153 KafkaMetricsGroup#newGauge should accept functional interface instead of `com.yammer.metrics.core.Gague` (#16618)
e9a8c3c455f is described below
commit e9a8c3c455f097144f06794533f93db181ad46bd
Author: Kuan-Po Tseng <br...@gmail.com>
AuthorDate: Sun Jul 21 18:17:04 2024 +0800
KAFKA-17153 KafkaMetricsGroup#newGauge should accept functional interface instead of `com.yammer.metrics.core.Gague` (#16618)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../main/java/kafka/log/remote/RemoteLogManager.java | 8 +-------
.../coordinator/group/GroupMetadataManager.scala | 5 +++--
.../kafka/server/metrics/KafkaMetricsGroup.java | 16 +++++++++++-----
.../org/apache/kafka/server/AssignmentsManager.java | 17 +++++------------
.../internals/log/RemoteStorageThreadPool.java | 19 +++++--------------
5 files changed, 25 insertions(+), 40 deletions(-)
diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
index 11c34a0f535..f3f0ccc11ad 100644
--- a/core/src/main/java/kafka/log/remote/RemoteLogManager.java
+++ b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
@@ -82,7 +82,6 @@ import org.apache.kafka.storage.internals.log.RemoteStorageThreadPool;
import org.apache.kafka.storage.internals.log.TransactionIndex;
import org.apache.kafka.storage.internals.log.TxnIndexSearchResult;
-import com.yammer.metrics.core.Gauge;
import com.yammer.metrics.core.Timer;
import org.slf4j.Logger;
@@ -253,12 +252,7 @@ public class RemoteLogManager implements Closeable {
followerThreadPool = new RLMScheduledThreadPool(rlmConfig.remoteLogManagerThreadPoolSize(),
"RLMFollowerScheduledThreadPool", "kafka-rlm-follower-thread-pool-");
- metricsGroup.newGauge(REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT_METRIC, new Gauge<Double>() {
- @Override
- public Double value() {
- return rlmCopyThreadPool.getIdlePercent();
- }
- });
+ metricsGroup.newGauge(REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT_METRIC, rlmCopyThreadPool::getIdlePercent);
remoteReadTimer = metricsGroup.newTimer(REMOTE_LOG_READER_FETCH_RATE_AND_TIME_METRIC,
TimeUnit.MILLISECONDS, TimeUnit.SECONDS);
diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
index b5fddac6f40..064df2f607d 100644
--- a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
+++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
@@ -24,6 +24,7 @@ import java.util.{Optional, OptionalInt}
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.locks.ReentrantLock
import java.util.concurrent.ConcurrentHashMap
+import java.util.function.Supplier
import com.yammer.metrics.core.Gauge
import kafka.common.OffsetAndMetadata
import kafka.coordinator.group.GroupMetadataManager.maybeConvertOffsetCommitError
@@ -126,9 +127,9 @@ class GroupMetadataManager(brokerId: Int,
this.logIdent = s"[GroupMetadataManager brokerId=$brokerId] "
- private def recreateGauge[T](name: String, gauge: Gauge[T]): Gauge[T] = {
+ private def recreateGauge[T](name: String, metric: Supplier[T]): Gauge[T] = {
metricsGroup.removeMetric(name)
- metricsGroup.newGauge(name, gauge)
+ metricsGroup.newGauge(name, metric)
}
recreateGauge("NumOffsets",
diff --git a/server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java b/server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java
index 20f09669194..f876ea76c19 100644
--- a/server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java
+++ b/server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java
@@ -29,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
public class KafkaMetricsGroup {
@@ -70,16 +71,21 @@ public class KafkaMetricsGroup {
return new MetricName(group, typeName, name, scope, nameBuilder.toString());
}
- public final <T> Gauge<T> newGauge(String name, Gauge<T> metric, Map<String, String> tags) {
- return KafkaYammerMetrics.defaultRegistry().newGauge(metricName(name, tags), metric);
+ public <T> Gauge<T> newGauge(String name, Supplier<T> metric, Map<String, String> tags) {
+ return newGauge(metricName(name, tags), metric);
}
- public final <T> Gauge<T> newGauge(String name, Gauge<T> metric) {
+ public <T> Gauge<T> newGauge(String name, Supplier<T> metric) {
return newGauge(name, metric, Collections.emptyMap());
}
- public final <T> Gauge<T> newGauge(MetricName metricName, Gauge<T> metric) {
- return KafkaYammerMetrics.defaultRegistry().newGauge(metricName, metric);
+ public <T> Gauge<T> newGauge(MetricName name, Supplier<T> metric) {
+ return KafkaYammerMetrics.defaultRegistry().newGauge(name, new Gauge<T>() {
+ @Override
+ public T value() {
+ return metric.get();
+ }
+ });
}
public final Meter newMeter(String name, String eventType,
diff --git a/server/src/main/java/org/apache/kafka/server/AssignmentsManager.java b/server/src/main/java/org/apache/kafka/server/AssignmentsManager.java
index ef9bc77ec6e..48c9e5a2d6e 100644
--- a/server/src/main/java/org/apache/kafka/server/AssignmentsManager.java
+++ b/server/src/main/java/org/apache/kafka/server/AssignmentsManager.java
@@ -36,8 +36,6 @@ import org.apache.kafka.queue.KafkaEventQueue;
import org.apache.kafka.server.common.TopicIdPartition;
import org.apache.kafka.server.metrics.KafkaMetricsGroup;
-import com.yammer.metrics.core.Gauge;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -106,16 +104,7 @@ public class AssignmentsManager {
"broker-" + brokerId + "-directory-assignments-manager-",
new ShutdownEvent());
channelManager.start();
- this.metricsGroup.newGauge(QUEUE_REPLICA_TO_DIR_ASSIGNMENTS_METRIC_NAME, new Gauge<Integer>() {
- @Override
- public Integer value() {
- return getMapSize(inflight) + getMapSize(pending);
- }
-
- private int getMapSize(Map<TopicIdPartition, AssignmentEvent> map) {
- return map == null ? 0 : map.size();
- }
- });
+ this.metricsGroup.newGauge(QUEUE_REPLICA_TO_DIR_ASSIGNMENTS_METRIC_NAME, () -> getMapSize(inflight) + getMapSize(pending));
if (dirIdToPath == null) dirIdToPath = id -> Optional.empty();
this.dirIdToPath = dirIdToPath;
if (topicIdToName == null) topicIdToName = id -> Optional.empty();
@@ -492,4 +481,8 @@ public class AssignmentsManager {
.setBrokerEpoch(brokerEpoch)
.setDirectories(new ArrayList<>(directoryMap.values()));
}
+
+ private static int getMapSize(Map<TopicIdPartition, AssignmentEvent> map) {
+ return map == null ? 0 : map.size();
+ }
}
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteStorageThreadPool.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteStorageThreadPool.java
index 36fcd616763..3949e919bb4 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteStorageThreadPool.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteStorageThreadPool.java
@@ -21,8 +21,6 @@ import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.server.metrics.KafkaMetricsGroup;
-import com.yammer.metrics.core.Gauge;
-
import org.slf4j.Logger;
import java.util.concurrent.LinkedBlockingQueue;
@@ -39,6 +37,7 @@ public class RemoteStorageThreadPool extends ThreadPoolExecutor {
private final Logger logger;
private final KafkaMetricsGroup metricsGroup = new KafkaMetricsGroup(this.getClass());
+ @SuppressWarnings("this-escape")
public RemoteStorageThreadPool(String threadNamePrefix,
int numThreads,
int maxPendingTasks) {
@@ -51,18 +50,10 @@ public class RemoteStorageThreadPool extends ThreadPoolExecutor {
}
}.logger(RemoteStorageThreadPool.class);
- metricsGroup.newGauge(REMOTE_LOG_READER_TASK_QUEUE_SIZE_METRIC.getName(), new Gauge<Integer>() {
- @Override
- public Integer value() {
- return RemoteStorageThreadPool.this.getQueue().size();
- }
- });
- metricsGroup.newGauge(REMOTE_LOG_READER_AVG_IDLE_PERCENT_METRIC.getName(), new Gauge<Double>() {
- @Override
- public Double value() {
- return 1 - (double) RemoteStorageThreadPool.this.getActiveCount() / (double) RemoteStorageThreadPool.this.getCorePoolSize();
- }
- });
+ metricsGroup.newGauge(REMOTE_LOG_READER_TASK_QUEUE_SIZE_METRIC.getName(),
+ () -> getQueue().size());
+ metricsGroup.newGauge(REMOTE_LOG_READER_AVG_IDLE_PERCENT_METRIC.getName(),
+ () -> 1 - (double) getActiveCount() / (double) getCorePoolSize());
}
@Override