You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2023/01/03 06:30:07 UTC
[GitHub] [kafka] ivanyu opened a new pull request, #13067: KAFKA-14524: Create Java counterpart of KafkaMetricsGroup
ivanyu opened a new pull request, #13067:
URL: https://github.com/apache/kafka/pull/13067
`KafkaMetricsGroup` is used by many classes. To facilitate the migration to Java, this commit adds the Java implementation of this class. The Java implementation is not supposed to be used as a base class, but as an object.
### 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
[GitHub] [kafka] ivanyu commented on pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on PR #13067:
URL: https://github.com/apache/kafka/pull/13067#issuecomment-1371142944
@ijuma I don't see much difference with `QuorumControllerMetrics` on the low level.
Tags are not used in `QuorumControllerMetrics` (this is situational I think); type names are set explicitly instead of being derived from the class names. Apart from that, the metric names are formed the same way.
The metric registry is passed explicitly to `QuorumControllerMetrics`, but it's the same `KafkaYammerMetrics.defaultRegistry()`.
On a higher level, the big difference is that `QuorumControllerMetrics` encapsulates the metric machinery and gives the user class just some meaningful methods. This has a downside, too: `QuorumControllerMetrics` doesn't have access to the state of the user class, so it has to store some bits of this state internally (these bunch of volatiles and atomics) to create gauges. If this is convenient to apply universally probably depends on how much state duplication is needed (e.g. fields like `active` or `metadataErrorCount` don't exist in the user class `SharedServer`, so `QuorumControllerMetrics` is the primary holder of those). It may be a bit inconvenient to do in all the classes like `Partition`, however in "big" classes like `SharedServer` this makes sense.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1072223447
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -293,14 +295,14 @@ class Partition(val topicPartition: TopicPartition,
private var controllerEpoch: Int = KafkaController.InitialControllerEpoch
this.logIdent = s"[Partition $topicPartition broker=$localBrokerId] "
- private val tags = Map("topic" -> topic, "partition" -> partitionId.toString)
+ private val tags = Map("topic" -> topic, "partition" -> partitionId.toString).asJava
- newGauge("UnderReplicated", () => if (isUnderReplicated) 1 else 0, tags)
- newGauge("InSyncReplicasCount", () => if (isLeader) partitionState.isr.size else 0, tags)
- newGauge("UnderMinIsr", () => if (isUnderMinIsr) 1 else 0, tags)
- newGauge("AtMinIsr", () => if (isAtMinIsr) 1 else 0, tags)
- newGauge("ReplicasCount", () => if (isLeader) assignmentState.replicationFactor else 0, tags)
- newGauge("LastStableOffsetLag", () => log.map(_.lastStableOffsetLag).getOrElse(0), tags)
+ Partition.metricsGroup.newGauge("UnderReplicated", () => if (isUnderReplicated) 1 else 0, tags)
Review Comment:
Removed (by adding `import Partition.metricsGroup`).
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
Review Comment:
Removed
--
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
[GitHub] [kafka] ijuma commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1103756147
##########
core/src/main/scala/kafka/log/LogSegment.scala:
##########
@@ -689,6 +689,7 @@ object LogSegment {
}
}
-object LogFlushStats extends KafkaMetricsGroup {
- val logFlushTimer: Timer = newTimer("LogFlushRateAndTimeMs", TimeUnit.MILLISECONDS, TimeUnit.SECONDS)
+object LogFlushStats {
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
Review Comment:
Let's pass `classOf[LogFlushStats]`.
##########
core/src/main/scala/kafka/log/UnifiedLog.scala:
##########
@@ -247,10 +248,19 @@ class UnifiedLog(@volatile var logStartOffset: Long,
@volatile private var _topicId: Option[Uuid],
val keepPartitionMetadataFile: Boolean,
val remoteStorageSystemEnable: Boolean = false,
- remoteLogManager: Option[RemoteLogManager] = None) extends Logging with KafkaMetricsGroup {
+ remoteLogManager: Option[RemoteLogManager] = None) extends Logging {
import kafka.log.UnifiedLog._
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass) {
+ // For compatibility, metrics are defined to be under `Log` class
+ override def metricName(name: String, tags: util.Map[String, String]): MetricName = {
+ val pkg = getClass.getPackage
+ val pkgStr = if (pkg == null) "" else pkg.getName
Review Comment:
This seems unnecessary, we know the package is not null.
##########
core/src/main/scala/kafka/server/DelayedDeleteRecords.scala:
##########
@@ -122,12 +121,12 @@ class DelayedDeleteRecords(delayMs: Long,
}
}
-object DelayedDeleteRecordsMetrics extends KafkaMetricsGroup {
+object DelayedDeleteRecordsMetrics {
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
Review Comment:
Pass `classOf[DelayedDeleteRecordsMetrics]`.
##########
core/src/main/scala/kafka/network/RequestChannel.scala:
##########
@@ -497,51 +501,58 @@ object RequestMetrics {
val ErrorsPerSec = "ErrorsPerSec"
}
-class RequestMetrics(name: String) extends KafkaMetricsGroup {
+class RequestMetrics(name: String) {
import RequestMetrics._
- val tags = Map("request" -> name)
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+
+ val tags = Map("request" -> name).asJava
val requestRateInternal = new Pool[Short, Meter]()
// time a request spent in a request queue
- val requestQueueTimeHist = newHistogram(RequestQueueTimeMs, biased = true, tags)
+ val requestQueueTimeHist = metricsGroup.newHistogram(RequestQueueTimeMs, true, tags)
// time a request takes to be processed at the local broker
- val localTimeHist = newHistogram(LocalTimeMs, biased = true, tags)
+ val localTimeHist = metricsGroup.newHistogram(LocalTimeMs, true, tags)
// time a request takes to wait on remote brokers (currently only relevant to fetch and produce requests)
- val remoteTimeHist = newHistogram(RemoteTimeMs, biased = true, tags)
+ val remoteTimeHist = metricsGroup.newHistogram(RemoteTimeMs, true, tags)
// time a request is throttled, not part of the request processing time (throttling is done at the client level
// for clients that support KIP-219 and by muting the channel for the rest)
- val throttleTimeHist = newHistogram(ThrottleTimeMs, biased = true, tags)
+ val throttleTimeHist = metricsGroup.newHistogram(ThrottleTimeMs, true, tags)
// time a response spent in a response queue
- val responseQueueTimeHist = newHistogram(ResponseQueueTimeMs, biased = true, tags)
+ val responseQueueTimeHist = metricsGroup.newHistogram(ResponseQueueTimeMs, true, tags)
// time to send the response to the requester
- val responseSendTimeHist = newHistogram(ResponseSendTimeMs, biased = true, tags)
- val totalTimeHist = newHistogram(TotalTimeMs, biased = true, tags)
+ val responseSendTimeHist = metricsGroup.newHistogram(ResponseSendTimeMs, true, tags)
+ val totalTimeHist = metricsGroup.newHistogram(TotalTimeMs, true, tags)
// request size in bytes
- val requestBytesHist = newHistogram(RequestBytes, biased = true, tags)
+ val requestBytesHist = metricsGroup.newHistogram(RequestBytes, true, tags)
// time for message conversions (only relevant to fetch and produce requests)
val messageConversionsTimeHist =
if (name == ApiKeys.FETCH.name || name == ApiKeys.PRODUCE.name)
- Some(newHistogram(MessageConversionsTimeMs, biased = true, tags))
+ Some(metricsGroup.newHistogram(MessageConversionsTimeMs, true, tags))
else
None
// Temporary memory allocated for processing request (only populated for fetch and produce requests)
// This shows the memory allocated for compression/conversions excluding the actual request size
val tempMemoryBytesHist =
if (name == ApiKeys.FETCH.name || name == ApiKeys.PRODUCE.name)
- Some(newHistogram(TemporaryMemoryBytes, biased = true, tags))
+ Some(metricsGroup.newHistogram(TemporaryMemoryBytes, true, tags))
else
None
private val errorMeters = mutable.Map[Errors, ErrorMeter]()
Errors.values.foreach(error => errorMeters.put(error, new ErrorMeter(name, error)))
- def requestRate(version: Short): Meter = {
- requestRateInternal.getAndMaybePut(version, newMeter(RequestsPerSec, "requests", TimeUnit.SECONDS, tags + ("version" -> version.toString)))
+ def requestRate(version: Short): Meter =
+ requestRateInternal.getAndMaybePut(version, metricsGroup.newMeter(RequestsPerSec, "requests", TimeUnit.SECONDS, tagsWithVersion(version)))
+
+ def tagsWithVersion(version: Short): java.util.Map[String, String] = {
+ val tags1 = new util.HashMap(tags)
Review Comment:
Maybe name this `nameAndVersionTags` or something better than `tags1`.
##########
core/src/main/scala/kafka/server/DelayedProduce.scala:
##########
@@ -129,15 +129,16 @@ class DelayedProduce(delayMs: Long,
}
}
-object DelayedProduceMetrics extends KafkaMetricsGroup {
+object DelayedProduceMetrics {
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
Review Comment:
Pass `classOf[DelayedProduceMetrics]`.
##########
core/src/main/scala/kafka/tools/MirrorMaker.scala:
##########
@@ -61,7 +61,8 @@ import scala.util.{Failure, Success, Try}
* @deprecated Since 3.0, use the Connect-based MirrorMaker instead (aka MM2).
*/
@deprecated(message = "Use the Connect-based MirrorMaker instead (aka MM2).", since = "3.0")
-object MirrorMaker extends Logging with KafkaMetricsGroup {
+object MirrorMaker extends Logging {
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
Review Comment:
Pass `classOf[MirrorMaker]`.
##########
core/src/main/scala/kafka/server/DelayedFetch.scala:
##########
@@ -180,9 +181,10 @@ class DelayedFetch(
}
}
-object DelayedFetchMetrics extends KafkaMetricsGroup {
+object DelayedFetchMetrics {
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
Review Comment:
Pass `classOf[DelayedFetchMetrics]`.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1072224151
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
Review Comment:
Can be static, yes. Made it so
--
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
[GitHub] [kafka] ijuma commented on pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ijuma commented on PR #13067:
URL: https://github.com/apache/kafka/pull/13067#issuecomment-1370707620
@ivanyu Does this add value over the approach used by `QuorumControllerMetrics`? I haven't compared the approaches closely yet, but interested in your thoughts.
--
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
[GitHub] [kafka] lbownik commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
lbownik commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1061466176
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
+ final String name, final Map<String, String> tags) {
+ final StringBuilder nameBuilder = new StringBuilder();
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ final String scope = toScope(tags).orElse(null);
+ final Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ return new MetricName(group, typeName, name, scope, nameBuilder.toString());
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newGauge(metricName(name, tags), metric);
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final MetricName metricName, final String eventType, final TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(final String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased) {
+ return newHistogram(name, biased, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final Map<String, String> tags) {
+ return newHistogram(name, true, tags);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit,
+ final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newTimer(metricName(name, tags), durationUnit, rateUnit);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit) {
+ return newTimer(name, durationUnit, rateUnit, Collections.emptyMap());
+ }
+
+ public final void removeMetric(final String name, final Map<String, String> tags) {
+ KafkaYammerMetrics.defaultRegistry().removeMetric(metricName(name, tags));
+ }
+
+ public final void removeMetric(final String name) {
+ removeMetric(name, Collections.emptyMap());
+ }
+
+ private Optional<String> toMBeanName(final Map<String, String> tags) {
+ final List<Map.Entry<String, String>> filteredTags = tags.entrySet().stream()
Review Comment:
you can improve performance (avoid unccecesary colletion to list) by writing
final String tags = tags.entrySet().stream()
.filter(entry -> !entry.getValue().equals(""))
.map(entry -> String.format("%s=%s", entry.getKey(), Sanitizer.jmxSanitize(entry.getValue())))
.collect(Collectors.joining(","));
return tags.isEmpty()? Optional.empty() : Optional.of(tagsString);
maybe cust returning "tags" would be simplier.
--
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
[GitHub] [kafka] satishd commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1127226556
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
Review Comment:
+1 Have a simpler statetment like below.
```
String pkg = klass.getPackage() == null ? "" : klass.getPackage().getName();
```
--
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
[GitHub] [kafka] satishd commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1127222852
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
Review Comment:
+1 Have a simpler statetment like below.
```
String pkg = klass.getPackage() == null ? "" : klass.getPackage().getName();
```
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1069826661
##########
core/src/test/scala/unit/kafka/log/LogCleanerIntegrationTest.scala:
##########
@@ -34,7 +33,7 @@ import scala.jdk.CollectionConverters._
/**
* This is an integration test that tests the fully integrated log cleaner
*/
-class LogCleanerIntegrationTest extends AbstractLogCleanerIntegrationTest with KafkaMetricsGroup {
+class LogCleanerIntegrationTest extends AbstractLogCleanerIntegrationTest {
Review Comment:
`KafkaMetricsGroup` wasn't used here.
--
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
[GitHub] [kafka] satishd commented on pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on PR #13067:
URL: https://github.com/apache/kafka/pull/13067#issuecomment-1450196726
> @satishd can you help review this?
@ijuma Sure, I will have cycles after this week. I will review it by next Monday.
--
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
[GitHub] [kafka] satishd merged pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd merged PR #13067:
URL: https://github.com/apache/kafka/pull/13067
--
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
[GitHub] [kafka] ivanyu commented on pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ivanyu (via GitHub)" <gi...@apache.org>.
ivanyu commented on PR #13067:
URL: https://github.com/apache/kafka/pull/13067#issuecomment-1442380235
Comments addressed, conflicts resolved.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1061663838
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
+ final String name, final Map<String, String> tags) {
+ final StringBuilder nameBuilder = new StringBuilder();
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ final String scope = toScope(tags).orElse(null);
+ final Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ return new MetricName(group, typeName, name, scope, nameBuilder.toString());
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newGauge(metricName(name, tags), metric);
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final MetricName metricName, final String eventType, final TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(final String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased) {
+ return newHistogram(name, biased, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final Map<String, String> tags) {
+ return newHistogram(name, true, tags);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit,
+ final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newTimer(metricName(name, tags), durationUnit, rateUnit);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit) {
+ return newTimer(name, durationUnit, rateUnit, Collections.emptyMap());
+ }
+
+ public final void removeMetric(final String name, final Map<String, String> tags) {
+ KafkaYammerMetrics.defaultRegistry().removeMetric(metricName(name, tags));
+ }
+
+ public final void removeMetric(final String name) {
+ removeMetric(name, Collections.emptyMap());
+ }
+
+ private Optional<String> toMBeanName(final Map<String, String> tags) {
+ final List<Map.Entry<String, String>> filteredTags = tags.entrySet().stream()
Review Comment:
This makes sense, I just tried to reproduce it relatively verbatim.
--
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
[GitHub] [kafka] ijuma commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ijuma commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1070526324
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
+ final String name, final Map<String, String> tags) {
+ final StringBuilder nameBuilder = new StringBuilder();
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ final String scope = toScope(tags).orElse(null);
+ final Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ return new MetricName(group, typeName, name, scope, nameBuilder.toString());
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newGauge(metricName(name, tags), metric);
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final MetricName metricName, final String eventType, final TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(final String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased) {
+ return newHistogram(name, biased, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final Map<String, String> tags) {
+ return newHistogram(name, true, tags);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit,
+ final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newTimer(metricName(name, tags), durationUnit, rateUnit);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit) {
+ return newTimer(name, durationUnit, rateUnit, Collections.emptyMap());
+ }
+
+ public final void removeMetric(final String name, final Map<String, String> tags) {
+ KafkaYammerMetrics.defaultRegistry().removeMetric(metricName(name, tags));
+ }
+
+ public final void removeMetric(final String name) {
+ removeMetric(name, Collections.emptyMap());
+ }
+
+ private Optional<String> toMBeanName(final Map<String, String> tags) {
Review Comment:
The ordering of `tags` matters when creating the mbean name. Unfortunately, we did not use an ordered map previously, which has resulted in unspecified behavior in some cases. When the number of tags is small, the order may have been maintained due to the Scala implementation having special implementations for these (eg https://www.scala-lang.org/api/2.13.x/scala/collection/immutable/Set$$Set1.html).
To make this change safe, we would need to have a test that prints _all_ mbean names and verifies that they remain the same after the conversion.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1069825536
##########
core/src/main/scala/kafka/log/remote/RemoteLogManager.scala:
##########
@@ -48,7 +47,7 @@ import scala.jdk.CollectionConverters._
*/
class RemoteLogManager(rlmConfig: RemoteLogManagerConfig,
brokerId: Int,
- logDir: String) extends Logging with Closeable with KafkaMetricsGroup {
Review Comment:
`KafkaMetricsGroup` wasn't used here.
--
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
[GitHub] [kafka] satishd commented on pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on PR #13067:
URL: https://github.com/apache/kafka/pull/13067#issuecomment-1442148593
@ivanyu Please resolve the conflicts and update the PR.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ivanyu (via GitHub)" <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1116206680
##########
core/src/main/scala/kafka/log/UnifiedLog.scala:
##########
@@ -247,10 +248,19 @@ class UnifiedLog(@volatile var logStartOffset: Long,
@volatile private var _topicId: Option[Uuid],
val keepPartitionMetadataFile: Boolean,
val remoteStorageSystemEnable: Boolean = false,
- remoteLogManager: Option[RemoteLogManager] = None) extends Logging with KafkaMetricsGroup {
+ remoteLogManager: Option[RemoteLogManager] = None) extends Logging {
import kafka.log.UnifiedLog._
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass) {
+ // For compatibility, metrics are defined to be under `Log` class
+ override def metricName(name: String, tags: util.Map[String, String]): MetricName = {
+ val pkg = getClass.getPackage
+ val pkgStr = if (pkg == null) "" else pkg.getName
Review Comment:
Sure, removed.
##########
core/src/main/scala/kafka/network/RequestChannel.scala:
##########
@@ -497,51 +501,58 @@ object RequestMetrics {
val ErrorsPerSec = "ErrorsPerSec"
}
-class RequestMetrics(name: String) extends KafkaMetricsGroup {
+class RequestMetrics(name: String) {
import RequestMetrics._
- val tags = Map("request" -> name)
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+
+ val tags = Map("request" -> name).asJava
val requestRateInternal = new Pool[Short, Meter]()
// time a request spent in a request queue
- val requestQueueTimeHist = newHistogram(RequestQueueTimeMs, biased = true, tags)
+ val requestQueueTimeHist = metricsGroup.newHistogram(RequestQueueTimeMs, true, tags)
// time a request takes to be processed at the local broker
- val localTimeHist = newHistogram(LocalTimeMs, biased = true, tags)
+ val localTimeHist = metricsGroup.newHistogram(LocalTimeMs, true, tags)
// time a request takes to wait on remote brokers (currently only relevant to fetch and produce requests)
- val remoteTimeHist = newHistogram(RemoteTimeMs, biased = true, tags)
+ val remoteTimeHist = metricsGroup.newHistogram(RemoteTimeMs, true, tags)
// time a request is throttled, not part of the request processing time (throttling is done at the client level
// for clients that support KIP-219 and by muting the channel for the rest)
- val throttleTimeHist = newHistogram(ThrottleTimeMs, biased = true, tags)
+ val throttleTimeHist = metricsGroup.newHistogram(ThrottleTimeMs, true, tags)
// time a response spent in a response queue
- val responseQueueTimeHist = newHistogram(ResponseQueueTimeMs, biased = true, tags)
+ val responseQueueTimeHist = metricsGroup.newHistogram(ResponseQueueTimeMs, true, tags)
// time to send the response to the requester
- val responseSendTimeHist = newHistogram(ResponseSendTimeMs, biased = true, tags)
- val totalTimeHist = newHistogram(TotalTimeMs, biased = true, tags)
+ val responseSendTimeHist = metricsGroup.newHistogram(ResponseSendTimeMs, true, tags)
+ val totalTimeHist = metricsGroup.newHistogram(TotalTimeMs, true, tags)
// request size in bytes
- val requestBytesHist = newHistogram(RequestBytes, biased = true, tags)
+ val requestBytesHist = metricsGroup.newHistogram(RequestBytes, true, tags)
// time for message conversions (only relevant to fetch and produce requests)
val messageConversionsTimeHist =
if (name == ApiKeys.FETCH.name || name == ApiKeys.PRODUCE.name)
- Some(newHistogram(MessageConversionsTimeMs, biased = true, tags))
+ Some(metricsGroup.newHistogram(MessageConversionsTimeMs, true, tags))
else
None
// Temporary memory allocated for processing request (only populated for fetch and produce requests)
// This shows the memory allocated for compression/conversions excluding the actual request size
val tempMemoryBytesHist =
if (name == ApiKeys.FETCH.name || name == ApiKeys.PRODUCE.name)
- Some(newHistogram(TemporaryMemoryBytes, biased = true, tags))
+ Some(metricsGroup.newHistogram(TemporaryMemoryBytes, true, tags))
else
None
private val errorMeters = mutable.Map[Errors, ErrorMeter]()
Errors.values.foreach(error => errorMeters.put(error, new ErrorMeter(name, error)))
- def requestRate(version: Short): Meter = {
- requestRateInternal.getAndMaybePut(version, newMeter(RequestsPerSec, "requests", TimeUnit.SECONDS, tags + ("version" -> version.toString)))
+ def requestRate(version: Short): Meter =
+ requestRateInternal.getAndMaybePut(version, metricsGroup.newMeter(RequestsPerSec, "requests", TimeUnit.SECONDS, tagsWithVersion(version)))
+
+ def tagsWithVersion(version: Short): java.util.Map[String, String] = {
+ val tags1 = new util.HashMap(tags)
Review Comment:
Yeah, thanks. Done
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ivanyu (via GitHub)" <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1126570518
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(String name, Map<String, String> tags) {
+ String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public static MetricName explicitMetricName(String group, String typeName,
+ String name, Map<String, String> tags) {
+ StringBuilder nameBuilder = new StringBuilder(100);
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ String scope = toScope(tags).orElse(null);
+ Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ 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 final <T> Gauge<T> newGauge(String name, Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(String name, String eventType,
+ TimeUnit timeUnit, Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(String name, String eventType,
+ TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(MetricName metricName, String eventType, TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(String name, boolean biased, Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(String name, boolean biased) {
Review Comment:
No, not really of this variant and the other one you mentioned. Removed 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
[GitHub] [kafka] lbownik commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
lbownik commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1061455377
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
Review Comment:
ternary operator would look great here
--
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
[GitHub] [kafka] satishd commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1125990307
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(String name, Map<String, String> tags) {
+ String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public static MetricName explicitMetricName(String group, String typeName,
+ String name, Map<String, String> tags) {
+ StringBuilder nameBuilder = new StringBuilder(100);
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ String scope = toScope(tags).orElse(null);
+ Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ 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 final <T> Gauge<T> newGauge(String name, Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(String name, String eventType,
+ TimeUnit timeUnit, Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(String name, String eventType,
+ TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(MetricName metricName, String eventType, TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(String name, boolean biased, Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(String name, boolean biased) {
Review Comment:
Are there any usages for this method? We can add it later if needed.
##########
core/src/main/scala/kafka/network/RequestChannel.scala:
##########
@@ -497,51 +501,59 @@ object RequestMetrics {
val ErrorsPerSec = "ErrorsPerSec"
}
-class RequestMetrics(name: String) extends KafkaMetricsGroup {
+class RequestMetrics(name: String) {
import RequestMetrics._
- val tags = Map("request" -> name)
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+
+ val tags = Map("request" -> name).asJava
val requestRateInternal = new Pool[Short, Meter]()
// time a request spent in a request queue
- val requestQueueTimeHist = newHistogram(RequestQueueTimeMs, biased = true, tags)
+ val requestQueueTimeHist = metricsGroup.newHistogram(RequestQueueTimeMs, true, tags)
// time a request takes to be processed at the local broker
- val localTimeHist = newHistogram(LocalTimeMs, biased = true, tags)
+ val localTimeHist = metricsGroup.newHistogram(LocalTimeMs, true, tags)
// time a request takes to wait on remote brokers (currently only relevant to fetch and produce requests)
- val remoteTimeHist = newHistogram(RemoteTimeMs, biased = true, tags)
+ val remoteTimeHist = metricsGroup.newHistogram(RemoteTimeMs, true, tags)
// time a request is throttled, not part of the request processing time (throttling is done at the client level
// for clients that support KIP-219 and by muting the channel for the rest)
- val throttleTimeHist = newHistogram(ThrottleTimeMs, biased = true, tags)
+ val throttleTimeHist = metricsGroup.newHistogram(ThrottleTimeMs, true, tags)
// time a response spent in a response queue
- val responseQueueTimeHist = newHistogram(ResponseQueueTimeMs, biased = true, tags)
+ val responseQueueTimeHist = metricsGroup.newHistogram(ResponseQueueTimeMs, true, tags)
// time to send the response to the requester
- val responseSendTimeHist = newHistogram(ResponseSendTimeMs, biased = true, tags)
- val totalTimeHist = newHistogram(TotalTimeMs, biased = true, tags)
+ val responseSendTimeHist = metricsGroup.newHistogram(ResponseSendTimeMs, true, tags)
+ val totalTimeHist = metricsGroup.newHistogram(TotalTimeMs, true, tags)
// request size in bytes
- val requestBytesHist = newHistogram(RequestBytes, biased = true, tags)
+ val requestBytesHist = metricsGroup.newHistogram(RequestBytes, true, tags)
// time for message conversions (only relevant to fetch and produce requests)
val messageConversionsTimeHist =
if (name == ApiKeys.FETCH.name || name == ApiKeys.PRODUCE.name)
- Some(newHistogram(MessageConversionsTimeMs, biased = true, tags))
+ Some(metricsGroup.newHistogram(MessageConversionsTimeMs, true, tags))
else
None
// Temporary memory allocated for processing request (only populated for fetch and produce requests)
// This shows the memory allocated for compression/conversions excluding the actual request size
val tempMemoryBytesHist =
if (name == ApiKeys.FETCH.name || name == ApiKeys.PRODUCE.name)
- Some(newHistogram(TemporaryMemoryBytes, biased = true, tags))
+ Some(metricsGroup.newHistogram(TemporaryMemoryBytes, true, tags))
else
None
private val errorMeters = mutable.Map[Errors, ErrorMeter]()
Errors.values.foreach(error => errorMeters.put(error, new ErrorMeter(name, error)))
- def requestRate(version: Short): Meter = {
- requestRateInternal.getAndMaybePut(version, newMeter(RequestsPerSec, "requests", TimeUnit.SECONDS, tags + ("version" -> version.toString)))
+ def requestRate(version: Short): Meter =
+ requestRateInternal.getAndMaybePut(version, metricsGroup.newMeter(RequestsPerSec, "requests", TimeUnit.SECONDS, tagsWithVersion(version)))
+
+ def tagsWithVersion(version: Short): java.util.Map[String, String] = {
Review Comment:
It can be a private method as it is accessed only with in the class.
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(String name, Map<String, String> tags) {
+ String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public static MetricName explicitMetricName(String group, String typeName,
+ String name, Map<String, String> tags) {
+ StringBuilder nameBuilder = new StringBuilder(100);
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ String scope = toScope(tags).orElse(null);
+ Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ 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 final <T> Gauge<T> newGauge(String name, Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(String name, String eventType,
+ TimeUnit timeUnit, Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(String name, String eventType,
+ TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(MetricName metricName, String eventType, TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(String name, boolean biased, Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(String name, boolean biased) {
+ return newHistogram(name, biased, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(String name, Map<String, String> tags) {
Review Comment:
Are there any usages for this method? We can add it later if needed.
--
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
[GitHub] [kafka] ijuma commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ijuma commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1070521647
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -293,14 +295,14 @@ class Partition(val topicPartition: TopicPartition,
private var controllerEpoch: Int = KafkaController.InitialControllerEpoch
this.logIdent = s"[Partition $topicPartition broker=$localBrokerId] "
- private val tags = Map("topic" -> topic, "partition" -> partitionId.toString)
+ private val tags = Map("topic" -> topic, "partition" -> partitionId.toString).asJava
- newGauge("UnderReplicated", () => if (isUnderReplicated) 1 else 0, tags)
- newGauge("InSyncReplicasCount", () => if (isLeader) partitionState.isr.size else 0, tags)
- newGauge("UnderMinIsr", () => if (isUnderMinIsr) 1 else 0, tags)
- newGauge("AtMinIsr", () => if (isAtMinIsr) 1 else 0, tags)
- newGauge("ReplicasCount", () => if (isLeader) assignmentState.replicationFactor else 0, tags)
- newGauge("LastStableOffsetLag", () => log.map(_.lastStableOffsetLag).getOrElse(0), tags)
+ Partition.metricsGroup.newGauge("UnderReplicated", () => if (isUnderReplicated) 1 else 0, tags)
Review Comment:
`Partition.` should not be needed.
##########
core/src/test/scala/unit/kafka/metrics/MetricsTest.scala:
##########
@@ -158,7 +159,7 @@ class MetricsTest extends KafkaServerTestHarness with Logging {
val path = "C:\\windows-path\\kafka-logs"
val tags = Map("dir" -> path)
val expectedMBeanName = Set(tags.keySet.head, ObjectName.quote(path)).mkString("=")
- val metric = KafkaMetricsGroup.metricName("test-metric", tags)
+ val metric = new KafkaMetricsGroup(this.getClass).metricName("test-metric", tags.asJava)
Review Comment:
Worth noting that this is slightly different than the old code, but seems ok for what we're testing here.
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
Review Comment:
Can we please file a JIRA for removing this? I don't think it's required if we don't pass the class from Scala objects (versus classes).
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
Review Comment:
We don't use `final` for parameters and local variables in Kafka (the stream modules are an exception).
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
+ final String name, final Map<String, String> tags) {
+ final StringBuilder nameBuilder = new StringBuilder();
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ final String scope = toScope(tags).orElse(null);
+ final Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ return new MetricName(group, typeName, name, scope, nameBuilder.toString());
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newGauge(metricName(name, tags), metric);
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final MetricName metricName, final String eventType, final TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(final String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased) {
+ return newHistogram(name, biased, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final Map<String, String> tags) {
+ return newHistogram(name, true, tags);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit,
+ final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newTimer(metricName(name, tags), durationUnit, rateUnit);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit) {
+ return newTimer(name, durationUnit, rateUnit, Collections.emptyMap());
+ }
+
+ public final void removeMetric(final String name, final Map<String, String> tags) {
+ KafkaYammerMetrics.defaultRegistry().removeMetric(metricName(name, tags));
+ }
+
+ public final void removeMetric(final String name) {
+ removeMetric(name, Collections.emptyMap());
+ }
+
+ private Optional<String> toMBeanName(final Map<String, String> tags) {
+ final List<Map.Entry<String, String>> filteredTags = tags.entrySet().stream()
+ .filter(entry -> !entry.getValue().equals(""))
+ .collect(Collectors.toList());
+ if (!filteredTags.isEmpty()) {
+ final String tagsString = filteredTags.stream()
+ .map(entry -> String.format("%s=%s", entry.getKey(), Sanitizer.jmxSanitize(entry.getValue())))
Review Comment:
Let's use string concat (i.e. `+`) instead of `String.format`, the former is more efficient.
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
Review Comment:
Should this be a static method? Does it rely on any class state?
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
+ final String name, final Map<String, String> tags) {
+ final StringBuilder nameBuilder = new StringBuilder();
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ final String scope = toScope(tags).orElse(null);
+ final Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ return new MetricName(group, typeName, name, scope, nameBuilder.toString());
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newGauge(metricName(name, tags), metric);
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final MetricName metricName, final String eventType, final TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(final String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased) {
+ return newHistogram(name, biased, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final Map<String, String> tags) {
+ return newHistogram(name, true, tags);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit,
+ final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newTimer(metricName(name, tags), durationUnit, rateUnit);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit) {
+ return newTimer(name, durationUnit, rateUnit, Collections.emptyMap());
+ }
+
+ public final void removeMetric(final String name, final Map<String, String> tags) {
+ KafkaYammerMetrics.defaultRegistry().removeMetric(metricName(name, tags));
+ }
+
+ public final void removeMetric(final String name) {
+ removeMetric(name, Collections.emptyMap());
+ }
+
+ private Optional<String> toMBeanName(final Map<String, String> tags) {
Review Comment:
The ordering of `tags` matters when creating the mbean name. Unfortunately, we did not use an ordered map previously, which has resulted in unspecified behavior in some cases. When the number of tags is small, the order may have been maintained due to the Scala implementation having special implementations for thise (eg https://www.scala-lang.org/api/2.13.x/scala/collection/immutable/Set$$Set1.html).
To make this change safe, we would need to have a test that prints _all_ mbean names and verifies that they remain the same after the conversion.
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -70,7 +70,9 @@ class DelayedOperations(topicPartition: TopicPartition,
def numDelayedDelete: Int = deleteRecords.numDelayed
}
-object Partition extends KafkaMetricsGroup {
+object Partition {
+ private val metricsGroup = new KafkaMetricsGroup(classOf[Partition])
Review Comment:
I think we can call this and other similar `val`s `metrics`. It's a bit more concise and the `Group` suffix doesn't add any particular value.
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
+ final String name, final Map<String, String> tags) {
+ final StringBuilder nameBuilder = new StringBuilder();
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ final String scope = toScope(tags).orElse(null);
+ final Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ return new MetricName(group, typeName, name, scope, nameBuilder.toString());
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newGauge(metricName(name, tags), metric);
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final MetricName metricName, final String eventType, final TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(final String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased) {
+ return newHistogram(name, biased, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final Map<String, String> tags) {
+ return newHistogram(name, true, tags);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit,
+ final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newTimer(metricName(name, tags), durationUnit, rateUnit);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit) {
+ return newTimer(name, durationUnit, rateUnit, Collections.emptyMap());
+ }
+
+ public final void removeMetric(final String name, final Map<String, String> tags) {
+ KafkaYammerMetrics.defaultRegistry().removeMetric(metricName(name, tags));
+ }
+
+ public final void removeMetric(final String name) {
+ removeMetric(name, Collections.emptyMap());
+ }
+
+ private Optional<String> toMBeanName(final Map<String, String> tags) {
+ final List<Map.Entry<String, String>> filteredTags = tags.entrySet().stream()
Review Comment:
We don't have to verbatim if it means unnecessary inefficiency.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1070539477
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -70,7 +70,9 @@ class DelayedOperations(topicPartition: TopicPartition,
def numDelayedDelete: Int = deleteRecords.numDelayed
}
-object Partition extends KafkaMetricsGroup {
+object Partition {
+ private val metricsGroup = new KafkaMetricsGroup(classOf[Partition])
Review Comment:
This works for some classes, but some other (e.g. `KafkaBroker`, `KafkaController`) already have the `metrics` field. Maybe let's keep the suffix for consistency?
--
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
[GitHub] [kafka] ivanyu commented on pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ivanyu (via GitHub)" <gi...@apache.org>.
ivanyu commented on PR #13067:
URL: https://github.com/apache/kafka/pull/13067#issuecomment-1418658516
I've fixed the conflicts. @ijuma could you please have a look if you have time?
--
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
[GitHub] [kafka] ijuma commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1103755706
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
Review Comment:
If you pass `classOf[Partition]`, then this code is not needed. It's only needed if you pass `getClass` from within a scala `object` (not `class`).
--
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
[GitHub] [kafka] ijuma commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1103755774
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -70,7 +70,9 @@ class DelayedOperations(topicPartition: TopicPartition,
def numDelayedDelete: Int = deleteRecords.numDelayed
}
-object Partition extends KafkaMetricsGroup {
+object Partition {
+ private val metricsGroup = new KafkaMetricsGroup(classOf[Partition])
Review Comment:
That's fine.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ivanyu (via GitHub)" <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1120551247
##########
core/src/main/scala/kafka/network/RequestChannel.scala:
##########
@@ -497,51 +501,59 @@ object RequestMetrics {
val ErrorsPerSec = "ErrorsPerSec"
}
-class RequestMetrics(name: String) extends KafkaMetricsGroup {
+class RequestMetrics(name: String) {
import RequestMetrics._
- val tags = Map("request" -> name)
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+
+ val tags = Map("request" -> name).asJava
val requestRateInternal = new Pool[Short, Meter]()
// time a request spent in a request queue
- val requestQueueTimeHist = newHistogram(RequestQueueTimeMs, biased = true, tags)
+ val requestQueueTimeHist = metricsGroup.newHistogram(RequestQueueTimeMs, true, tags)
// time a request takes to be processed at the local broker
- val localTimeHist = newHistogram(LocalTimeMs, biased = true, tags)
+ val localTimeHist = metricsGroup.newHistogram(LocalTimeMs, true, tags)
// time a request takes to wait on remote brokers (currently only relevant to fetch and produce requests)
- val remoteTimeHist = newHistogram(RemoteTimeMs, biased = true, tags)
+ val remoteTimeHist = metricsGroup.newHistogram(RemoteTimeMs, true, tags)
// time a request is throttled, not part of the request processing time (throttling is done at the client level
// for clients that support KIP-219 and by muting the channel for the rest)
- val throttleTimeHist = newHistogram(ThrottleTimeMs, biased = true, tags)
+ val throttleTimeHist = metricsGroup.newHistogram(ThrottleTimeMs, true, tags)
// time a response spent in a response queue
- val responseQueueTimeHist = newHistogram(ResponseQueueTimeMs, biased = true, tags)
+ val responseQueueTimeHist = metricsGroup.newHistogram(ResponseQueueTimeMs, true, tags)
// time to send the response to the requester
- val responseSendTimeHist = newHistogram(ResponseSendTimeMs, biased = true, tags)
- val totalTimeHist = newHistogram(TotalTimeMs, biased = true, tags)
+ val responseSendTimeHist = metricsGroup.newHistogram(ResponseSendTimeMs, true, tags)
+ val totalTimeHist = metricsGroup.newHistogram(TotalTimeMs, true, tags)
// request size in bytes
- val requestBytesHist = newHistogram(RequestBytes, biased = true, tags)
+ val requestBytesHist = metricsGroup.newHistogram(RequestBytes, true, tags)
// time for message conversions (only relevant to fetch and produce requests)
val messageConversionsTimeHist =
if (name == ApiKeys.FETCH.name || name == ApiKeys.PRODUCE.name)
- Some(newHistogram(MessageConversionsTimeMs, biased = true, tags))
+ Some(metricsGroup.newHistogram(MessageConversionsTimeMs, true, tags))
else
None
// Temporary memory allocated for processing request (only populated for fetch and produce requests)
// This shows the memory allocated for compression/conversions excluding the actual request size
val tempMemoryBytesHist =
if (name == ApiKeys.FETCH.name || name == ApiKeys.PRODUCE.name)
- Some(newHistogram(TemporaryMemoryBytes, biased = true, tags))
+ Some(metricsGroup.newHistogram(TemporaryMemoryBytes, true, tags))
else
None
private val errorMeters = mutable.Map[Errors, ErrorMeter]()
Errors.values.foreach(error => errorMeters.put(error, new ErrorMeter(name, error)))
- def requestRate(version: Short): Meter = {
- requestRateInternal.getAndMaybePut(version, newMeter(RequestsPerSec, "requests", TimeUnit.SECONDS, tags + ("version" -> version.toString)))
+ def requestRate(version: Short): Meter =
+ requestRateInternal.getAndMaybePut(version, metricsGroup.newMeter(RequestsPerSec, "requests", TimeUnit.SECONDS, tagsWithVersion(version)))
+
+ def tagsWithVersion(version: Short): java.util.Map[String, String] = {
+ val nameAndVersionTags = new util.HashMap[String, String]()
Review Comment:
I didn't do this because considered this an infrequent operation, but sure, have done this now.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1069827409
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##########
@@ -42,7 +43,7 @@ class BrokerMetadataListener(
val snapshotter: Option[MetadataSnapshotter],
brokerMetrics: BrokerServerMetrics,
_metadataLoadingFaultHandler: FaultHandler
-) extends RaftClient.Listener[ApiMessageAndVersion] with KafkaMetricsGroup {
+) extends RaftClient.Listener[ApiMessageAndVersion] with Logging {
Review Comment:
`KafkaMetricsGroup` wasn't used here.
##########
core/src/main/scala/kafka/server/ZkAdminManager.scala:
##########
@@ -69,7 +68,7 @@ object ZkAdminManager {
class ZkAdminManager(val config: KafkaConfig,
val metrics: Metrics,
val metadataCache: MetadataCache,
- val zkClient: KafkaZkClient) extends Logging with KafkaMetricsGroup {
+ val zkClient: KafkaZkClient) extends Logging {
Review Comment:
`KafkaMetricsGroup` wasn't used here.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1070962067
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
+ final String name, final Map<String, String> tags) {
+ final StringBuilder nameBuilder = new StringBuilder();
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ final String scope = toScope(tags).orElse(null);
+ final Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ return new MetricName(group, typeName, name, scope, nameBuilder.toString());
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newGauge(metricName(name, tags), metric);
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final MetricName metricName, final String eventType, final TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(final String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased) {
+ return newHistogram(name, biased, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final Map<String, String> tags) {
+ return newHistogram(name, true, tags);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit,
+ final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newTimer(metricName(name, tags), durationUnit, rateUnit);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit) {
+ return newTimer(name, durationUnit, rateUnit, Collections.emptyMap());
+ }
+
+ public final void removeMetric(final String name, final Map<String, String> tags) {
+ KafkaYammerMetrics.defaultRegistry().removeMetric(metricName(name, tags));
+ }
+
+ public final void removeMetric(final String name) {
+ removeMetric(name, Collections.emptyMap());
+ }
+
+ private Optional<String> toMBeanName(final Map<String, String> tags) {
Review Comment:
> The ordering of `tags` matters when creating the mbean name.
Does it? According to the doc [the doc](https://www.oracle.com/java/technologies/javase/management-extensions-best-practices.html#mozTocId434075), "the order of key properties is not significant". I.e. `topic=my-topic,partition=0` is equivalent to `partition=0,topic=my-topic`. AFAICS from the OpenJDK [code](https://github.com/openjdk/jdk/blob/master/src/java.management/share/classes/javax/management/ObjectName.java#L753), it's canonicalized anyway.
--
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
[GitHub] [kafka] ijuma commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1103748681
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
+ final String name, final Map<String, String> tags) {
+ final StringBuilder nameBuilder = new StringBuilder();
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ final String scope = toScope(tags).orElse(null);
+ final Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ return new MetricName(group, typeName, name, scope, nameBuilder.toString());
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newGauge(metricName(name, tags), metric);
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final MetricName metricName, final String eventType, final TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(final String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased) {
+ return newHistogram(name, biased, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final Map<String, String> tags) {
+ return newHistogram(name, true, tags);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit,
+ final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newTimer(metricName(name, tags), durationUnit, rateUnit);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit) {
+ return newTimer(name, durationUnit, rateUnit, Collections.emptyMap());
+ }
+
+ public final void removeMetric(final String name, final Map<String, String> tags) {
+ KafkaYammerMetrics.defaultRegistry().removeMetric(metricName(name, tags));
+ }
+
+ public final void removeMetric(final String name) {
+ removeMetric(name, Collections.emptyMap());
+ }
+
+ private Optional<String> toMBeanName(final Map<String, String> tags) {
Review Comment:
Good point, my bad.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ivanyu (via GitHub)" <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1127463069
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
Review Comment:
Done
--
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
[GitHub] [kafka] lbownik commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
lbownik commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1061458021
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
+ final String name, final Map<String, String> tags) {
+ final StringBuilder nameBuilder = new StringBuilder();
Review Comment:
the default capacity of 16 will cause internal buffere realocation,, how about putting new StringBuilder(50) or even new StringBuilder(100) here?
--
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
[GitHub] [kafka] ivanyu commented on pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on PR #13067:
URL: https://github.com/apache/kafka/pull/13067#issuecomment-1369441609
@ijuma @satishd what do you think about this approach to rewriting `KafkaMetricsGroup` in Java? Currently, classes-users of the `KafkaMetricsGroup` trait sometimes override its methods to tune its behavior (mostly, names). Instead, they can create anonymous subclasses. In this PR I only demonstrate the idea. If you think it's viable, I can spread it and get rid of the `KafkaMetricsGroup` trait.
--
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
[GitHub] [kafka] fvaleri commented on pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
fvaleri commented on PR #13067:
URL: https://github.com/apache/kafka/pull/13067#issuecomment-1370083367
Hi @ivanyu, thanks for looking into this.
>Currently, classes-users of the KafkaMetricsGroup trait sometimes override its methods to tune its behavior (mostly, names).
If that's the case, why not directly use `explicitMetricName`?
--
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
[GitHub] [kafka] ijuma commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1120488775
##########
core/src/main/scala/kafka/network/RequestChannel.scala:
##########
@@ -497,51 +501,59 @@ object RequestMetrics {
val ErrorsPerSec = "ErrorsPerSec"
}
-class RequestMetrics(name: String) extends KafkaMetricsGroup {
+class RequestMetrics(name: String) {
import RequestMetrics._
- val tags = Map("request" -> name)
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+
+ val tags = Map("request" -> name).asJava
val requestRateInternal = new Pool[Short, Meter]()
// time a request spent in a request queue
- val requestQueueTimeHist = newHistogram(RequestQueueTimeMs, biased = true, tags)
+ val requestQueueTimeHist = metricsGroup.newHistogram(RequestQueueTimeMs, true, tags)
// time a request takes to be processed at the local broker
- val localTimeHist = newHistogram(LocalTimeMs, biased = true, tags)
+ val localTimeHist = metricsGroup.newHistogram(LocalTimeMs, true, tags)
// time a request takes to wait on remote brokers (currently only relevant to fetch and produce requests)
- val remoteTimeHist = newHistogram(RemoteTimeMs, biased = true, tags)
+ val remoteTimeHist = metricsGroup.newHistogram(RemoteTimeMs, true, tags)
// time a request is throttled, not part of the request processing time (throttling is done at the client level
// for clients that support KIP-219 and by muting the channel for the rest)
- val throttleTimeHist = newHistogram(ThrottleTimeMs, biased = true, tags)
+ val throttleTimeHist = metricsGroup.newHistogram(ThrottleTimeMs, true, tags)
// time a response spent in a response queue
- val responseQueueTimeHist = newHistogram(ResponseQueueTimeMs, biased = true, tags)
+ val responseQueueTimeHist = metricsGroup.newHistogram(ResponseQueueTimeMs, true, tags)
// time to send the response to the requester
- val responseSendTimeHist = newHistogram(ResponseSendTimeMs, biased = true, tags)
- val totalTimeHist = newHistogram(TotalTimeMs, biased = true, tags)
+ val responseSendTimeHist = metricsGroup.newHistogram(ResponseSendTimeMs, true, tags)
+ val totalTimeHist = metricsGroup.newHistogram(TotalTimeMs, true, tags)
// request size in bytes
- val requestBytesHist = newHistogram(RequestBytes, biased = true, tags)
+ val requestBytesHist = metricsGroup.newHistogram(RequestBytes, true, tags)
// time for message conversions (only relevant to fetch and produce requests)
val messageConversionsTimeHist =
if (name == ApiKeys.FETCH.name || name == ApiKeys.PRODUCE.name)
- Some(newHistogram(MessageConversionsTimeMs, biased = true, tags))
+ Some(metricsGroup.newHistogram(MessageConversionsTimeMs, true, tags))
else
None
// Temporary memory allocated for processing request (only populated for fetch and produce requests)
// This shows the memory allocated for compression/conversions excluding the actual request size
val tempMemoryBytesHist =
if (name == ApiKeys.FETCH.name || name == ApiKeys.PRODUCE.name)
- Some(newHistogram(TemporaryMemoryBytes, biased = true, tags))
+ Some(metricsGroup.newHistogram(TemporaryMemoryBytes, true, tags))
else
None
private val errorMeters = mutable.Map[Errors, ErrorMeter]()
Errors.values.foreach(error => errorMeters.put(error, new ErrorMeter(name, error)))
- def requestRate(version: Short): Meter = {
- requestRateInternal.getAndMaybePut(version, newMeter(RequestsPerSec, "requests", TimeUnit.SECONDS, tags + ("version" -> version.toString)))
+ def requestRate(version: Short): Meter =
+ requestRateInternal.getAndMaybePut(version, metricsGroup.newMeter(RequestsPerSec, "requests", TimeUnit.SECONDS, tagsWithVersion(version)))
+
+ def tagsWithVersion(version: Short): java.util.Map[String, String] = {
+ val nameAndVersionTags = new util.HashMap[String, String]()
Review Comment:
Maybe we can presize the map correctly.
--
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
[GitHub] [kafka] lbownik commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "lbownik (via GitHub)" <gi...@apache.org>.
lbownik commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1120661170
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(String name, Map<String, String> tags) {
+ String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public static MetricName explicitMetricName(String group, String typeName,
+ String name, Map<String, String> tags) {
+ StringBuilder nameBuilder = new StringBuilder();
Review Comment:
maybe initialize the builder with some initial calacity (default is 16 - a little low).
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ivanyu (via GitHub)" <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1116207172
##########
core/src/main/scala/kafka/server/DelayedDeleteRecords.scala:
##########
@@ -122,12 +121,12 @@ class DelayedDeleteRecords(delayMs: Long,
}
}
-object DelayedDeleteRecordsMetrics extends KafkaMetricsGroup {
+object DelayedDeleteRecordsMetrics {
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
Review Comment:
Same as with `LogFlushStats`.
##########
core/src/main/scala/kafka/server/DelayedFetch.scala:
##########
@@ -180,9 +181,10 @@ class DelayedFetch(
}
}
-object DelayedFetchMetrics extends KafkaMetricsGroup {
+object DelayedFetchMetrics {
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
Review Comment:
Same as with `LogFlushStats`.
##########
core/src/main/scala/kafka/server/DelayedProduce.scala:
##########
@@ -129,15 +129,16 @@ class DelayedProduce(delayMs: Long,
}
}
-object DelayedProduceMetrics extends KafkaMetricsGroup {
+object DelayedProduceMetrics {
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
Review Comment:
Same as with `LogFlushStats`.
##########
core/src/main/scala/kafka/tools/MirrorMaker.scala:
##########
@@ -61,7 +61,8 @@ import scala.util.{Failure, Success, Try}
* @deprecated Since 3.0, use the Connect-based MirrorMaker instead (aka MM2).
*/
@deprecated(message = "Use the Connect-based MirrorMaker instead (aka MM2).", since = "3.0")
-object MirrorMaker extends Logging with KafkaMetricsGroup {
+object MirrorMaker extends Logging {
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
Review Comment:
Same as with `LogFlushStats`.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ivanyu (via GitHub)" <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1116205167
##########
core/src/main/scala/kafka/log/LogSegment.scala:
##########
@@ -689,6 +689,7 @@ object LogSegment {
}
}
-object LogFlushStats extends KafkaMetricsGroup {
- val logFlushTimer: Timer = newTimer("LogFlushRateAndTimeMs", TimeUnit.MILLISECONDS, TimeUnit.SECONDS)
+object LogFlushStats {
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
Review Comment:
Since it's an object without the class, `classOf` won't work. I replaced with `LogFlushStats.getClass` here and in other similar places you mentioned.
##########
core/src/main/scala/kafka/log/LogSegment.scala:
##########
@@ -689,6 +689,7 @@ object LogSegment {
}
}
-object LogFlushStats extends KafkaMetricsGroup {
- val logFlushTimer: Timer = newTimer("LogFlushRateAndTimeMs", TimeUnit.MILLISECONDS, TimeUnit.SECONDS)
+object LogFlushStats {
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
Review Comment:
Since it's an object without the class, `classOf` won't work. I replaced with `LogFlushStats.getClass` here and in other similar places you mentioned. Makes sense?
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ivanyu (via GitHub)" <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1126571253
##########
core/src/main/scala/kafka/network/RequestChannel.scala:
##########
@@ -497,51 +501,59 @@ object RequestMetrics {
val ErrorsPerSec = "ErrorsPerSec"
}
-class RequestMetrics(name: String) extends KafkaMetricsGroup {
+class RequestMetrics(name: String) {
import RequestMetrics._
- val tags = Map("request" -> name)
+ private val metricsGroup = new KafkaMetricsGroup(this.getClass)
+
+ val tags = Map("request" -> name).asJava
val requestRateInternal = new Pool[Short, Meter]()
// time a request spent in a request queue
- val requestQueueTimeHist = newHistogram(RequestQueueTimeMs, biased = true, tags)
+ val requestQueueTimeHist = metricsGroup.newHistogram(RequestQueueTimeMs, true, tags)
// time a request takes to be processed at the local broker
- val localTimeHist = newHistogram(LocalTimeMs, biased = true, tags)
+ val localTimeHist = metricsGroup.newHistogram(LocalTimeMs, true, tags)
// time a request takes to wait on remote brokers (currently only relevant to fetch and produce requests)
- val remoteTimeHist = newHistogram(RemoteTimeMs, biased = true, tags)
+ val remoteTimeHist = metricsGroup.newHistogram(RemoteTimeMs, true, tags)
// time a request is throttled, not part of the request processing time (throttling is done at the client level
// for clients that support KIP-219 and by muting the channel for the rest)
- val throttleTimeHist = newHistogram(ThrottleTimeMs, biased = true, tags)
+ val throttleTimeHist = metricsGroup.newHistogram(ThrottleTimeMs, true, tags)
// time a response spent in a response queue
- val responseQueueTimeHist = newHistogram(ResponseQueueTimeMs, biased = true, tags)
+ val responseQueueTimeHist = metricsGroup.newHistogram(ResponseQueueTimeMs, true, tags)
// time to send the response to the requester
- val responseSendTimeHist = newHistogram(ResponseSendTimeMs, biased = true, tags)
- val totalTimeHist = newHistogram(TotalTimeMs, biased = true, tags)
+ val responseSendTimeHist = metricsGroup.newHistogram(ResponseSendTimeMs, true, tags)
+ val totalTimeHist = metricsGroup.newHistogram(TotalTimeMs, true, tags)
// request size in bytes
- val requestBytesHist = newHistogram(RequestBytes, biased = true, tags)
+ val requestBytesHist = metricsGroup.newHistogram(RequestBytes, true, tags)
// time for message conversions (only relevant to fetch and produce requests)
val messageConversionsTimeHist =
if (name == ApiKeys.FETCH.name || name == ApiKeys.PRODUCE.name)
- Some(newHistogram(MessageConversionsTimeMs, biased = true, tags))
+ Some(metricsGroup.newHistogram(MessageConversionsTimeMs, true, tags))
else
None
// Temporary memory allocated for processing request (only populated for fetch and produce requests)
// This shows the memory allocated for compression/conversions excluding the actual request size
val tempMemoryBytesHist =
if (name == ApiKeys.FETCH.name || name == ApiKeys.PRODUCE.name)
- Some(newHistogram(TemporaryMemoryBytes, biased = true, tags))
+ Some(metricsGroup.newHistogram(TemporaryMemoryBytes, true, tags))
else
None
private val errorMeters = mutable.Map[Errors, ErrorMeter]()
Errors.values.foreach(error => errorMeters.put(error, new ErrorMeter(name, error)))
- def requestRate(version: Short): Meter = {
- requestRateInternal.getAndMaybePut(version, newMeter(RequestsPerSec, "requests", TimeUnit.SECONDS, tags + ("version" -> version.toString)))
+ def requestRate(version: Short): Meter =
+ requestRateInternal.getAndMaybePut(version, metricsGroup.newMeter(RequestsPerSec, "requests", TimeUnit.SECONDS, tagsWithVersion(version)))
+
+ def tagsWithVersion(version: Short): java.util.Map[String, String] = {
Review Comment:
You're right, did this
--
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
[GitHub] [kafka] ivanyu commented on pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on PR #13067:
URL: https://github.com/apache/kafka/pull/13067#issuecomment-1370097055
There are four places where `KafkaMetricsGroup` methods are overridden, it's always `metricName`. Sometimes it comes with some logic like in `UnifiedLog`. We of course can change this in various ways, but probably it's a good idea to keep it as is for the initial refactoring.
--
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
[GitHub] [kafka] ivanyu commented on pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ivanyu (via GitHub)" <gi...@apache.org>.
ivanyu commented on PR #13067:
URL: https://github.com/apache/kafka/pull/13067#issuecomment-1457689418
Thank you @satishd! I addressed your last comment.
> Please have incremental commits instead of a single squashed/merged commit. This is helpful for reviewers to review the incremental changes done in the PR that address the review comments.
I relied on Github being able to show the diff on force-push, but sure, I'll do incremental commits and can squash before merge.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1072223766
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
+ final String name, final Map<String, String> tags) {
+ final StringBuilder nameBuilder = new StringBuilder();
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ final String scope = toScope(tags).orElse(null);
+ final Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ return new MetricName(group, typeName, name, scope, nameBuilder.toString());
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newGauge(metricName(name, tags), metric);
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final MetricName metricName, final String eventType, final TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(final String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased) {
+ return newHistogram(name, biased, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final Map<String, String> tags) {
+ return newHistogram(name, true, tags);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit,
+ final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newTimer(metricName(name, tags), durationUnit, rateUnit);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit) {
+ return newTimer(name, durationUnit, rateUnit, Collections.emptyMap());
+ }
+
+ public final void removeMetric(final String name, final Map<String, String> tags) {
+ KafkaYammerMetrics.defaultRegistry().removeMetric(metricName(name, tags));
+ }
+
+ public final void removeMetric(final String name) {
+ removeMetric(name, Collections.emptyMap());
+ }
+
+ private Optional<String> toMBeanName(final Map<String, String> tags) {
+ final List<Map.Entry<String, String>> filteredTags = tags.entrySet().stream()
+ .filter(entry -> !entry.getValue().equals(""))
+ .collect(Collectors.toList());
+ if (!filteredTags.isEmpty()) {
+ final String tagsString = filteredTags.stream()
+ .map(entry -> String.format("%s=%s", entry.getKey(), Sanitizer.jmxSanitize(entry.getValue())))
Review Comment:
Ok, done this
--
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
[GitHub] [kafka] lbownik commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
lbownik commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1061467816
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
+ final String name, final Map<String, String> tags) {
+ final StringBuilder nameBuilder = new StringBuilder();
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ final String scope = toScope(tags).orElse(null);
+ final Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ return new MetricName(group, typeName, name, scope, nameBuilder.toString());
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newGauge(metricName(name, tags), metric);
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final MetricName metricName, final String eventType, final TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(final String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased) {
+ return newHistogram(name, biased, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final Map<String, String> tags) {
+ return newHistogram(name, true, tags);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit,
+ final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newTimer(metricName(name, tags), durationUnit, rateUnit);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit) {
+ return newTimer(name, durationUnit, rateUnit, Collections.emptyMap());
+ }
+
+ public final void removeMetric(final String name, final Map<String, String> tags) {
+ KafkaYammerMetrics.defaultRegistry().removeMetric(metricName(name, tags));
+ }
+
+ public final void removeMetric(final String name) {
+ removeMetric(name, Collections.emptyMap());
+ }
+
+ private Optional<String> toMBeanName(final Map<String, String> tags) {
+ final List<Map.Entry<String, String>> filteredTags = tags.entrySet().stream()
+ .filter(entry -> !entry.getValue().equals(""))
+ .collect(Collectors.toList());
+ if (!filteredTags.isEmpty()) {
+ final String tagsString = filteredTags.stream()
+ .map(entry -> String.format("%s=%s", entry.getKey(), Sanitizer.jmxSanitize(entry.getValue())))
+ .collect(Collectors.joining(","));
+ return Optional.of(tagsString);
+ } else {
+ return Optional.empty();
+ }
+ }
+
+ private Optional<String> toScope(final Map<String, String> tags) {
+ final List<Map.Entry<String, String>> filteredTags = tags.entrySet().stream()
Review Comment:
you can improve performance (avoid unccecesary colletion to list) by writing
final String tags = tags.entrySet().stream()
.filter(entry -> !entry.getValue().equals(""))
.sorted(Map.Entry.comparingByKey())
.map(entry -> String.format("%s.%s", entry.getKey(), entry.getValue().replaceAll("\\.", "_")))
.collect(Collectors.joining("."));
return tags.isEmpty()? Optional.empty() : Optional.of(tagsString);
maybe cust returning "tags" would be simplier.
--
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
[GitHub] [kafka] satishd commented on pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on PR #13067:
URL: https://github.com/apache/kafka/pull/13067#issuecomment-1457307583
@ivanyu Please have incremental commits instead of a single squashed/merged commit. This is helpful for reviewers to review the incremental changes done in the PR that address the review comments.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "ivanyu (via GitHub)" <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1121367465
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(String name, Map<String, String> tags) {
+ String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public static MetricName explicitMetricName(String group, String typeName,
+ String name, Map<String, String> tags) {
+ StringBuilder nameBuilder = new StringBuilder();
Review Comment:
This line will execute only few times during the process run time, but sure, I can put some arbitrary number there.
--
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
[GitHub] [kafka] satishd commented on pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on PR #13067:
URL: https://github.com/apache/kafka/pull/13067#issuecomment-1459952200
A couple of failed tests in the jobs do not seem to be related to this change, merging the changes to the trunk.
--
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
[GitHub] [kafka] ijuma commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ijuma commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1070523929
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
+ return explicitMetricName(pkg, simpleName, name, tags);
+ }
+
+ public final MetricName explicitMetricName(final String group, final String typeName,
+ final String name, final Map<String, String> tags) {
+ final StringBuilder nameBuilder = new StringBuilder();
+ nameBuilder.append(group);
+ nameBuilder.append(":type=");
+ nameBuilder.append(typeName);
+
+ if (!name.isEmpty()) {
+ nameBuilder.append(",name=");
+ nameBuilder.append(name);
+ }
+
+ final String scope = toScope(tags).orElse(null);
+ final Optional<String> tagsName = toMBeanName(tags);
+ tagsName.ifPresent(s -> nameBuilder.append(",").append(s));
+
+ return new MetricName(group, typeName, name, scope, nameBuilder.toString());
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newGauge(metricName(name, tags), metric);
+ }
+
+ public final <T> Gauge<T> newGauge(final String name, final Gauge<T> metric) {
+ return newGauge(name, metric, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName(name, tags), eventType, timeUnit);
+ }
+
+ public final Meter newMeter(final String name, final String eventType,
+ final TimeUnit timeUnit) {
+ return newMeter(name, eventType, timeUnit, Collections.emptyMap());
+ }
+
+ public final Meter newMeter(final MetricName metricName, final String eventType, final TimeUnit timeUnit) {
+ return KafkaYammerMetrics.defaultRegistry().newMeter(metricName, eventType, timeUnit);
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased, final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newHistogram(metricName(name, tags), biased);
+ }
+
+ public final Histogram newHistogram(final String name) {
+ return newHistogram(name, true, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final boolean biased) {
+ return newHistogram(name, biased, Collections.emptyMap());
+ }
+
+ public final Histogram newHistogram(final String name, final Map<String, String> tags) {
+ return newHistogram(name, true, tags);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit,
+ final Map<String, String> tags) {
+ return KafkaYammerMetrics.defaultRegistry().newTimer(metricName(name, tags), durationUnit, rateUnit);
+ }
+
+ public final Timer newTimer(final String name, final TimeUnit durationUnit, final TimeUnit rateUnit) {
+ return newTimer(name, durationUnit, rateUnit, Collections.emptyMap());
+ }
+
+ public final void removeMetric(final String name, final Map<String, String> tags) {
+ KafkaYammerMetrics.defaultRegistry().removeMetric(metricName(name, tags));
+ }
+
+ public final void removeMetric(final String name) {
+ removeMetric(name, Collections.emptyMap());
+ }
+
+ private Optional<String> toMBeanName(final Map<String, String> tags) {
+ final List<Map.Entry<String, String>> filteredTags = tags.entrySet().stream()
Review Comment:
We don't have to do it verbatim if it means unnecessary inefficiency.
--
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
[GitHub] [kafka] ivanyu commented on pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on PR #13067:
URL: https://github.com/apache/kafka/pull/13067#issuecomment-1382303264
I've finished porting `KafkaMetricsGroup` to Java and I'm opening this for review.
I decided to not implement the performance improvements @lbownik proposed above and stick to verbatim porting to limit the review surface. The methods are far from the hot path so their performance doesn't affect much.
--
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
[GitHub] [kafka] ivanyu commented on a diff in pull request #13067: KAFKA-14524: Rewrite KafkaMetricsGroup in Java
Posted by GitBox <gi...@apache.org>.
ivanyu commented on code in PR #13067:
URL: https://github.com/apache/kafka/pull/13067#discussion_r1070536098
##########
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.kafka.server.metrics;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Histogram;
+import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.utils.Sanitizer;
+
+public class KafkaMetricsGroup {
+ private final Class<?> klass;
+
+ public KafkaMetricsGroup(final Class<?> klass) {
+ this.klass = klass;
+ }
+
+ /**
+ * Creates a new MetricName object for gauges, meters, etc. created for this
+ * metrics group.
+ * @param name Descriptive name of the metric.
+ * @param tags Additional attributes which mBean will have.
+ * @return Sanitized metric name object.
+ */
+ public MetricName metricName(final String name, final Map<String, String> tags) {
+ final String pkg;
+ if (klass.getPackage() == null) {
+ pkg = "";
+ } else {
+ pkg = klass.getPackage().getName();
+ }
+ final String simpleName = klass.getSimpleName().replaceAll("\\$$", "");
Review Comment:
Do you mean, this JIRA must be implemented _after_ we don't have `KafkaMetricsGroup` instantiated in Scala objects (currently we have it in a couple of places, e.g. `Partition` and `DelayedDeleteRecordsMetrics`)? Also alternatively, it can be done by overriding `metricName` in objects
--
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