From f1332af2d35c679b18d7228a70d3aee403e382df Mon Sep 17 00:00:00 2001 From: zentol Date: Tue, 2 May 2017 17:31:27 +0200 Subject: [PATCH 1/3] [FLINK-6431] [metrics] Activate strict checkstyle in flink-metrics --- .../apache/flink/metrics/CharacterFilter.java | 2 +- .../org/apache/flink/metrics/Histogram.java | 2 +- .../flink/metrics/HistogramStatistics.java | 6 +- .../org/apache/flink/metrics/MeterView.java | 23 +- .../apache/flink/metrics/MetricConfig.java | 1 + .../org/apache/flink/metrics/MetricGroup.java | 20 +- .../apache/flink/metrics/SimpleCounter.java | 3 +- .../java/org/apache/flink/metrics/View.java | 3 +- .../metrics/reporter/AbstractReporter.java | 1 + .../metrics/reporter/MetricReporter.java | 4 +- .../apache/flink/metrics/util/TestMeter.java | 3 + .../flink/metrics/datadog/DCounter.java | 8 +- .../apache/flink/metrics/datadog/DGauge.java | 9 +- .../apache/flink/metrics/datadog/DMeter.java | 6 +- .../apache/flink/metrics/datadog/DMetric.java | 8 +- .../apache/flink/metrics/datadog/DSeries.java | 8 +- .../metrics/datadog/DatadogHttpClient.java | 8 +- .../metrics/datadog/DatadogHttpReporter.java | 30 +- .../flink/metrics/datadog/MetricType.java | 8 +- .../datadog/DatadogHttpClientTest.java | 317 +++++++++--------- .../ScheduledDropwizardReporter.java | 12 +- .../DropwizardHistogramStatistics.java | 3 +- .../metrics/FlinkCounterWrapper.java | 4 + .../dropwizard/metrics/FlinkGaugeWrapper.java | 5 +- .../metrics/FlinkHistogramWrapper.java | 3 +- .../dropwizard/metrics/FlinkMeterWrapper.java | 3 +- .../metrics/HistogramStatisticsWrapper.java | 6 +- .../metrics/ganglia/GangliaReporter.java | 15 +- .../metrics/graphite/GraphiteReporter.java | 14 +- .../apache/flink/metrics/jmx/JMXReporter.java | 46 ++- .../flink/metrics/statsd/StatsDReporter.java | 18 +- .../metrics/statsd/StatsDReporterTest.java | 17 +- flink-metrics/pom.xml | 38 +++ 33 files changed, 362 insertions(+), 292 deletions(-) diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/CharacterFilter.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/CharacterFilter.java index 1e9fbc45468ab..10cd9ce692397 100644 --- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/CharacterFilter.java +++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/CharacterFilter.java @@ -27,7 +27,7 @@ public interface CharacterFilter { /** * Filter the given string and generate a resulting string from it. * - * For example, one implementation could filter out invalid characters from the input string. + *

For example, one implementation could filter out invalid characters from the input string. * * @param input Input string * @return Filtered result string diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/Histogram.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/Histogram.java index af5c9b0b50ecc..070a52ab78285 100644 --- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/Histogram.java +++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/Histogram.java @@ -21,7 +21,7 @@ /** * Histogram interface to be used with Flink's metrics system. * - * The histogram allows to record values, get the current count of recorded values and create + *

The histogram allows to record values, get the current count of recorded values and create * histogram statistics for the currently seen elements. */ public interface Histogram extends Metric { diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/HistogramStatistics.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/HistogramStatistics.java index b2e45074c38e8..7a4b0a9088924 100644 --- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/HistogramStatistics.java +++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/HistogramStatistics.java @@ -21,7 +21,7 @@ /** * Histogram statistics represent the current snapshot of elements recorded in the histogram. * - * The histogram statistics allow to calculate values for quantiles, the mean, the standard + *

The histogram statistics allow to calculate values for quantiles, the mean, the standard * deviation, the minimum and the maximum. */ public abstract class HistogramStatistics { @@ -35,14 +35,14 @@ public abstract class HistogramStatistics { public abstract double getQuantile(double quantile); /** - * Returns the elements of the statistics' sample + * Returns the elements of the statistics' sample. * * @return Elements of the statistics' sample */ public abstract long[] getValues(); /** - * Returns the size of the statistics' sample + * Returns the size of the statistics' sample. * * @return Size of the statistics' sample */ diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MeterView.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MeterView.java index b87b983e92c12..8df0e868f8f8e 100644 --- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MeterView.java +++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MeterView.java @@ -15,32 +15,33 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.flink.metrics; /** * A MeterView provides an average rate of events per second over a given time period. - * - * The primary advantage of this class is that the rate is neither updated by the computing thread nor for every event. + * + *

The primary advantage of this class is that the rate is neither updated by the computing thread nor for every event. * Instead, a history of counts is maintained that is updated in regular intervals by a background thread. From this * history a rate is derived on demand, which represents the average rate of events over the given time span. - * - * Setting the time span to a low value reduces memory-consumption and will more accurately report short-term changes. + * + *

Setting the time span to a low value reduces memory-consumption and will more accurately report short-term changes. * The minimum value possible is {@link View#UPDATE_INTERVAL_SECONDS}. * A high value in turn increases memory-consumption, since a longer history has to be maintained, but will result in * smoother transitions between rates. - * - * The events are counted by a {@link Counter}. + * + *

The events are counted by a {@link Counter}. */ public class MeterView implements Meter, View { - /** The underlying counter maintaining the count */ + /** The underlying counter maintaining the count. */ private final Counter counter; - /** The time-span over which the average is calculated */ + /** The time-span over which the average is calculated. */ private final int timeSpanInSeconds; - /** Circular array containing the history of values */ + /** Circular array containing the history of values. */ private final long[] values; - /** The index in the array for the current time */ + /** The index in the array for the current time. */ private int time = 0; - /** The last rate we computed */ + /** The last rate we computed. */ private double currentRate = 0; public MeterView(int timeSpanInSeconds) { diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricConfig.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricConfig.java index 4a2e616a3a37c..699afdfd3dc66 100644 --- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricConfig.java +++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricConfig.java @@ -15,6 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.flink.metrics; import java.util.Properties; diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricGroup.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricGroup.java index d4221ef76eb65..39ab3b60f2c93 100644 --- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricGroup.java +++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricGroup.java @@ -22,10 +22,10 @@ /** * A MetricGroup is a named container for {@link Metric Metrics} and further metric subgroups. - * + * *

Instances of this class can be used to register new metrics with Flink and to create a nested * hierarchy based on the group names. - * + * *

A MetricGroup is uniquely identified by it's place in the hierarchy and name. */ public interface MetricGroup { @@ -69,7 +69,7 @@ public interface MetricGroup { * @return the given counter */ C counter(String name, C counter); - + /** * Registers a new {@link org.apache.flink.metrics.Gauge} with Flink. * @@ -95,7 +95,7 @@ public interface MetricGroup { * * @param name name of the histogram * @param histogram histogram to register - * @param histogram type + * @param histogram type * @return the registered histogram */ H histogram(String name, H histogram); @@ -105,7 +105,7 @@ public interface MetricGroup { * * @param name name of the histogram * @param histogram histogram to register - * @param histogram type + * @param histogram type * @return the registered histogram */ H histogram(int name, H histogram); @@ -156,7 +156,7 @@ public interface MetricGroup { /** * Gets the scope as an array of the scope components, for example - * {@code ["host-7", "taskmanager-2", "window_word_count", "my-mapper"]} + * {@code ["host-7", "taskmanager-2", "window_word_count", "my-mapper"]}. * * @see #getMetricIdentifier(String) * @see #getMetricIdentifier(String, CharacterFilter) @@ -165,15 +165,15 @@ public interface MetricGroup { /** * Returns a map of all variables and their associated value, for example - * {@code {""="host-7", ""="taskmanager-2"}} - * + * {@code {""="host-7", ""="taskmanager-2"}}. + * * @return map of all variables and their associated value */ Map getAllVariables(); /** * Returns the fully qualified metric name, for example - * {@code "host-7.taskmanager-2.window_word_count.my-mapper.metricName"} + * {@code "host-7.taskmanager-2.window_word_count.my-mapper.metricName"}. * * @param metricName metric name * @return fully qualified metric name @@ -182,7 +182,7 @@ public interface MetricGroup { /** * Returns the fully qualified metric name, for example - * {@code "host-7.taskmanager-2.window_word_count.my-mapper.metricName"} + * {@code "host-7.taskmanager-2.window_word_count.my-mapper.metricName"}. * * @param metricName metric name * @param filter character filter which is applied to the scope components if not null. diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/SimpleCounter.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/SimpleCounter.java index 6ec3b28f189a9..2efc0c9f0dbf8 100644 --- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/SimpleCounter.java +++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/SimpleCounter.java @@ -15,6 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.flink.metrics; /** @@ -22,7 +23,7 @@ */ public class SimpleCounter implements Counter { - /** the current count */ + /** the current count. */ private long count; /** diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/View.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/View.java index 17801302596fb..2bbb3a922f0e4 100644 --- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/View.java +++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/View.java @@ -15,13 +15,14 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.flink.metrics; /** * An interface for metrics which should be updated in regular intervals by a background thread. */ public interface View { - /** The interval in which metrics are updated */ + /** The interval in which metrics are updated. */ int UPDATE_INTERVAL_SECONDS = 5; /** diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/reporter/AbstractReporter.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/reporter/AbstractReporter.java index 0c8d9ada0efa6..c0aeb4bc4eb32 100644 --- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/reporter/AbstractReporter.java +++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/reporter/AbstractReporter.java @@ -25,6 +25,7 @@ import org.apache.flink.metrics.Meter; import org.apache.flink.metrics.Metric; import org.apache.flink.metrics.MetricGroup; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/reporter/MetricReporter.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/reporter/MetricReporter.java index ee92a1057a9cd..5c8085fead028 100644 --- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/reporter/MetricReporter.java +++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/reporter/MetricReporter.java @@ -24,7 +24,7 @@ /** * Reporters are used to export {@link Metric Metrics} to an external backend. - * + * *

Reporters are instantiated via reflection and must be public, non-abstract, and have a * public no-argument constructor. */ @@ -37,7 +37,7 @@ public interface MetricReporter { /** * Configures this reporter. Since reporters are instantiated generically and hence parameter-less, * this method is the place where the reporters set their basic fields based on configuration values. - * + * *

This method is always called first on a newly instantiated reporter. * * @param config A properties object that contains all parameters set for this reporter. diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/util/TestMeter.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/util/TestMeter.java index b44b99688e838..b1ec3a3cf7c36 100644 --- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/util/TestMeter.java +++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/util/TestMeter.java @@ -20,6 +20,9 @@ import org.apache.flink.metrics.Meter; +/** + * A dummy {@link Meter} implementation. + */ public class TestMeter implements Meter { @Override diff --git a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DCounter.java b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DCounter.java index 58abbd6a4daec..e187efb54bc9d 100644 --- a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DCounter.java +++ b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DCounter.java @@ -23,8 +23,8 @@ import java.util.List; /** - * Mapping of counter between Flink and Datadog - * */ + * Mapping of counter between Flink and Datadog. + */ public class DCounter extends DMetric { private final Counter counter; @@ -35,8 +35,8 @@ public DCounter(Counter c, String metricName, String host, List tags) { /** * Visibility of this method must not be changed - * since we deliberately not map it to json object in a Datadog-defined format - * */ + * since we deliberately not map it to json object in a Datadog-defined format. + */ @Override public Number getMetricValue() { return counter.getCount(); diff --git a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DGauge.java b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DGauge.java index 8deb11758bb06..ba97d59327b51 100644 --- a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DGauge.java +++ b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DGauge.java @@ -18,14 +18,13 @@ package org.apache.flink.metrics.datadog; - import org.apache.flink.metrics.Gauge; import java.util.List; /** - * Mapping of gauge between Flink and Datadog - * */ + * Mapping of gauge between Flink and Datadog. + */ public class DGauge extends DMetric { private final Gauge gauge; @@ -36,8 +35,8 @@ public DGauge(Gauge g, String metricName, String host, List tags /** * Visibility of this method must not be changed - * since we deliberately not map it to json object in a Datadog-defined format - * */ + * since we deliberately not map it to json object in a Datadog-defined format. + */ @Override public Number getMetricValue() { return gauge.getValue(); diff --git a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMeter.java b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMeter.java index 181a00c4c0c99..68c61cfe9a1a6 100644 --- a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMeter.java +++ b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMeter.java @@ -23,10 +23,10 @@ import java.util.List; /** - * Mapping of meter between Flink and Datadog + * Mapping of meter between Flink and Datadog. * - * Only consider rate of the meter, due to Datadog HTTP API's limited support of meter - * */ + *

Only consider rate of the meter, due to Datadog HTTP API's limited support of meter + */ public class DMeter extends DMetric { private final Meter meter; diff --git a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMetric.java b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMetric.java index 3f9d6ffba592c..e55a9f0744dba 100644 --- a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMetric.java +++ b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMetric.java @@ -25,16 +25,16 @@ import java.util.List; /** - * Abstract metric of Datadog for serialization - * */ + * Abstract metric of Datadog for serialization. + */ @JsonInclude(JsonInclude.Include.NON_NULL) public abstract class DMetric { private static final long MILLIS_TO_SEC = 1000L; /** * Names of metric/type/tags field and their getters must not be changed - * since they are mapped to json objects in a Datadog-defined format - * */ + * since they are mapped to json objects in a Datadog-defined format. + */ private final String metric; // Metric name private final MetricType type; private final String host; diff --git a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DSeries.java b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DSeries.java index fb0bb09ec6851..54d907c927b29 100644 --- a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DSeries.java +++ b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DSeries.java @@ -22,13 +22,13 @@ import java.util.List; /** - * Json serialization between Flink and Datadog - **/ + * Json serialization between Flink and Datadog. + */ public class DSeries { /** * Names of series field and its getters must not be changed - * since they are mapped to json objects in a Datadog-defined format - * */ + * since they are mapped to json objects in a Datadog-defined format. + */ private List series; public DSeries() { diff --git a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpClient.java b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpClient.java index c94a2b7ea13ab..086966b4d1ff5 100644 --- a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpClient.java +++ b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpClient.java @@ -23,15 +23,15 @@ import okhttp3.MediaType; import okhttp3.OkHttpClient; import okhttp3.Request; -import okhttp3.Response; import okhttp3.RequestBody; +import okhttp3.Response; import java.io.IOException; import java.util.concurrent.TimeUnit; /** - * Http client talking to Datadog - * */ + * Http client talking to Datadog. + */ public class DatadogHttpClient{ private static final String SERIES_URL_FORMAT = "https://app.datadoghq.com/api/v1/series?api_key=%s"; private static final String VALIDATE_URL_FORMAT = "https://app.datadoghq.com/api/v1/validate?api_key=%s"; @@ -69,7 +69,7 @@ private void validateApiKey() { throw new IllegalArgumentException( String.format("API key: %s is invalid", apiKey)); } - } catch(IOException e) { + } catch (IOException e) { throw new IllegalStateException("Failed contacting Datadog to validate API key", e); } } diff --git a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java index fcb5c4b92b9c5..66190148c0b27 100644 --- a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java +++ b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java @@ -20,13 +20,14 @@ import org.apache.flink.metrics.Counter; import org.apache.flink.metrics.Gauge; -import org.apache.flink.metrics.Meter; import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.Meter; import org.apache.flink.metrics.Metric; import org.apache.flink.metrics.MetricConfig; import org.apache.flink.metrics.MetricGroup; import org.apache.flink.metrics.reporter.MetricReporter; import org.apache.flink.metrics.reporter.Scheduled; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,12 +37,11 @@ import java.util.Map; import java.util.concurrent.ConcurrentHashMap; - /** - * Metric Reporter for Datadog + * Metric Reporter for Datadog. * - * Variables in metrics scope will be sent to Datadog as tags - * */ + *

Variables in metrics scope will be sent to Datadog as tags. + */ public class DatadogHttpReporter implements MetricReporter, Scheduled { private static final Logger LOGGER = LoggerFactory.getLogger(DatadogHttpReporter.class); private static final String HOST_VARIABLE = ""; @@ -146,20 +146,20 @@ public void report() { } /** - * Get config tags from config 'metrics.reporter.dghttp.tags' - * */ + * Get config tags from config 'metrics.reporter.dghttp.tags'. + */ private List getTagsFromConfig(String str) { return Arrays.asList(str.split(",")); } /** - * Get tags from MetricGroup#getAllVariables(), excluding 'host' - * */ + * Get tags from MetricGroup#getAllVariables(), excluding 'host'. + */ private List getTagsFromMetricGroup(MetricGroup metricGroup) { List tags = new ArrayList<>(); for (Map.Entry entry: metricGroup.getAllVariables().entrySet()) { - if(!entry.getKey().equals(HOST_VARIABLE)) { + if (!entry.getKey().equals(HOST_VARIABLE)) { tags.add(getVariableName(entry.getKey()) + ":" + entry.getValue()); } } @@ -167,23 +167,17 @@ private List getTagsFromMetricGroup(MetricGroup metricGroup) { return tags; } - /** - * Get host from MetricGroup#getAllVariables() if it exists; returns Null otherwise - * */ private String getHostFromMetricGroup(MetricGroup metricGroup) { return metricGroup.getAllVariables().get(HOST_VARIABLE); } - /** - * Given "", return "xxx" - * */ private String getVariableName(String str) { return str.substring(1, str.length() - 1); } /** - * Compact metrics in batch, serialize them, and send to Datadog via HTTP - * */ + * Compact metrics in batch, serialize them, and send to Datadog via HTTP. + */ static class DatadogHttpRequest { private final DSeries series; diff --git a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/MetricType.java b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/MetricType.java index 97f9b29ede309..9681514807077 100644 --- a/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/MetricType.java +++ b/flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/MetricType.java @@ -19,12 +19,12 @@ package org.apache.flink.metrics.datadog; /** - * Metric types supported by Datadog - * */ + * Metric types supported by Datadog. + */ public enum MetricType { /** * Names of 'gauge' and 'counter' must not be changed - * since they are mapped to json objects in a Datadog-defined format - * */ + * since they are mapped to json objects in a Datadog-defined format. + */ gauge, counter } diff --git a/flink-metrics/flink-metrics-datadog/src/test/java/org/apache/flink/metrics/datadog/DatadogHttpClientTest.java b/flink-metrics/flink-metrics-datadog/src/test/java/org/apache/flink/metrics/datadog/DatadogHttpClientTest.java index bda5d4749bd7f..06802521e38b3 100644 --- a/flink-metrics/flink-metrics-datadog/src/test/java/org/apache/flink/metrics/datadog/DatadogHttpClientTest.java +++ b/flink-metrics/flink-metrics-datadog/src/test/java/org/apache/flink/metrics/datadog/DatadogHttpClientTest.java @@ -18,13 +18,13 @@ package org.apache.flink.metrics.datadog; -import com.fasterxml.jackson.core.JsonProcessingException; import org.apache.flink.metrics.Counter; import org.apache.flink.metrics.Gauge; import org.apache.flink.metrics.Meter; + +import com.fasterxml.jackson.core.JsonProcessingException; import org.junit.Before; import org.junit.Test; -import org.junit.experimental.runners.Enclosed; import org.junit.runner.RunWith; import org.powermock.api.mockito.PowerMockito; import org.powermock.core.classloader.annotations.PrepareForTest; @@ -35,165 +35,164 @@ import static org.junit.Assert.assertEquals; -@RunWith(Enclosed.class) +/** + * Tests for the DatadogHttpClient. + */ +@RunWith(PowerMockRunner.class) +@PrepareForTest(DMetric.class) public class DatadogHttpClientTest { - public static class TestApiKey { - @Test(expected = IllegalArgumentException.class) - public void testClientWithEmptyKey() { - new DatadogHttpClient(""); - } - - @Test(expected = IllegalArgumentException.class) - public void testClientWithNullKey() { - new DatadogHttpClient(null); - } + + private static List tags = Arrays.asList("tag1", "tag2"); + + private static final long MOCKED_SYSTEM_MILLIS = 123L; + + @Before + public void mockSystemMillis() { + PowerMockito.mockStatic(DMetric.class); + PowerMockito.when(DMetric.getUnixEpochTimestamp()).thenReturn(MOCKED_SYSTEM_MILLIS); + } + + @Test(expected = IllegalArgumentException.class) + public void testClientWithEmptyKey() { + new DatadogHttpClient(""); } - @RunWith(PowerMockRunner.class) - @PrepareForTest(DMetric.class) - public static class TestSerialization { - private static List tags = Arrays.asList("tag1", "tag2"); - - private static final long MOCKED_SYSTEM_MILLIS = 123L; - - @Before - public void mockSystemMillis() { - PowerMockito.mockStatic(DMetric.class); - PowerMockito.when(DMetric.getUnixEpochTimestamp()).thenReturn(MOCKED_SYSTEM_MILLIS); - } - - @Test - public void serializeGauge() throws JsonProcessingException { - - DGauge g = new DGauge(new Gauge() { - @Override - public Number getValue() { - return 1; - } - }, "testCounter", "localhost", tags); - - assertEquals( - "{\"metric\":\"testCounter\",\"type\":\"gauge\",\"host\":\"localhost\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1]]}", - DatadogHttpClient.serialize(g)); - } - - @Test - public void serializeGaugeWithoutHost() throws JsonProcessingException { - - DGauge g = new DGauge(new Gauge() { - @Override - public Number getValue() { - return 1; - } - }, "testCounter", null, tags); - - assertEquals( - "{\"metric\":\"testCounter\",\"type\":\"gauge\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1]]}", - DatadogHttpClient.serialize(g)); - } - - @Test - public void serializeCounter() throws JsonProcessingException { - DCounter c = new DCounter(new Counter() { - @Override - public void inc() {} - - @Override - public void inc(long n) {} - - @Override - public void dec() {} - - @Override - public void dec(long n) {} - - @Override - public long getCount() { - return 1; - } - }, "testCounter", "localhost", tags); - - assertEquals( - "{\"metric\":\"testCounter\",\"type\":\"counter\",\"host\":\"localhost\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1]]}", - DatadogHttpClient.serialize(c)); - } - - @Test - public void serializeCounterWithoutHost() throws JsonProcessingException { - DCounter c = new DCounter(new Counter() { - @Override - public void inc() {} - - @Override - public void inc(long n) {} - - @Override - public void dec() {} - - @Override - public void dec(long n) {} - - @Override - public long getCount() { - return 1; - } - }, "testCounter", null, tags); - - assertEquals( - "{\"metric\":\"testCounter\",\"type\":\"counter\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1]]}", - DatadogHttpClient.serialize(c)); - } - - @Test - public void serializeMeter() throws JsonProcessingException { - - DMeter m = new DMeter(new Meter() { - @Override - public void markEvent() {} - - @Override - public void markEvent(long n) {} - - @Override - public double getRate() { - return 1; - } - - @Override - public long getCount() { - return 0; - } - }, "testMeter","localhost", tags); - - assertEquals( - "{\"metric\":\"testMeter\",\"type\":\"gauge\",\"host\":\"localhost\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1.0]]}", - DatadogHttpClient.serialize(m)); - } - - @Test - public void serializeMeterWithoutHost() throws JsonProcessingException { - - DMeter m = new DMeter(new Meter() { - @Override - public void markEvent() {} - - @Override - public void markEvent(long n) {} - - @Override - public double getRate() { - return 1; - } - - @Override - public long getCount() { - return 0; - } - }, "testMeter", null, tags); - - assertEquals( - "{\"metric\":\"testMeter\",\"type\":\"gauge\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1.0]]}", - DatadogHttpClient.serialize(m)); - } + @Test(expected = IllegalArgumentException.class) + public void testClientWithNullKey() { + new DatadogHttpClient(null); + } + + @Test + public void serializeGauge() throws JsonProcessingException { + + DGauge g = new DGauge(new Gauge() { + @Override + public Number getValue() { + return 1; + } + }, "testCounter", "localhost", tags); + + assertEquals( + "{\"metric\":\"testCounter\",\"type\":\"gauge\",\"host\":\"localhost\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1]]}", + DatadogHttpClient.serialize(g)); + } + + @Test + public void serializeGaugeWithoutHost() throws JsonProcessingException { + + DGauge g = new DGauge(new Gauge() { + @Override + public Number getValue() { + return 1; + } + }, "testCounter", null, tags); + + assertEquals( + "{\"metric\":\"testCounter\",\"type\":\"gauge\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1]]}", + DatadogHttpClient.serialize(g)); + } + + @Test + public void serializeCounter() throws JsonProcessingException { + DCounter c = new DCounter(new Counter() { + @Override + public void inc() {} + + @Override + public void inc(long n) {} + + @Override + public void dec() {} + + @Override + public void dec(long n) {} + + @Override + public long getCount() { + return 1; + } + }, "testCounter", "localhost", tags); + + assertEquals( + "{\"metric\":\"testCounter\",\"type\":\"counter\",\"host\":\"localhost\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1]]}", + DatadogHttpClient.serialize(c)); + } + + @Test + public void serializeCounterWithoutHost() throws JsonProcessingException { + DCounter c = new DCounter(new Counter() { + @Override + public void inc() {} + + @Override + public void inc(long n) {} + + @Override + public void dec() {} + + @Override + public void dec(long n) {} + + @Override + public long getCount() { + return 1; + } + }, "testCounter", null, tags); + + assertEquals( + "{\"metric\":\"testCounter\",\"type\":\"counter\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1]]}", + DatadogHttpClient.serialize(c)); + } + + @Test + public void serializeMeter() throws JsonProcessingException { + + DMeter m = new DMeter(new Meter() { + @Override + public void markEvent() {} + + @Override + public void markEvent(long n) {} + + @Override + public double getRate() { + return 1; + } + + @Override + public long getCount() { + return 0; + } + }, "testMeter", "localhost", tags); + + assertEquals( + "{\"metric\":\"testMeter\",\"type\":\"gauge\",\"host\":\"localhost\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1.0]]}", + DatadogHttpClient.serialize(m)); + } + + @Test + public void serializeMeterWithoutHost() throws JsonProcessingException { + + DMeter m = new DMeter(new Meter() { + @Override + public void markEvent() {} + + @Override + public void markEvent(long n) {} + + @Override + public double getRate() { + return 1; + } + + @Override + public long getCount() { + return 0; + } + }, "testMeter", null, tags); + + assertEquals( + "{\"metric\":\"testMeter\",\"type\":\"gauge\",\"tags\":[\"tag1\",\"tag2\"],\"points\":[[123,1.0]]}", + DatadogHttpClient.serialize(m)); } } diff --git a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/ScheduledDropwizardReporter.java b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/ScheduledDropwizardReporter.java index 380abc4582ef9..4d4a18fcf5336 100644 --- a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/ScheduledDropwizardReporter.java +++ b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/ScheduledDropwizardReporter.java @@ -18,9 +18,6 @@ package org.apache.flink.dropwizard; -import com.codahale.metrics.MetricRegistry; -import com.codahale.metrics.Reporter; -import com.codahale.metrics.ScheduledReporter; import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper; @@ -39,6 +36,11 @@ import org.apache.flink.metrics.MetricGroup; import org.apache.flink.metrics.reporter.MetricReporter; import org.apache.flink.metrics.reporter.Scheduled; + +import com.codahale.metrics.MetricRegistry; +import com.codahale.metrics.Reporter; +import com.codahale.metrics.ScheduledReporter; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -161,7 +163,7 @@ else if (metric instanceof Gauge) { public void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup group) { synchronized (this) { String fullName; - + if (metric instanceof Counter) { fullName = counters.remove(metric); } else if (metric instanceof Gauge) { @@ -173,7 +175,7 @@ public void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup } else { fullName = null; } - + if (fullName != null) { registry.remove(fullName); } diff --git a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/DropwizardHistogramStatistics.java b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/DropwizardHistogramStatistics.java index 6f4eab2c33893..8acf401143ed8 100644 --- a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/DropwizardHistogramStatistics.java +++ b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/DropwizardHistogramStatistics.java @@ -18,9 +18,10 @@ package org.apache.flink.dropwizard.metrics; -import com.codahale.metrics.Snapshot; import org.apache.flink.metrics.HistogramStatistics; +import com.codahale.metrics.Snapshot; + /** * Dropwizard histogram statistics implementation returned by {@link DropwizardHistogramWrapper}. * The statistics class wraps a {@link Snapshot} instance and forwards the method calls accordingly. diff --git a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkCounterWrapper.java b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkCounterWrapper.java index a44c3f5c1012e..b4ea8dc01c93d 100644 --- a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkCounterWrapper.java +++ b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkCounterWrapper.java @@ -15,10 +15,14 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.flink.dropwizard.metrics; import org.apache.flink.metrics.Counter; +/** + * A wrapper that allows a Flink counter to be used as a DropWizard counter. + */ public class FlinkCounterWrapper extends com.codahale.metrics.Counter { private final Counter counter; diff --git a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkGaugeWrapper.java b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkGaugeWrapper.java index 058ecaddc94bb..0d53a9eb97c70 100644 --- a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkGaugeWrapper.java +++ b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkGaugeWrapper.java @@ -20,8 +20,11 @@ import org.apache.flink.metrics.Gauge; +/** + * A wrapper that allows a Flink gauge to be used as a DropWizard gauge. + */ public class FlinkGaugeWrapper implements com.codahale.metrics.Gauge { - + private final Gauge gauge; public FlinkGaugeWrapper(Gauge gauge) { diff --git a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkHistogramWrapper.java b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkHistogramWrapper.java index 8bd8078bf5150..d2167d082ff4f 100644 --- a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkHistogramWrapper.java +++ b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkHistogramWrapper.java @@ -18,9 +18,10 @@ package org.apache.flink.dropwizard.metrics; -import com.codahale.metrics.Snapshot; import org.apache.flink.metrics.Histogram; +import com.codahale.metrics.Snapshot; + /** * Wrapper to use a Flink {@link Histogram} as a Dropwizard {@link com.codahale.metrics.Histogram}. * This is necessary to report Flink's histograms via the Dropwizard diff --git a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkMeterWrapper.java b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkMeterWrapper.java index d0b84838f22b4..213c21c5de951 100644 --- a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkMeterWrapper.java +++ b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/FlinkMeterWrapper.java @@ -18,9 +18,10 @@ package org.apache.flink.dropwizard.metrics; -import com.codahale.metrics.Clock; import org.apache.flink.metrics.Meter; +import com.codahale.metrics.Clock; + /** * Wrapper to use a Flink {@link Meter} as a Dropwizard {@link com.codahale.metrics.Meter}. * This is necessary to report Flink's meters via the Dropwizard diff --git a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/HistogramStatisticsWrapper.java b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/HistogramStatisticsWrapper.java index 6d3a69b507692..7b1ff74b17118 100644 --- a/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/HistogramStatisticsWrapper.java +++ b/flink-metrics/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/HistogramStatisticsWrapper.java @@ -18,9 +18,10 @@ package org.apache.flink.dropwizard.metrics; -import com.codahale.metrics.Snapshot; import org.apache.flink.metrics.HistogramStatistics; +import com.codahale.metrics.Snapshot; + import java.io.OutputStream; import java.io.OutputStreamWriter; import java.io.PrintWriter; @@ -39,6 +40,7 @@ class HistogramStatisticsWrapper extends Snapshot { HistogramStatisticsWrapper(HistogramStatistics histogramStatistics) { this.histogramStatistics = histogramStatistics; } + @Override public double getValue(double quantile) { return histogramStatistics.getQuantile(quantile); @@ -76,7 +78,7 @@ public double getStdDev() { @Override public void dump(OutputStream output) { - try(PrintWriter printWriter = new PrintWriter(new OutputStreamWriter(output, UTF_8))){ + try (PrintWriter printWriter = new PrintWriter(new OutputStreamWriter(output, UTF_8))) { for (Long value : histogramStatistics.getValues()) { printWriter.printf("%d%n", value); diff --git a/flink-metrics/flink-metrics-ganglia/src/main/java/org/apache/flink/metrics/ganglia/GangliaReporter.java b/flink-metrics/flink-metrics-ganglia/src/main/java/org/apache/flink/metrics/ganglia/GangliaReporter.java index de9da74a2068f..87199011fc563 100644 --- a/flink-metrics/flink-metrics-ganglia/src/main/java/org/apache/flink/metrics/ganglia/GangliaReporter.java +++ b/flink-metrics/flink-metrics-ganglia/src/main/java/org/apache/flink/metrics/ganglia/GangliaReporter.java @@ -18,20 +18,23 @@ package org.apache.flink.metrics.ganglia; -import com.codahale.metrics.ScheduledReporter; - -import info.ganglia.gmetric4j.gmetric.GMetric; - import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.dropwizard.ScheduledDropwizardReporter; import org.apache.flink.metrics.MetricConfig; +import com.codahale.metrics.ScheduledReporter; +import info.ganglia.gmetric4j.gmetric.GMetric; + import java.io.IOException; import java.util.concurrent.TimeUnit; +/** + * This class acts as a factory for the {@link com.codahale.metrics.ganglia.GangliaReporter} and allows using it as a + * Flink reporter. + */ @PublicEvolving public class GangliaReporter extends ScheduledDropwizardReporter { - + public static final String ARG_DMAX = "dmax"; public static final String ARG_TMAX = "tmax"; public static final String ARG_TTL = "ttl"; @@ -72,7 +75,7 @@ public ScheduledReporter getReporter(MetricConfig config) { builder.withTMax(tMax); log.info("Configured GangliaReporter with {host:{}, port:{}, dmax:{}, tmax:{}, ttl:{}, addressingMode:{}}", - host, port, dMax, tMax, ttl, addressingMode); + host, port, dMax, tMax, ttl, addressingMode); return builder.build(gMetric); } catch (IOException e) { throw new RuntimeException("Error while instantiating GangliaReporter.", e); diff --git a/flink-metrics/flink-metrics-graphite/src/main/java/org/apache/flink/metrics/graphite/GraphiteReporter.java b/flink-metrics/flink-metrics-graphite/src/main/java/org/apache/flink/metrics/graphite/GraphiteReporter.java index 47a9d872e1f24..3338af2985fdd 100644 --- a/flink-metrics/flink-metrics-graphite/src/main/java/org/apache/flink/metrics/graphite/GraphiteReporter.java +++ b/flink-metrics/flink-metrics-graphite/src/main/java/org/apache/flink/metrics/graphite/GraphiteReporter.java @@ -18,16 +18,20 @@ package org.apache.flink.metrics.graphite; -import com.codahale.metrics.ScheduledReporter; -import com.codahale.metrics.graphite.Graphite; - -import com.codahale.metrics.graphite.GraphiteUDP; import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.dropwizard.ScheduledDropwizardReporter; import org.apache.flink.metrics.MetricConfig; +import com.codahale.metrics.ScheduledReporter; +import com.codahale.metrics.graphite.Graphite; +import com.codahale.metrics.graphite.GraphiteUDP; + import java.util.concurrent.TimeUnit; +/** + * This class acts as a factory for the {@link com.codahale.metrics.graphite.GraphiteReporter} and allows using it as a + * Flink reporter. + */ @PublicEvolving public class GraphiteReporter extends ScheduledDropwizardReporter { @@ -78,7 +82,7 @@ public ScheduledReporter getReporter(MetricConfig config) { log.info("Configured GraphiteReporter with {host:{}, port:{}, protocol:{}}", host, port, prot); switch(prot) { case UDP: - return builder.build(new GraphiteUDP(host, port)); + return builder.build(new GraphiteUDP(host, port)); case TCP: default: return builder.build(new Graphite(host, port)); diff --git a/flink-metrics/flink-metrics-jmx/src/main/java/org/apache/flink/metrics/jmx/JMXReporter.java b/flink-metrics/flink-metrics-jmx/src/main/java/org/apache/flink/metrics/jmx/JMXReporter.java index f0c0fcbf68243..1cc7d38685eeb 100644 --- a/flink-metrics/flink-metrics-jmx/src/main/java/org/apache/flink/metrics/jmx/JMXReporter.java +++ b/flink-metrics/flink-metrics-jmx/src/main/java/org/apache/flink/metrics/jmx/JMXReporter.java @@ -30,6 +30,7 @@ import org.apache.flink.runtime.metrics.groups.AbstractMetricGroup; import org.apache.flink.runtime.metrics.groups.FrontMetricGroup; import org.apache.flink.util.NetUtils; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -42,6 +43,7 @@ import javax.management.remote.JMXConnectorServer; import javax.management.remote.JMXConnectorServerFactory; import javax.management.remote.JMXServiceURL; + import java.io.IOException; import java.lang.management.ManagementFactory; import java.net.MalformedURLException; @@ -57,7 +59,7 @@ /** * {@link MetricReporter} that exports {@link Metric Metrics} via JMX. * - * Largely based on the JmxReporter class of the dropwizard metrics library + *

Largely based on the JmxReporter class of the dropwizard metrics library * https://github.com/dropwizard/metrics/blob/master/metrics-core/src/main/java/io/dropwizard/metrics/JmxReporter.java */ public class JMXReporter implements MetricReporter { @@ -77,18 +79,15 @@ public String filterCharacters(String input) { // ------------------------------------------------------------------------ - /** The server where the management beans are registered and deregistered */ + /** The server where the management beans are registered and deregistered. */ private final MBeanServer mBeanServer; - /** The names under which the registered metrics have been added to the MBeanServer */ + /** The names under which the registered metrics have been added to the MBeanServer. */ private final Map registeredMetrics; /** The server to which JMX clients connect to. ALlows for better control over port usage. */ private JMXServer jmxServer; - /** - * Creates a new JMXReporter - */ public JMXReporter() { this.mBeanServer = ManagementFactory.getPlatformMBeanServer(); this.registeredMetrics = new HashMap<>(); @@ -140,7 +139,7 @@ public void close() { } } } - + public int getPort() { if (jmxServer == null) { throw new NullPointerException("No server was opened. Did you specify a port?"); @@ -220,7 +219,7 @@ public void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup } // ------------------------------------------------------------------------ - // Utilities + // Utilities // ------------------------------------------------------------------------ static Hashtable generateJmxTable(Map variables) { @@ -239,9 +238,9 @@ static String generateJmxDomain(String metricName, MetricGroup group) { * Lightweight method to replace unsupported characters. * If the string does not contain any unsupported characters, this method creates no * new string (and in fact no new objects at all). - * + * *

Replacements: - * + * *