You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2016/10/31 12:16:11 UTC
[1/2] flink git commit: [FLINK-4245] JMXReporter exposes all defined
variables
Repository: flink
Updated Branches:
refs/heads/master 344fe94db -> 9f4be44c2
[FLINK-4245] JMXReporter exposes all defined variables
This closes #2418.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/94d1b63c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/94d1b63c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/94d1b63c
Branch: refs/heads/master
Commit: 94d1b63c05ded616120e8283f584a7affe7fe744
Parents: 344fe94
Author: zentol <ch...@apache.org>
Authored: Wed Jul 27 12:29:10 2016 +0200
Committer: zentol <ch...@apache.org>
Committed: Mon Oct 31 13:15:04 2016 +0100
----------------------------------------------------------------------
flink-metrics/flink-metrics-jmx/pom.xml | 10 +--
.../apache/flink/metrics/jmx/JMXReporter.java | 62 +++++++++---------
.../flink/metrics/jmx/JMXReporterTest.java | 66 +++++++++++++-------
.../jobmanager/JMXJobManagerMetricTest.java | 7 ++-
.../metrics/groups/AbstractMetricGroup.java | 41 ++++++++++++
.../metrics/groups/FrontMetricGroup.java | 8 +++
.../metrics/groups/GenericMetricGroup.java | 5 ++
.../metrics/groups/JobManagerMetricGroup.java | 5 ++
.../runtime/metrics/groups/JobMetricGroup.java | 5 ++
.../metrics/groups/OperatorMetricGroup.java | 5 ++
.../metrics/groups/TaskManagerMetricGroup.java | 5 ++
.../runtime/metrics/groups/TaskMetricGroup.java | 5 ++
.../metrics/groups/AbstractMetricGroupTest.java | 5 ++
.../runtime/metrics/groups/MetricGroupTest.java | 5 ++
.../connectors/kafka/KafkaConsumerTestBase.java | 6 +-
15 files changed, 181 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-metrics/flink-metrics-jmx/pom.xml
----------------------------------------------------------------------
diff --git a/flink-metrics/flink-metrics-jmx/pom.xml b/flink-metrics/flink-metrics-jmx/pom.xml
index 626f063..8ee983b 100644
--- a/flink-metrics/flink-metrics-jmx/pom.xml
+++ b/flink-metrics/flink-metrics-jmx/pom.xml
@@ -49,20 +49,20 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-metrics-core</artifactId>
+ <artifactId>flink-runtime_2.10</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
- <!-- test dependencies -->
-
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime_2.10</artifactId>
+ <artifactId>flink-metrics-core</artifactId>
<version>${project.version}</version>
- <scope>test</scope>
+ <scope>provided</scope>
</dependency>
+ <!-- test dependencies -->
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime_2.10</artifactId>
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-metrics/flink-metrics-jmx/src/main/java/org/apache/flink/metrics/jmx/JMXReporter.java
----------------------------------------------------------------------
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 39a5aa2..9c1fabb 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
@@ -18,6 +18,7 @@
package org.apache.flink.metrics.jmx;
+import org.apache.flink.metrics.CharacterFilter;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Histogram;
@@ -25,8 +26,9 @@ 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.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;
@@ -48,6 +50,7 @@ import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.HashMap;
+import java.util.Hashtable;
import java.util.Iterator;
import java.util.Map;
@@ -59,13 +62,19 @@ import java.util.Map;
*/
public class JMXReporter implements MetricReporter {
- private static final String PREFIX = "org.apache.flink.metrics:";
- private static final String KEY_PREFIX = "key";
+ static final String JMX_DOMAIN_PREFIX = "org.apache.flink.";
public static final String ARG_PORT = "port";
private static final Logger LOG = LoggerFactory.getLogger(JMXReporter.class);
+ private static final CharacterFilter CHARACTER_FILTER = new CharacterFilter() {
+ @Override
+ public String filterCharacters(String input) {
+ return replaceInvalidChars(input);
+ }
+ };
+
// ------------------------------------------------------------------------
/** The server where the management beans are registered and deregistered */
@@ -144,14 +153,19 @@ public class JMXReporter implements MetricReporter {
@Override
public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
- final String name = generateJmxName(metricName, group.getScopeComponents());
+ final String domain = generateJmxDomain(metricName, group);
+ final Hashtable<String, String> table = generateJmxTable(group.getAllVariables());
AbstractBean jmxMetric;
ObjectName jmxName;
try {
- jmxName = new ObjectName(name);
+ jmxName = new ObjectName(domain, table);
} catch (MalformedObjectNameException e) {
- LOG.error("Metric name did not conform to JMX ObjectName rules: " + name, e);
+ /**
+ * There is an implementation error on our side if this occurs. Either the domain was modified and no longer
+ * conforms to the JMX domain rules or the table wasn't properly generated.
+ */
+ LOG.debug("Implementation error. The domain or table does not conform to JMX rules." , e);
return;
}
@@ -176,12 +190,11 @@ public class JMXReporter implements MetricReporter {
}
} catch (NotCompliantMBeanException e) {
// implementation error on our side
- LOG.error("Metric did not comply with JMX MBean naming rules.", e);
+ LOG.debug("Metric did not comply with JMX MBean rules.", e);
} catch (InstanceAlreadyExistsException e) {
- LOG.debug("A metric with the name " + jmxName + " was already registered.", e);
- LOG.error("A metric with the name " + jmxName + " was already registered.");
+ LOG.warn("A metric with the name " + jmxName + " was already registered.", e);
} catch (Throwable t) {
- LOG.error("Failed to register metric", t);
+ LOG.warn("Failed to register metric", t);
}
}
@@ -209,27 +222,18 @@ public class JMXReporter implements MetricReporter {
// Utilities
// ------------------------------------------------------------------------
- static String generateJmxName(String metricName, String[] scopeComponents) {
- final StringBuilder nameBuilder = new StringBuilder(128);
- nameBuilder.append(PREFIX);
-
- for (int x = 0; x < scopeComponents.length; x++) {
- // write keyX=
- nameBuilder.append(KEY_PREFIX);
- nameBuilder.append(x);
- nameBuilder.append("=");
-
- // write scope component
- nameBuilder.append(replaceInvalidChars(scopeComponents[x]));
- nameBuilder.append(",");
+ static Hashtable<String, String> generateJmxTable(Map<String, String> variables) {
+ Hashtable<String, String> ht = new Hashtable<>(variables.size());
+ for (Map.Entry<String, String> variable : variables.entrySet()) {
+ ht.put(replaceInvalidChars(variable.getKey()), replaceInvalidChars(variable.getValue()));
}
+ return ht;
+ }
- // write the name
- nameBuilder.append("name=").append(replaceInvalidChars(metricName));
-
- return nameBuilder.toString();
+ static String generateJmxDomain(String metricName, MetricGroup group) {
+ return JMX_DOMAIN_PREFIX + ((FrontMetricGroup<AbstractMetricGroup<?>>) group).getLogicalScope(CHARACTER_FILTER, '.') + '.' + metricName;
}
-
+
/**
* Lightweight method to replace unsupported characters.
* If the string does not contain any unsupported characters, this method creates no
@@ -251,6 +255,8 @@ public class JMXReporter implements MetricReporter {
for (int i = 0; i < strLen; i++) {
final char c = str.charAt(i);
switch (c) {
+ case '>':
+ case '<':
case '"':
// remove character by not moving cursor
if (chars == null) {
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/metrics/jmx/JMXReporterTest.java
----------------------------------------------------------------------
diff --git a/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/metrics/jmx/JMXReporterTest.java b/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/metrics/jmx/JMXReporterTest.java
index 8968587..1a96287 100644
--- a/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/metrics/jmx/JMXReporterTest.java
+++ b/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/metrics/jmx/JMXReporterTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.metrics.util.TestMeter;
import org.apache.flink.metrics.reporter.MetricReporter;
import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.metrics.MetricRegistryConfiguration;
+import org.apache.flink.runtime.metrics.groups.FrontMetricGroup;
import org.apache.flink.runtime.metrics.groups.TaskManagerMetricGroup;
import org.apache.flink.runtime.metrics.util.TestReporter;
import org.apache.flink.runtime.metrics.util.TestingHistogram;
@@ -40,8 +41,12 @@ import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import java.lang.management.ManagementFactory;
+import java.util.HashMap;
+import java.util.Hashtable;
import java.util.List;
+import java.util.Map;
+import static org.apache.flink.metrics.jmx.JMXReporter.JMX_DOMAIN_PREFIX;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -65,14 +70,20 @@ public class JMXReporterTest extends TestLogger {
}
/**
- * Verifies that the JMXReporter properly generates the JMX name.
+ * Verifies that the JMXReporter properly generates the JMX table.
*/
@Test
- public void testGenerateName() {
- String[] scope = {"value0", "value1", "\"value2 (test),=;:?'"};
- String jmxName = JMXReporter.generateJmxName("TestMetric", scope);
+ public void testGenerateTable() {
+ Map<String, String> vars = new HashMap<>();
+ vars.put("key0", "value0");
+ vars.put("key1", "value1");
+ vars.put("\"key2,=;:?'", "\"value2 (test),=;:?'");
- assertEquals("org.apache.flink.metrics:key0=value0,key1=value1,key2=value2_(test)------,name=TestMetric", jmxName);
+ Hashtable<String, String> jmxTable = JMXReporter.generateJmxTable(vars);
+
+ assertEquals("value0", jmxTable.get("key0"));
+ assertEquals("value1", jmxTable.get("key1"));
+ assertEquals("value2_(test)------", jmxTable.get("key2------"));
}
/**
@@ -102,28 +113,34 @@ public class JMXReporterTest extends TestLogger {
MetricReporter rep1 = reporters.get(0);
MetricReporter rep2 = reporters.get(1);
- rep1.notifyOfAddedMetric(new Gauge<Integer>() {
+ Gauge<Integer> g1 = new Gauge<Integer>() {
@Override
public Integer getValue() {
return 1;
}
- }, "rep1", new TaskManagerMetricGroup(reg, "host", "tm"));
-
- rep2.notifyOfAddedMetric(new Gauge<Integer>() {
+ };
+ Gauge<Integer> g2 = new Gauge<Integer>() {
@Override
public Integer getValue() {
return 2;
}
- }, "rep2", new TaskManagerMetricGroup(reg, "host", "tm"));
+ };
+
+ rep1.notifyOfAddedMetric(g1, "rep1", new FrontMetricGroup<>(0, new TaskManagerMetricGroup(reg, "host", "tm")));
+ rep2.notifyOfAddedMetric(g2, "rep2", new FrontMetricGroup<>(0, new TaskManagerMetricGroup(reg, "host", "tm")));
MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
- ObjectName objectName1 = new ObjectName(JMXReporter.generateJmxName("rep1", mg.getScopeComponents()));
- ObjectName objectName2 = new ObjectName(JMXReporter.generateJmxName("rep2", mg.getScopeComponents()));
+ ObjectName objectName1 = new ObjectName(JMX_DOMAIN_PREFIX + "taskmanager.rep1", JMXReporter.generateJmxTable(mg.getAllVariables()));
+ ObjectName objectName2 = new ObjectName(JMX_DOMAIN_PREFIX + "taskmanager.rep2", JMXReporter.generateJmxTable(mg.getAllVariables()));
assertEquals(1, mBeanServer.getAttribute(objectName1, "Value"));
assertEquals(2, mBeanServer.getAttribute(objectName2, "Value"));
+ rep1.notifyOfRemovedMetric(g1, "rep1", null);
+ rep1.notifyOfRemovedMetric(g2, "rep2", null);
+
+ mg.close();
reg.shutdown();
}
@@ -156,22 +173,25 @@ public class JMXReporterTest extends TestLogger {
MetricReporter rep1 = reporters.get(0);
MetricReporter rep2 = reporters.get(1);
- rep1.notifyOfAddedMetric(new Gauge<Integer>() {
+ Gauge<Integer> g1 = new Gauge<Integer>() {
@Override
public Integer getValue() {
return 1;
}
- }, "rep1", new TaskManagerMetricGroup(reg, "host", "tm"));
-
- rep2.notifyOfAddedMetric(new Gauge<Integer>() {
+ };
+ Gauge<Integer> g2 = new Gauge<Integer>() {
@Override
public Integer getValue() {
return 2;
}
- }, "rep2", new TaskManagerMetricGroup(reg, "host", "tm"));
+ };
- ObjectName objectName1 = new ObjectName(JMXReporter.generateJmxName("rep1", mg.getScopeComponents()));
- ObjectName objectName2 = new ObjectName(JMXReporter.generateJmxName("rep2", mg.getScopeComponents()));
+ rep1.notifyOfAddedMetric(g1, "rep1", new FrontMetricGroup<>(0, new TaskManagerMetricGroup(reg, "host", "tm")));
+
+ rep2.notifyOfAddedMetric(g2, "rep2", new FrontMetricGroup<>(1, new TaskManagerMetricGroup(reg, "host", "tm")));
+
+ ObjectName objectName1 = new ObjectName(JMX_DOMAIN_PREFIX + "taskmanager.rep1", JMXReporter.generateJmxTable(mg.getAllVariables()));
+ ObjectName objectName2 = new ObjectName(JMX_DOMAIN_PREFIX + "taskmanager.rep2", JMXReporter.generateJmxTable(mg.getAllVariables()));
JMXServiceURL url1 = new JMXServiceURL("service:jmx:rmi://localhost:" + ((JMXReporter)rep1).getPort() + "/jndi/rmi://localhost:" + ((JMXReporter)rep1).getPort() + "/jmxrmi");
JMXConnector jmxCon1 = JMXConnectorFactory.connect(url1);
@@ -189,10 +209,14 @@ public class JMXReporterTest extends TestLogger {
assertEquals(1, mCon2.getAttribute(objectName1, "Value"));
assertEquals(2, mCon2.getAttribute(objectName2, "Value"));
+ rep1.notifyOfRemovedMetric(g1, "rep1", null);
+ rep1.notifyOfRemovedMetric(g2, "rep2", null);
+
jmxCon2.close();
rep1.close();
rep2.close();
+ mg.close();
reg.shutdown();
}
@@ -219,7 +243,7 @@ public class JMXReporterTest extends TestLogger {
MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
- ObjectName objectName = new ObjectName(JMXReporter.generateJmxName(histogramName, metricGroup.getScopeComponents()));
+ ObjectName objectName = new ObjectName(JMX_DOMAIN_PREFIX + "taskmanager." + histogramName, JMXReporter.generateJmxTable(metricGroup.getAllVariables()));
MBeanInfo info = mBeanServer.getMBeanInfo(objectName);
@@ -269,7 +293,7 @@ public class JMXReporterTest extends TestLogger {
MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
- ObjectName objectName = new ObjectName(JMXReporter.generateJmxName(meterName, metricGroup.getScopeComponents()));
+ ObjectName objectName = new ObjectName(JMX_DOMAIN_PREFIX + "taskmanager." + meterName, JMXReporter.generateJmxTable(metricGroup.getAllVariables()));
MBeanInfo info = mBeanServer.getMBeanInfo(objectName);
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/runtime/jobmanager/JMXJobManagerMetricTest.java
----------------------------------------------------------------------
diff --git a/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/runtime/jobmanager/JMXJobManagerMetricTest.java b/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/runtime/jobmanager/JMXJobManagerMetricTest.java
index 0954241..3ae224f 100644
--- a/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/runtime/jobmanager/JMXJobManagerMetricTest.java
+++ b/flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/runtime/jobmanager/JMXJobManagerMetricTest.java
@@ -28,6 +28,7 @@ import org.apache.flink.runtime.jobgraph.tasks.ExternalizedCheckpointSettings;
import org.apache.flink.runtime.jobgraph.tasks.JobSnapshottingSettings;
import org.apache.flink.runtime.testingUtils.TestingCluster;
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages;
+import org.junit.Assert;
import org.junit.Test;
import scala.concurrent.Await;
import scala.concurrent.Future;
@@ -38,6 +39,7 @@ import javax.management.MBeanServer;
import javax.management.ObjectName;
import java.lang.management.ManagementFactory;
import java.util.Collections;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
@@ -83,8 +85,9 @@ public class JMXJobManagerMetricTest {
Await.ready(jobRunning, deadline.timeLeft());
MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
- ObjectName objectName1 = new ObjectName("org.apache.flink.metrics:key0=jobmanager,key1=TestingJob,name=lastCheckpointSize");
- assertEquals(-1L, mBeanServer.getAttribute(objectName1, "Value"));
+ Set<ObjectName> nameSet = mBeanServer.queryNames(new ObjectName("org.apache.flink.jobmanager.job.lastCheckpointSize:job_name=TestingJob,*"), null);
+ Assert.assertEquals(1, nameSet.size());
+ assertEquals(-1L, mBeanServer.getAttribute(nameSet.iterator().next(), "Value"));
Future<Object> jobFinished = flink.getLeaderGateway(deadline.timeLeft())
.ask(new TestingJobManagerMessages.NotifyWhenJobRemoved(jobGraph.getJobID()), deadline.timeLeft());
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
index 907c655..04b8158 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
@@ -89,6 +89,10 @@ public abstract class AbstractMetricGroup<A extends AbstractMetricGroup<?>> impl
* For example: "host-7.taskmanager-2.window_word_count.my-mapper" */
private String scopeString;
+ /** The logical metrics scope represented by this group, as a concatenated string, lazily computed.
+ * For example: "taskmanager.job.task" */
+ private String logicalScopeString;
+
/** The metrics query service scope represented by this group, lazily computed. */
protected QueryScopeInfo queryServiceScopeInfo;
@@ -119,6 +123,43 @@ public abstract class AbstractMetricGroup<A extends AbstractMetricGroup<?>> impl
}
/**
+ * Returns the logical scope of this group, for example
+ * {@code "taskmanager.job.task"}
+ *
+ * @param filter character filter which is applied to the scope components
+ * @return logical scope
+ */
+ public String getLogicalScope(CharacterFilter filter) {
+ return getLogicalScope(filter, registry.getDelimiter());
+ }
+
+ /**
+ * Returns the logical scope of this group, for example
+ * {@code "taskmanager.job.task"}
+ *
+ * @param filter character filter which is applied to the scope components
+ * @return logical scope
+ */
+ public String getLogicalScope(CharacterFilter filter, char delimiter) {
+ if (logicalScopeString == null) {
+ if (parent == null) {
+ logicalScopeString = getGroupName(filter);
+ } else {
+ logicalScopeString = parent.getLogicalScope(filter, delimiter) + delimiter + getGroupName(filter);
+ }
+ }
+ return logicalScopeString;
+ }
+
+ /**
+ * Returns the name for this group, meaning what kind of entity it represents, for example "taskmanager".
+ *
+ * @param filter character filter which is applied to the name
+ * @return logical name for this group
+ */
+ protected abstract String getGroupName(CharacterFilter filter);
+
+ /**
* Gets the scope as an array of the scope components, for example
* {@code ["host-7", "taskmanager-2", "window_word_count", "my-mapper"]}
*
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroup.java
index 885e6d6..63842fe 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroup.java
@@ -46,4 +46,12 @@ public class FrontMetricGroup<P extends AbstractMetricGroup<?>> extends ProxyMet
public String getMetricIdentifier(String metricName, CharacterFilter filter) {
return parentMetricGroup.getMetricIdentifier(metricName, filter, this.reporterIndex);
}
+
+ public String getLogicalScope(CharacterFilter filter) {
+ return parentMetricGroup.getLogicalScope(filter);
+ }
+
+ public String getLogicalScope(CharacterFilter filter, char delimiter) {
+ return parentMetricGroup.getLogicalScope(filter, delimiter);
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/GenericMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/GenericMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/GenericMetricGroup.java
index 54fbed4..5978f2d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/GenericMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/GenericMetricGroup.java
@@ -56,4 +56,9 @@ public class GenericMetricGroup extends AbstractMetricGroup<AbstractMetricGroup<
}
return new String[] { name };
}
+
+ @Override
+ protected String getGroupName(CharacterFilter filter) {
+ return filter.filterCharacters(name);
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java
index 2f6b07a..5a35110 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobManagerMetricGroup.java
@@ -107,5 +107,10 @@ public class JobManagerMetricGroup extends ComponentMetricGroup<JobManagerMetric
protected Iterable<? extends ComponentMetricGroup> subComponents() {
return jobs.values();
}
+
+ @Override
+ protected String getGroupName(CharacterFilter filter) {
+ return "jobmanager";
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobMetricGroup.java
index 091807f..17f6189 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/JobMetricGroup.java
@@ -81,4 +81,9 @@ public abstract class JobMetricGroup<C extends ComponentMetricGroup<C>> extends
variables.put(ScopeFormat.SCOPE_JOB_ID, jobId.toString());
variables.put(ScopeFormat.SCOPE_JOB_NAME, jobName);
}
+
+ @Override
+ protected String getGroupName(CharacterFilter filter) {
+ return "job";
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/OperatorMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/OperatorMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/OperatorMetricGroup.java
index 0c823ea..37c9dd8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/OperatorMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/OperatorMetricGroup.java
@@ -83,4 +83,9 @@ public class OperatorMetricGroup extends ComponentMetricGroup<TaskMetricGroup> {
protected Iterable<? extends ComponentMetricGroup> subComponents() {
return Collections.emptyList();
}
+
+ @Override
+ protected String getGroupName(CharacterFilter filter) {
+ return "operator";
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java
index a8fa828..c3ca5fe 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskManagerMetricGroup.java
@@ -135,5 +135,10 @@ public class TaskManagerMetricGroup extends ComponentMetricGroup<TaskManagerMetr
protected Iterable<? extends ComponentMetricGroup> subComponents() {
return jobs.values();
}
+
+ @Override
+ protected String getGroupName(CharacterFilter filter) {
+ return "taskmanager";
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroup.java
index 75b8bd8..43e8e1b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskMetricGroup.java
@@ -169,4 +169,9 @@ public class TaskMetricGroup extends ComponentMetricGroup<TaskManagerJobMetricGr
protected Iterable<? extends ComponentMetricGroup> subComponents() {
return operators.values();
}
+
+ @Override
+ protected String getGroupName(CharacterFilter filter) {
+ return "task";
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java
index ca3810a..c7b392f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java
@@ -39,6 +39,11 @@ public class AbstractMetricGroupTest {
protected QueryScopeInfo createQueryServiceMetricInfo(CharacterFilter filter) {
return null;
}
+
+ @Override
+ protected String getGroupName(CharacterFilter filter) {
+ return "";
+ }
};
assertTrue(group.getAllVariables().isEmpty());
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java
index 8a1a006..665abb1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java
@@ -177,6 +177,11 @@ public class MetricGroupTest extends TestLogger {
}
@Override
+ protected String getGroupName(CharacterFilter filter) {
+ return "";
+ }
+
+ @Override
protected void addMetric(String name, Metric metric) {}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/94d1b63c/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
index 7b06cfd..aa7ea49 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
@@ -1504,13 +1504,13 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
// connect to JMX
MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
// wait until we've found all 5 offset metrics
- Set<ObjectName> offsetMetrics = mBeanServer.queryNames(new ObjectName("*:key7=current-offsets,*"), null);
+ Set<ObjectName> offsetMetrics = mBeanServer.queryNames(new ObjectName("*current-offsets*:*"), null);
while (offsetMetrics.size() < 5) { // test will time out if metrics are not properly working
if (error.f0 != null) {
// fail test early
throw error.f0;
}
- offsetMetrics = mBeanServer.queryNames(new ObjectName("*:key7=current-offsets,*"), null);
+ offsetMetrics = mBeanServer.queryNames(new ObjectName("*current-offsets*:*"), null);
Thread.sleep(50);
}
Assert.assertEquals(5, offsetMetrics.size());
@@ -1534,7 +1534,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
}
// check if producer metrics are also available.
- Set<ObjectName> producerMetrics = mBeanServer.queryNames(new ObjectName("*:key6=KafkaProducer,*"), null);
+ Set<ObjectName> producerMetrics = mBeanServer.queryNames(new ObjectName("*KafkaProducer*:*"), null);
Assert.assertTrue("No producer metrics found", producerMetrics.size() > 30);
[2/2] flink git commit: [hotfix] [metrics] Make JmxMeter class static
Posted by ch...@apache.org.
[hotfix] [metrics] Make JmxMeter class static
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9f4be44c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9f4be44c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9f4be44c
Branch: refs/heads/master
Commit: 9f4be44c2c38a06ebc0f4adc318962f9d406aa58
Parents: 94d1b63
Author: zentol <ch...@apache.org>
Authored: Fri Oct 28 14:46:17 2016 +0200
Committer: zentol <ch...@apache.org>
Committed: Mon Oct 31 13:15:18 2016 +0100
----------------------------------------------------------------------
.../src/main/java/org/apache/flink/metrics/jmx/JMXReporter.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/9f4be44c/flink-metrics/flink-metrics-jmx/src/main/java/org/apache/flink/metrics/jmx/JMXReporter.java
----------------------------------------------------------------------
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 9c1fabb..7186d30 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
@@ -432,7 +432,7 @@ public class JMXReporter implements MetricReporter {
long getCount();
}
- private class JmxMeter extends AbstractBean implements JmxMeterMBean {
+ private static class JmxMeter extends AbstractBean implements JmxMeterMBean {
private final Meter meter;