RockteMQ-AI commented on code in PR #1339: URL: https://github.com/apache/rocketmq-clients/pull/1339#discussion_r3783663991
########## java/client/src/main/java/org/apache/rocketmq/client/java/metrics/ClientJmxReporter.java: ########## @@ -0,0 +1,421 @@ +/* + * 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.rocketmq.client.java.metrics; + +import io.opentelemetry.api.common.Attributes; +import java.lang.management.ManagementFactory; +import java.math.BigDecimal; +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.DoubleAdder; +import java.util.concurrent.atomic.LongAdder; +import java.util.regex.Pattern; +import javax.management.Attribute; +import javax.management.AttributeList; +import javax.management.AttributeNotFoundException; +import javax.management.DynamicMBean; +import javax.management.InvalidAttributeValueException; +import javax.management.JMException; +import javax.management.MBeanAttributeInfo; +import javax.management.MBeanException; +import javax.management.MBeanInfo; +import javax.management.MBeanOperationInfo; +import javax.management.MBeanServer; +import javax.management.ObjectName; +import javax.management.ReflectionException; +import org.apache.rocketmq.client.java.misc.ClientId; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Registers RocketMQ client metrics in the platform MBean server, following Kafka's JMX reporter model. + * + * <p>The reporter does not open a network port or depend on a Prometheus library. When enabled, applications that + * already run a Prometheus JMX exporter can collect these MBeans. The reporter is disabled by default and can be + * enabled before creating a client with the {@code rocketmq.client.jmx.enabled} system property. + */ +final class ClientJmxReporter { + static final String DOMAIN = "org.apache.rocketmq.client"; + static final String TYPE = "message-metrics"; + static final String ENABLE_PROPERTY = "rocketmq.client.jmx.enabled"; + private static final Logger log = LoggerFactory.getLogger(ClientJmxReporter.class); + private static final Pattern SAFE_OBJECT_NAME_VALUE = Pattern.compile("[\\w-%\\. \\t]*"); + private static final MBeanOperationInfo[] NO_OPERATIONS = new MBeanOperationInfo[0]; + + private final ClientId clientId; + private final MBeanServer mBeanServer; + private final AtomicBoolean enabled; + private final Object registrationLock = new Object(); + private final Map<ObjectName, ClientMetricsMBean> mBeans; + private final Map<HistogramEnum, ConcurrentMap<Attributes, JmxHistogram>> histograms; + private volatile GaugeObserver gaugeObserver = GaugeObserver.EMPTY; + + ClientJmxReporter(ClientId clientId) { + this.clientId = clientId; + MBeanServer server = null; + boolean initialized = false; + try { + if (Boolean.parseBoolean(System.getProperty(ENABLE_PROPERTY, Boolean.FALSE.toString()))) { + server = ManagementFactory.getPlatformMBeanServer(); + initialized = true; + } + } catch (Throwable t) { + log.warn("Failed to initialize the client JMX reporter, clientId={}", clientId, t); + } + this.mBeanServer = server; + this.enabled = new AtomicBoolean(initialized); + if (initialized) { + this.mBeans = new HashMap<>(); + this.histograms = new EnumMap<>(HistogramEnum.class); + for (HistogramEnum histogram : HistogramEnum.values()) { + this.histograms.put(histogram, new ConcurrentHashMap<>()); + } + } else { + this.mBeans = Collections.emptyMap(); + this.histograms = Collections.emptyMap(); + } + } + + boolean isEnabled() { + return enabled.get(); + } + + void setGaugeObserver(GaugeObserver gaugeObserver) { + this.gaugeObserver = gaugeObserver; + } + + void record(HistogramEnum histogramType, Attributes attributes, double value) { + if (!enabled.get()) { + return; + } + ConcurrentMap<Attributes, JmxHistogram> series = histograms.get(histogramType); + JmxHistogram histogram = series.get(attributes); + if (null == histogram) { + histogram = registerHistogram(histogramType, attributes, series); + } + if (null != histogram) { + histogram.record(value); + } + } + + void refreshGauges() { + if (!enabled.get()) { + return; + } + GaugeObserver observer = gaugeObserver; + try { + for (GaugeEnum gauge : observer.getGauges()) { + Map<Attributes, Double> values = observer.getValues(gauge); + for (Attributes attributes : values.keySet()) { + registerGauge(gauge, attributes); + } + } + } catch (RuntimeException e) { + log.warn("Failed to refresh client JMX gauges, clientId={}", clientId, e); + } + } + + void shutdown() { + if (!enabled.compareAndSet(true, false)) { + return; + } + synchronized (registrationLock) { + for (ObjectName objectName : new ArrayList<>(mBeans.keySet())) { + try { + if (mBeanServer.isRegistered(objectName)) { + mBeanServer.unregisterMBean(objectName); + } + } catch (JMException | RuntimeException e) { + log.warn("Failed to unregister client metrics MBean, objectName={}, clientId={}", + objectName, clientId, e); + } + } + mBeans.clear(); + } + } + + private JmxHistogram registerHistogram(HistogramEnum histogramType, Attributes attributes, + ConcurrentMap<Attributes, JmxHistogram> series) { + synchronized (registrationLock) { + JmxHistogram existed = series.get(attributes); + if (null != existed) { + return existed; + } + JmxHistogram histogram = new JmxHistogram(histogramType.getBoundaries()); + Map<String, MetricValue> metrics = new LinkedHashMap<>(); + String prefix = histogramType.getName(); + metrics.put(prefix + "_count", new LongMetricValue(histogram.count::sum)); + metrics.put(prefix + "_sum", new DoubleMetricValue(histogram.sum::sum)); + for (int i = 0; i < histogram.boundaries.size(); i++) { + final int bucketIndex = i; + String boundary = formatBoundary(histogram.boundaries.get(i)); + metrics.put(prefix + "_bucket_le_" + boundary, + new LongMetricValue(() -> histogram.cumulativeBucketCount(bucketIndex))); + } + metrics.put(prefix + "_bucket_le_inf", new LongMetricValue(histogram.count::sum)); + if (!registerMetrics(attributes, metrics)) { + return null; + } + series.put(attributes, histogram); + return histogram; + } + } + + private void registerGauge(GaugeEnum gauge, Attributes attributes) { + Map<String, MetricValue> metrics = Collections.singletonMap(gauge.getName(), + new DoubleMetricValue(() -> getGaugeValue(gauge, attributes))); + registerMetrics(attributes, metrics); + } + + private double getGaugeValue(GaugeEnum gauge, Attributes attributes) { + try { + Double value = gaugeObserver.getValues(gauge).get(attributes); + return null == value ? 0 : value; + } catch (RuntimeException e) { + log.warn("Failed to read client JMX gauge, gauge={}, clientId={}", gauge, clientId, e); + return 0; + } + } + + private boolean registerMetrics(Attributes attributes, Map<String, MetricValue> metrics) { + synchronized (registrationLock) { + try { + ObjectName objectName = objectName(attributes); + ClientMetricsMBean mBean = mBeans.get(objectName); + if (null == mBean) { + mBean = new ClientMetricsMBean(); + mBean.putAllIfAbsent(metrics); + mBeanServer.registerMBean(mBean, objectName); + mBeans.put(objectName, mBean); + return true; + } + boolean changed = mBean.putAllIfAbsent(metrics); + if (!changed && mBeanServer.isRegistered(objectName)) { + return true; + } + if (mBeanServer.isRegistered(objectName)) { + mBeanServer.unregisterMBean(objectName); + } + mBeanServer.registerMBean(mBean, objectName); + return true; + } catch (JMException | RuntimeException e) { + log.warn("Failed to register client metrics MBean, clientId={}", clientId, e); + return false; + } + } + } + + ObjectName objectName(Attributes attributes) throws JMException { + SortedMap<String, String> labels = new TreeMap<>(); + attributes.forEach((key, value) -> { + if (null != value && !String.valueOf(value).isEmpty()) { + labels.put(key.getKey(), String.valueOf(value)); + } + }); + labels.put(MetricLabels.CLIENT_ID.getKey(), clientId.toString()); + StringBuilder builder = new StringBuilder(DOMAIN).append(":type=").append(TYPE); + for (Map.Entry<String, String> entry : labels.entrySet()) { + builder.append(',').append(entry.getKey()).append('=').append(sanitize(entry.getValue())); + } + return new ObjectName(builder.toString()); + } + + private static String sanitize(String value) { + return SAFE_OBJECT_NAME_VALUE.matcher(value).matches() ? value : ObjectName.quote(value); + } + + private static String formatBoundary(double value) { + // Encode signs and decimal points as JMX-friendly suffixes, for example, -1.5 becomes n1_5. + String boundary = BigDecimal.valueOf(value).stripTrailingZeros().toPlainString(); + return boundary.replace('-', 'n').replace('.', '_'); + } + + private interface LongValueSupplier { + long get(); + } + + private interface DoubleValueSupplier { + double get(); + } + + private interface MetricValue { + Number get(); + + String getType(); + } + + private static final class LongMetricValue implements MetricValue { + private final LongValueSupplier supplier; + + private LongMetricValue(LongValueSupplier supplier) { + this.supplier = supplier; + } + + @Override + public Number get() { + return supplier.get(); + } + + @Override + public String getType() { + return Long.class.getName(); + } + } + + private static final class DoubleMetricValue implements MetricValue { + private final DoubleValueSupplier supplier; + + private DoubleMetricValue(DoubleValueSupplier supplier) { + this.supplier = supplier; + } + + @Override + public Number get() { + return supplier.get(); + } + + @Override + public String getType() { + return Double.class.getName(); + } + } + + private static final class JmxHistogram { + private final List<Double> boundaries; + private final LongAdder count = new LongAdder(); + private final DoubleAdder sum = new DoubleAdder(); + private final LongAdder[] buckets; + + private JmxHistogram(List<Double> boundaries) { + this.boundaries = boundaries; + this.buckets = new LongAdder[boundaries.size() + 1]; + for (int i = 0; i < buckets.length; i++) { + buckets[i] = new LongAdder(); Review Comment: **[Info]** The `cumulativeBucketCount` method iterates from bucket 0 to `bucketIndex` on each call, which is O(n). For JMX polling this is acceptable, but if this becomes a hot path, consider maintaining a running cumulative sum. ########## java/client/src/main/java/org/apache/rocketmq/client/java/metrics/ClientJmxReporter.java: ########## @@ -0,0 +1,421 @@ +/* + * 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.rocketmq.client.java.metrics; + +import io.opentelemetry.api.common.Attributes; +import java.lang.management.ManagementFactory; +import java.math.BigDecimal; +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.SortedMap; +import java.util.TreeMap; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.DoubleAdder; +import java.util.concurrent.atomic.LongAdder; +import java.util.regex.Pattern; +import javax.management.Attribute; +import javax.management.AttributeList; +import javax.management.AttributeNotFoundException; +import javax.management.DynamicMBean; +import javax.management.InvalidAttributeValueException; +import javax.management.JMException; +import javax.management.MBeanAttributeInfo; +import javax.management.MBeanException; +import javax.management.MBeanInfo; +import javax.management.MBeanOperationInfo; +import javax.management.MBeanServer; +import javax.management.ObjectName; +import javax.management.ReflectionException; +import org.apache.rocketmq.client.java.misc.ClientId; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Registers RocketMQ client metrics in the platform MBean server, following Kafka's JMX reporter model. + * + * <p>The reporter does not open a network port or depend on a Prometheus library. When enabled, applications that + * already run a Prometheus JMX exporter can collect these MBeans. The reporter is disabled by default and can be + * enabled before creating a client with the {@code rocketmq.client.jmx.enabled} system property. + */ +final class ClientJmxReporter { + static final String DOMAIN = "org.apache.rocketmq.client"; + static final String TYPE = "message-metrics"; + static final String ENABLE_PROPERTY = "rocketmq.client.jmx.enabled"; + private static final Logger log = LoggerFactory.getLogger(ClientJmxReporter.class); + private static final Pattern SAFE_OBJECT_NAME_VALUE = Pattern.compile("[\\w-%\\. \\t]*"); + private static final MBeanOperationInfo[] NO_OPERATIONS = new MBeanOperationInfo[0]; + + private final ClientId clientId; + private final MBeanServer mBeanServer; + private final AtomicBoolean enabled; + private final Object registrationLock = new Object(); + private final Map<ObjectName, ClientMetricsMBean> mBeans; + private final Map<HistogramEnum, ConcurrentMap<Attributes, JmxHistogram>> histograms; + private volatile GaugeObserver gaugeObserver = GaugeObserver.EMPTY; + + ClientJmxReporter(ClientId clientId) { + this.clientId = clientId; + MBeanServer server = null; + boolean initialized = false; + try { + if (Boolean.parseBoolean(System.getProperty(ENABLE_PROPERTY, Boolean.FALSE.toString()))) { + server = ManagementFactory.getPlatformMBeanServer(); + initialized = true; + } + } catch (Throwable t) { + log.warn("Failed to initialize the client JMX reporter, clientId={}", clientId, t); + } + this.mBeanServer = server; + this.enabled = new AtomicBoolean(initialized); + if (initialized) { + this.mBeans = new HashMap<>(); + this.histograms = new EnumMap<>(HistogramEnum.class); + for (HistogramEnum histogram : HistogramEnum.values()) { + this.histograms.put(histogram, new ConcurrentHashMap<>()); + } + } else { + this.mBeans = Collections.emptyMap(); + this.histograms = Collections.emptyMap(); + } + } + + boolean isEnabled() { + return enabled.get(); + } + + void setGaugeObserver(GaugeObserver gaugeObserver) { + this.gaugeObserver = gaugeObserver; + } + + void record(HistogramEnum histogramType, Attributes attributes, double value) { + if (!enabled.get()) { + return; + } + ConcurrentMap<Attributes, JmxHistogram> series = histograms.get(histogramType); + JmxHistogram histogram = series.get(attributes); + if (null == histogram) { + histogram = registerHistogram(histogramType, attributes, series); + } + if (null != histogram) { + histogram.record(value); + } + } + + void refreshGauges() { + if (!enabled.get()) { + return; + } + GaugeObserver observer = gaugeObserver; + try { + for (GaugeEnum gauge : observer.getGauges()) { + Map<Attributes, Double> values = observer.getValues(gauge); + for (Attributes attributes : values.keySet()) { + registerGauge(gauge, attributes); + } + } + } catch (RuntimeException e) { + log.warn("Failed to refresh client JMX gauges, clientId={}", clientId, e); + } + } + + void shutdown() { + if (!enabled.compareAndSet(true, false)) { + return; + } + synchronized (registrationLock) { + for (ObjectName objectName : new ArrayList<>(mBeans.keySet())) { + try { + if (mBeanServer.isRegistered(objectName)) { + mBeanServer.unregisterMBean(objectName); + } + } catch (JMException | RuntimeException e) { + log.warn("Failed to unregister client metrics MBean, objectName={}, clientId={}", + objectName, clientId, e); + } + } + mBeans.clear(); + } + } + + private JmxHistogram registerHistogram(HistogramEnum histogramType, Attributes attributes, + ConcurrentMap<Attributes, JmxHistogram> series) { + synchronized (registrationLock) { + JmxHistogram existed = series.get(attributes); + if (null != existed) { + return existed; + } + JmxHistogram histogram = new JmxHistogram(histogramType.getBoundaries()); + Map<String, MetricValue> metrics = new LinkedHashMap<>(); + String prefix = histogramType.getName(); + metrics.put(prefix + "_count", new LongMetricValue(histogram.count::sum)); + metrics.put(prefix + "_sum", new DoubleMetricValue(histogram.sum::sum)); + for (int i = 0; i < histogram.boundaries.size(); i++) { + final int bucketIndex = i; + String boundary = formatBoundary(histogram.boundaries.get(i)); + metrics.put(prefix + "_bucket_le_" + boundary, + new LongMetricValue(() -> histogram.cumulativeBucketCount(bucketIndex))); + } + metrics.put(prefix + "_bucket_le_inf", new LongMetricValue(histogram.count::sum)); + if (!registerMetrics(attributes, metrics)) { + return null; + } + series.put(attributes, histogram); + return histogram; + } + } + + private void registerGauge(GaugeEnum gauge, Attributes attributes) { + Map<String, MetricValue> metrics = Collections.singletonMap(gauge.getName(), + new DoubleMetricValue(() -> getGaugeValue(gauge, attributes))); + registerMetrics(attributes, metrics); + } + + private double getGaugeValue(GaugeEnum gauge, Attributes attributes) { + try { + Double value = gaugeObserver.getValues(gauge).get(attributes); + return null == value ? 0 : value; + } catch (RuntimeException e) { + log.warn("Failed to read client JMX gauge, gauge={}, clientId={}", gauge, clientId, e); + return 0; + } + } + + private boolean registerMetrics(Attributes attributes, Map<String, MetricValue> metrics) { + synchronized (registrationLock) { + try { + ObjectName objectName = objectName(attributes); + ClientMetricsMBean mBean = mBeans.get(objectName); + if (null == mBean) { + mBean = new ClientMetricsMBean(); + mBean.putAllIfAbsent(metrics); + mBeanServer.registerMBean(mBean, objectName); + mBeans.put(objectName, mBean); + return true; + } + boolean changed = mBean.putAllIfAbsent(metrics); + if (!changed && mBeanServer.isRegistered(objectName)) { + return true; + } + if (mBeanServer.isRegistered(objectName)) { + mBeanServer.unregisterMBean(objectName); + } + mBeanServer.registerMBean(mBean, objectName); + return true; + } catch (JMException | RuntimeException e) { + log.warn("Failed to register client metrics MBean, clientId={}", clientId, e); + return false; + } + } + } + + ObjectName objectName(Attributes attributes) throws JMException { + SortedMap<String, String> labels = new TreeMap<>(); + attributes.forEach((key, value) -> { + if (null != value && !String.valueOf(value).isEmpty()) { + labels.put(key.getKey(), String.valueOf(value)); + } + }); + labels.put(MetricLabels.CLIENT_ID.getKey(), clientId.toString()); + StringBuilder builder = new StringBuilder(DOMAIN).append(":type=").append(TYPE); + for (Map.Entry<String, String> entry : labels.entrySet()) { + builder.append(',').append(entry.getKey()).append('=').append(sanitize(entry.getValue())); + } + return new ObjectName(builder.toString()); + } + + private static String sanitize(String value) { + return SAFE_OBJECT_NAME_VALUE.matcher(value).matches() ? value : ObjectName.quote(value); + } + + private static String formatBoundary(double value) { + // Encode signs and decimal points as JMX-friendly suffixes, for example, -1.5 becomes n1_5. + String boundary = BigDecimal.valueOf(value).stripTrailingZeros().toPlainString(); + return boundary.replace('-', 'n').replace('.', '_'); + } + + private interface LongValueSupplier { + long get(); + } + + private interface DoubleValueSupplier { + double get(); + } + + private interface MetricValue { + Number get(); Review Comment: **[Info]** The `putAllIfAbsent` method only adds new metrics but never removes them. If metrics can be dynamically unregistered (e.g., when a consumer is shut down), consider adding a cleanup mechanism to avoid memory leaks in long-running applications. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
