You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2016/05/30 20:43:32 UTC
[1/7] flink git commit: [hotfix] [contrib] Fix robustness of
DataStreamUtils stream collecting
Repository: flink
Updated Branches:
refs/heads/master 035f62969 -> da23ee38e
[hotfix] [contrib] Fix robustness of DataStreamUtils stream collecting
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d3e3bd52
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d3e3bd52
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d3e3bd52
Branch: refs/heads/master
Commit: d3e3bd528ed972680baebda05acce0bf39af6798
Parents: 77afe28
Author: Stephan Ewen <se...@apache.org>
Authored: Thu May 26 20:23:20 2016 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon May 30 14:41:34 2016 +0200
----------------------------------------------------------------------
.../flink/contrib/streaming/CollectSink.java | 118 ++++--------
.../contrib/streaming/DataStreamIterator.java | 134 -------------
.../contrib/streaming/DataStreamUtils.java | 65 ++++---
.../contrib/streaming/SocketStreamIterator.java | 190 +++++++++++++++++++
.../flink/contrib/streaming/CollectITCase.java | 48 ++---
.../streaming/SocketStreamIteratorTest.java | 104 ++++++++++
6 files changed, 398 insertions(+), 261 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d3e3bd52/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/CollectSink.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/CollectSink.java b/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/CollectSink.java
index 53b84ef..161eb16 100644
--- a/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/CollectSink.java
+++ b/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/CollectSink.java
@@ -17,35 +17,30 @@
package org.apache.flink.contrib.streaming;
-import java.io.DataOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
import java.net.InetAddress;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
-import org.apache.flink.streaming.api.functions.sink.SocketClientSink;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.core.memory.DataOutputView;
/**
* A specialized data sink to be used by DataStreamUtils.collect.
*/
class CollectSink<IN> extends RichSinkFunction<IN> {
+
private static final long serialVersionUID = 1L;
- private static final Logger LOG = LoggerFactory.getLogger(SocketClientSink.class);
-
private final InetAddress hostIp;
private final int port;
private final TypeSerializer<IN> serializer;
+
private transient Socket client;
- private transient DataOutputStream dataOutputStream;
- private StreamWriterDataOutputView streamWriter;
+ private transient OutputStream outputStream;
+ private transient DataOutputViewStreamWrapper streamWriter;
/**
* Creates a CollectSink that will send the data to the specified host.
@@ -60,57 +55,13 @@ class CollectSink<IN> extends RichSinkFunction<IN> {
this.serializer = serializer;
}
- /**
- * Initializes the connection to Socket.
- */
- public void initializeConnection() {
- OutputStream outputStream;
- try {
- client = new Socket(hostIp, port);
- outputStream = client.getOutputStream();
- streamWriter = new StreamWriterDataOutputView(outputStream);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- dataOutputStream = new DataOutputStream(outputStream);
- }
-
- /**
- * Called when new data arrives to the sink, and forwards it to Socket.
- *
- * @param value
- * The incoming data
- */
@Override
- public void invoke(IN value) {
+ public void invoke(IN value) throws Exception {
try {
serializer.serialize(value, streamWriter);
- } catch (IOException e) {
- if(LOG.isErrorEnabled()){
- LOG.error("Cannot send message to socket server at " + hostIp.toString() + ":" + port, e);
- }
}
- }
-
- /**
- * Closes the connection of the Socket client.
- */
- private void closeConnection(){
- try {
- dataOutputStream.flush();
- client.close();
- } catch (IOException e) {
- throw new RuntimeException("Error while closing connection with socket server at "
- + hostIp.toString() + ":" + port, e);
- } finally {
- if (client != null) {
- try {
- client.close();
- } catch (IOException e) {
- LOG.error("Cannot close connection with socket server at "
- + hostIp.toString() + ":" + port, e);
- }
- }
+ catch (Exception e) {
+ throw new IOException("Error sending data back to client (" + hostIp.toString() + ":" + port + ')', e);
}
}
@@ -119,34 +70,47 @@ class CollectSink<IN> extends RichSinkFunction<IN> {
* @param parameters Configuration.
*/
@Override
- public void open(Configuration parameters) {
- initializeConnection();
+ public void open(Configuration parameters) throws Exception {
+ try {
+ client = new Socket(hostIp, port);
+ outputStream = client.getOutputStream();
+ streamWriter = new DataOutputViewStreamWrapper(outputStream);
+ }
+ catch (IOException e) {
+ throw new IOException("Cannot connect to the client to send back the stream", e);
+ }
}
/**
* Closes the connection with the Socket server.
*/
@Override
- public void close() {
- closeConnection();
- }
-
- private static class StreamWriterDataOutputView extends DataOutputStream implements DataOutputView {
-
- public StreamWriterDataOutputView(OutputStream stream) {
- super(stream);
- }
-
- public void skipBytesToWrite(int numBytes) throws IOException {
- for (int i = 0; i < numBytes; i++) {
- write(0);
+ public void close() throws Exception {
+ try {
+ if (outputStream != null) {
+ outputStream.flush();
+ outputStream.close();
+ }
+
+ // first regular attempt to cleanly close. Failing that will escalate
+ if (client != null) {
+ client.close();
}
}
-
- public void write(DataInputView source, int numBytes) throws IOException {
- byte[] data = new byte[numBytes];
- source.readFully(data);
- write(data);
+ catch (Exception e) {
+ throw new IOException("Error while closing connection that streams data back to client at "
+ + hostIp.toString() + ":" + port, e);
+ }
+ finally {
+ // if we failed prior to closing the client, close it
+ if (client != null) {
+ try {
+ client.close();
+ }
+ catch (Throwable t) {
+ // best effort to close, we do not care about an exception here any more
+ }
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d3e3bd52/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/DataStreamIterator.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/DataStreamIterator.java b/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/DataStreamIterator.java
deleted file mode 100644
index a98740e..0000000
--- a/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/DataStreamIterator.java
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * 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.flink.contrib.streaming;
-
-import java.util.Iterator;
-import java.net.ServerSocket;
-import java.io.InputStream;
-import java.io.IOException;
-import java.io.EOFException;
-import java.util.NoSuchElementException;
-import java.util.concurrent.CountDownLatch;
-import java.io.DataInputStream;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.core.memory.DataInputView;
-
-class DataStreamIterator<T> implements Iterator<T> {
-
- ServerSocket socket;
- InputStream tcpStream;
- T next;
- private final CountDownLatch connectionAccepted = new CountDownLatch(1);
- private volatile StreamReaderDataInputView streamReader;
- private final TypeSerializer<T> serializer;
-
- DataStreamIterator(TypeSerializer serializer) {
- this.serializer = serializer;
- try {
- socket = new ServerSocket(0, 1, null);
- } catch (IOException e) {
- throw new RuntimeException("DataStreamIterator: an I/O error occurred when opening the socket", e);
- }
- (new AcceptThread()).start();
- }
-
- private class AcceptThread extends Thread {
- public void run() {
- try {
- tcpStream = socket.accept().getInputStream();
- streamReader = new StreamReaderDataInputView(tcpStream);
- connectionAccepted.countDown();
- } catch (IOException e) {
- throw new RuntimeException("DataStreamIterator.AcceptThread failed", e);
- }
- }
- }
-
- /**
- * Returns the port on which the iterator is getting the data. (Used internally.)
- * @return The port
- */
- public int getPort() {
- return socket.getLocalPort();
- }
-
- /**
- * Returns true if the DataStream has more elements.
- * (Note: blocks if there will be more elements, but they are not available yet.)
- * @return true if the DataStream has more elements
- */
- @Override
- public boolean hasNext() {
- if (next == null) {
- readNextFromStream();
- }
- return next != null;
- }
-
- /**
- * Returns the next element of the DataStream. (Blocks if it is not available yet.)
- * @return The element
- * @throws NoSuchElementException if the stream has already ended
- */
- @Override
- public T next() {
- if (next == null) {
- readNextFromStream();
- if (next == null) {
- throw new NoSuchElementException();
- }
- }
- T current = next;
- next = null;
- return current;
- }
-
- private void readNextFromStream(){
- try {
- connectionAccepted.await();
- } catch (InterruptedException e) {
- throw new RuntimeException("The calling thread of DataStreamIterator.readNextFromStream was interrupted.");
- }
- try {
- next = serializer.deserialize(streamReader);
- } catch (EOFException e) {
- next = null;
- } catch (IOException e) {
- throw new RuntimeException("DataStreamIterator could not read from deserializedStream", e);
- }
- }
-
- private static class StreamReaderDataInputView extends DataInputStream implements DataInputView {
-
- public StreamReaderDataInputView(InputStream stream) {
- super(stream);
- }
-
- public void skipBytesToRead(int numBytes) throws IOException {
- while (numBytes > 0) {
- int skipped = skipBytes(numBytes);
- numBytes -= skipped;
- }
- }
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d3e3bd52/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/DataStreamUtils.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/DataStreamUtils.java b/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/DataStreamUtils.java
index 6f4e8d9..d4ef9ee 100644
--- a/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/DataStreamUtils.java
+++ b/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/DataStreamUtils.java
@@ -17,17 +17,18 @@
package org.apache.flink.contrib.streaming;
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.UnknownHostException;
-import java.util.Iterator;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.net.ConnectionUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.RemoteStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.runtime.net.ConnectionUtils;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.Iterator;
public final class DataStreamUtils {
@@ -35,59 +36,69 @@ public final class DataStreamUtils {
* Returns an iterator to iterate over the elements of the DataStream.
* @return The iterator
*/
- public static <OUT> Iterator<OUT> collect(DataStream<OUT> stream) {
- TypeSerializer serializer = stream.getType().createSerializer(stream.getExecutionEnvironment().getConfig());
- DataStreamIterator<OUT> it = new DataStreamIterator<OUT>(serializer);
+ public static <OUT> Iterator<OUT> collect(DataStream<OUT> stream) throws IOException {
+
+ TypeSerializer<OUT> serializer = stream.getType().createSerializer(
+ stream.getExecutionEnvironment().getConfig());
+
+ SocketStreamIterator<OUT> iter = new SocketStreamIterator<OUT>(serializer);
//Find out what IP of us should be given to CollectSink, that it will be able to connect to
StreamExecutionEnvironment env = stream.getExecutionEnvironment();
InetAddress clientAddress;
- if(env instanceof RemoteStreamEnvironment) {
+
+ if (env instanceof RemoteStreamEnvironment) {
String host = ((RemoteStreamEnvironment)env).getHost();
int port = ((RemoteStreamEnvironment)env).getPort();
try {
clientAddress = ConnectionUtils.findConnectingAddress(new InetSocketAddress(host, port), 2000, 400);
- } catch (IOException e) {
- throw new RuntimeException("IOException while trying to connect to the master", e);
+ }
+ catch (Exception e) {
+ throw new IOException("Could not determine an suitable network address to " +
+ "receive back data from the streaming program.", e);
}
} else {
try {
clientAddress = InetAddress.getLocalHost();
} catch (UnknownHostException e) {
- throw new RuntimeException("getLocalHost failed", e);
+ throw new IOException("Could not determine this machines own local address to " +
+ "receive back data from the streaming program.", e);
}
}
- DataStreamSink<OUT> sink = stream.addSink(new CollectSink<OUT>(clientAddress, it.getPort(), serializer));
+ DataStreamSink<OUT> sink = stream.addSink(new CollectSink<OUT>(clientAddress, iter.getPort(), serializer));
sink.setParallelism(1); // It would not work if multiple instances would connect to the same port
- (new CallExecute(stream)).start();
-
- return it;
+ (new CallExecute(env, iter)).start();
+
+ return iter;
}
- private static class CallExecute<OUT> extends Thread {
+ private static class CallExecute extends Thread {
- DataStream<OUT> stream;
+ private final StreamExecutionEnvironment toTrigger;
+ private final SocketStreamIterator<?> toNotify;
- public CallExecute(DataStream<OUT> stream) {
- this.stream = stream;
+ private CallExecute(StreamExecutionEnvironment toTrigger, SocketStreamIterator<?> toNotify) {
+ this.toTrigger = toTrigger;
+ this.toNotify = toNotify;
}
@Override
public void run(){
try {
- stream.getExecutionEnvironment().execute();
- } catch (Exception e) {
- throw new RuntimeException("Exception in execute()", e);
+ toTrigger.execute();
+ }
+ catch (Throwable t) {
+ toNotify.notifyOfError(t);
}
}
}
+ // ------------------------------------------------------------------------
+
/**
* Private constructor to prevent instantiation.
*/
- private DataStreamUtils() {
- throw new RuntimeException();
- }
+ private DataStreamUtils() {}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d3e3bd52/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/SocketStreamIterator.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/SocketStreamIterator.java b/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/SocketStreamIterator.java
new file mode 100644
index 0000000..c65be85
--- /dev/null
+++ b/flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/SocketStreamIterator.java
@@ -0,0 +1,190 @@
+/*
+ * 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.flink.contrib.streaming;
+
+import java.net.InetAddress;
+import java.net.Socket;
+import java.util.Iterator;
+import java.net.ServerSocket;
+import java.io.IOException;
+import java.io.EOFException;
+import java.util.NoSuchElementException;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+
+/**
+ * An iterator that returns the data from a socket stream.
+ *
+ * <p>The iterator's constructor opens a server socket. In the first call to {@link #next()}
+ * or {@link #hasNext()}, the iterator waits for a socket to connect, and starts receiving,
+ * deserializing, and returning the data from that socket.
+ *
+ * @param <T> The type of elements returned from the iterator.
+ */
+class SocketStreamIterator<T> implements Iterator<T> {
+
+ /** Server socket to listen at */
+ private final ServerSocket socket;
+
+ /** Serializer to deserialize stream */
+ private final TypeSerializer<T> serializer;
+
+ /** Set by the same thread that reads it */
+ private DataInputViewStreamWrapper inStream;
+
+ /** Next element, handover from hasNext() to next() */
+ private T next;
+
+ /** The socket for the specific stream */
+ private Socket connectedSocket;
+
+ /** Async error, for example by the executor of the program that produces the stream */
+ private volatile Throwable error;
+
+
+ SocketStreamIterator(TypeSerializer<T> serializer) throws IOException {
+ this.serializer = serializer;
+ try {
+ socket = new ServerSocket(0, 1);
+ }
+ catch (IOException e) {
+ throw new RuntimeException("Could not open socket to receive back stream results");
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // properties
+ // ------------------------------------------------------------------------
+
+ /**
+ * Returns the port on which the iterator is getting the data. (Used internally.)
+ * @return The port
+ */
+ public int getPort() {
+ return socket.getLocalPort();
+ }
+
+ public InetAddress getBindAddress() {
+ return socket.getInetAddress();
+ }
+
+ public void close() {
+ if (connectedSocket != null) {
+ try {
+ connectedSocket.close();
+ } catch (Throwable ignored) {}
+ }
+
+ try {
+ socket.close();
+ } catch (Throwable ignored) {}
+ }
+
+ // ------------------------------------------------------------------------
+ // iterator semantics
+ // ------------------------------------------------------------------------
+
+ /**
+ * Returns true if the DataStream has more elements.
+ * (Note: blocks if there will be more elements, but they are not available yet.)
+ * @return true if the DataStream has more elements
+ */
+ @Override
+ public boolean hasNext() {
+ if (next == null) {
+ try {
+ next = readNextFromStream();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to receive next element: " + e.getMessage(), e);
+ }
+ }
+
+ return next != null;
+ }
+
+ /**
+ * Returns the next element of the DataStream. (Blocks if it is not available yet.)
+ * @return The element
+ * @throws NoSuchElementException if the stream has already ended
+ */
+ @Override
+ public T next() {
+ if (hasNext()) {
+ T current = next;
+ next = null;
+ return current;
+ } else {
+ throw new NoSuchElementException();
+ }
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ private T readNextFromStream() throws Exception {
+ try {
+ if (inStream == null) {
+ connectedSocket = socket.accept();
+ inStream = new DataInputViewStreamWrapper(connectedSocket.getInputStream());
+ }
+
+ return serializer.deserialize(inStream);
+ }
+ catch (EOFException e) {
+ try {
+ connectedSocket.close();
+ } catch (Throwable ignored) {}
+
+ try {
+ socket.close();
+ } catch (Throwable ignored) {}
+
+ return null;
+ }
+ catch (Exception e) {
+ if (error == null) {
+ throw e;
+ }
+ else {
+ // throw the root cause error
+ throw new Exception("Receiving stream failed: " + error.getMessage(), error);
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // errors
+ // ------------------------------------------------------------------------
+
+ public void notifyOfError(Throwable error) {
+ if (error != null && this.error == null) {
+ this.error = error;
+
+ // this should wake up any blocking calls
+ try {
+ connectedSocket.close();
+ } catch (Throwable ignored) {}
+ try {
+ socket.close();
+ } catch (Throwable ignored) {}
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d3e3bd52/flink-contrib/flink-streaming-contrib/src/test/java/org/apache/flink/contrib/streaming/CollectITCase.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-streaming-contrib/src/test/java/org/apache/flink/contrib/streaming/CollectITCase.java b/flink-contrib/flink-streaming-contrib/src/test/java/org/apache/flink/contrib/streaming/CollectITCase.java
index fab5c9a..10ea85c 100644
--- a/flink-contrib/flink-streaming-contrib/src/test/java/org/apache/flink/contrib/streaming/CollectITCase.java
+++ b/flink-contrib/flink-streaming-contrib/src/test/java/org/apache/flink/contrib/streaming/CollectITCase.java
@@ -18,43 +18,45 @@
package org.apache.flink.contrib.streaming;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.test.util.ForkableFlinkMiniCluster;
+
import org.junit.Test;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.junit.Assert;
import java.util.Iterator;
+import static org.junit.Assert.*;
+
/**
* This test verifies the behavior of DataStreamUtils.collect.
*/
public class CollectITCase {
@Test
- public void testCollect() {
-
- Configuration config = new Configuration();
- ForkableFlinkMiniCluster cluster = new ForkableFlinkMiniCluster(config, false);
- cluster.start();
-
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment(
- "localhost", cluster.getLeaderRPCPort());
-
- long N = 10;
- DataStream<Long> stream = env.generateSequence(1, N);
-
- long i = 1;
- for(Iterator it = DataStreamUtils.collect(stream); it.hasNext(); ) {
- Long x = (Long) it.next();
- if(x != i) {
- Assert.fail(String.format("Should have got %d, got %d instead.", i, x));
+ public void testCollect() throws Exception {
+ final ForkableFlinkMiniCluster cluster = new ForkableFlinkMiniCluster(new Configuration(), false);
+ try {
+ cluster.start();
+
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment(
+ "localhost", cluster.getLeaderRPCPort());
+
+ final long N = 10;
+ DataStream<Long> stream = env.generateSequence(1, N);
+
+ long i = 1;
+ for (Iterator<Long> it = DataStreamUtils.collect(stream); it.hasNext(); ) {
+ long x = it.next();
+ assertEquals("received wrong element", i, x);
+ i++;
}
- i++;
+
+ assertEquals("received wrong number of elements", N + 1, i);
}
- if(i != N + 1) {
- Assert.fail(String.format("Should have collected %d numbers, got %d instead.", N, i - 1));
+ finally {
+ cluster.stop();
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d3e3bd52/flink-contrib/flink-streaming-contrib/src/test/java/org/apache/flink/contrib/streaming/SocketStreamIteratorTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-streaming-contrib/src/test/java/org/apache/flink/contrib/streaming/SocketStreamIteratorTest.java b/flink-contrib/flink-streaming-contrib/src/test/java/org/apache/flink/contrib/streaming/SocketStreamIteratorTest.java
new file mode 100644
index 0000000..f8739a3
--- /dev/null
+++ b/flink-contrib/flink-streaming-contrib/src/test/java/org/apache/flink/contrib/streaming/SocketStreamIteratorTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.flink.contrib.streaming;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+
+import org.junit.Test;
+
+import java.net.Socket;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.*;
+
+public class SocketStreamIteratorTest {
+
+ @Test
+ public void testIterator() throws Exception {
+
+ final AtomicReference<Throwable> error = new AtomicReference<>();
+
+ final long seed = new Random().nextLong();
+ final int numElements = 1000;
+
+ final SocketStreamIterator<Long> iterator = new SocketStreamIterator<>(LongSerializer.INSTANCE);
+
+ Thread writer = new Thread() {
+
+ @Override
+ public void run() {
+ try {
+ try (Socket sock = new Socket(iterator.getBindAddress(), iterator.getPort());
+ DataOutputViewStreamWrapper out = new DataOutputViewStreamWrapper(sock.getOutputStream()))
+ {
+ final TypeSerializer<Long> serializer = LongSerializer.INSTANCE;
+ final Random rnd = new Random(seed);
+
+ for (int i = 0; i < numElements; i++) {
+ serializer.serialize(rnd.nextLong(), out);
+ }
+ }
+ }
+ catch (Throwable t) {
+ error.set(t);
+ }
+ }
+ };
+
+ writer.start();
+
+ final Random validator = new Random(seed);
+ for (int i = 0; i < numElements; i++) {
+ assertTrue(iterator.hasNext());
+ assertTrue(iterator.hasNext());
+ assertEquals(validator.nextLong(), iterator.next().longValue());
+ }
+
+ assertFalse(iterator.hasNext());
+ writer.join();
+ assertFalse(iterator.hasNext());
+ }
+
+ @Test
+ public void testIteratorWithException() throws Exception {
+
+ final SocketStreamIterator<Long> iterator = new SocketStreamIterator<>(LongSerializer.INSTANCE);
+
+ // asynchronously set an error
+ new Thread() {
+ @Override
+ public void run() {
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException ignored) {}
+ iterator.notifyOfError(new Exception("test"));
+ }
+ }.start();
+
+ try {
+ iterator.hasNext();
+ }
+ catch (Exception e) {
+ assertTrue(e.getCause().getMessage().contains("test"));
+ }
+ }
+}
[3/7] flink git commit: [FLINK-1502] [core] Cleanups, robustness,
and performance improvements in the metrics system
Posted by se...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyReporter.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyReporter.java b/flink-core/src/test/java/org/apache/flink/metrics/util/DummyReporter.java
deleted file mode 100644
index 23a7768..0000000
--- a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyReporter.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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.flink.metrics.util;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.Metric;
-import org.apache.flink.metrics.reporter.MetricReporter;
-
-import java.util.List;
-
-public class DummyReporter implements MetricReporter {
- @Override
- public void open(Configuration config) {
- }
-
- @Override
- public void close() {
- }
-
- @Override
- public void notifyOfAddedMetric(Metric metric, String name) {
- }
-
- @Override
- public void notifyOfRemovedMetric(Metric metric, String name) {
- }
-
- @Override
- public String generateName(String name, List<String> scope) {
- return "";
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyTaskManagerMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyTaskManagerMetricGroup.java b/flink-core/src/test/java/org/apache/flink/metrics/util/DummyTaskManagerMetricGroup.java
deleted file mode 100644
index 8b7714f..0000000
--- a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyTaskManagerMetricGroup.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.flink.metrics.util;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.metrics.Metric;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.groups.TaskManagerMetricGroup;
-
-public class DummyTaskManagerMetricGroup extends TaskManagerMetricGroup {
-
- public DummyTaskManagerMetricGroup() {
- super(new DummyMetricRegistry(), "host", "id");
- }
-
- public DummyJobMetricGroup addJob(JobID id, String name) {
- return new DummyJobMetricGroup();
- }
-
- @Override
- protected void addMetric(String name, Metric metric) {}
-
- @Override
- public MetricGroup addGroup(String name) {
- return new DummyMetricGroup();
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyTaskMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyTaskMetricGroup.java b/flink-core/src/test/java/org/apache/flink/metrics/util/DummyTaskMetricGroup.java
deleted file mode 100644
index db2c557..0000000
--- a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyTaskMetricGroup.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.flink.metrics.util;
-
-import org.apache.flink.metrics.Metric;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.groups.TaskMetricGroup;
-import org.apache.flink.util.AbstractID;
-
-public class DummyTaskMetricGroup extends TaskMetricGroup {
-
- public DummyTaskMetricGroup() {
- super(new DummyMetricRegistry(), new DummyJobMetricGroup(), new AbstractID(), new AbstractID(), 0, "task");
- }
-
- public DummyOperatorMetricGroup addOperator(String name) {
- return new DummyOperatorMetricGroup();
- }
-
- @Override
- protected void addMetric(String name, Metric metric) {}
-
- @Override
- public MetricGroup addGroup(String name) {
- return new DummyMetricGroup();
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/util/TestReporter.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/util/TestReporter.java b/flink-core/src/test/java/org/apache/flink/metrics/util/TestReporter.java
index 5d8a8e0..0c139f0 100644
--- a/flink-core/src/test/java/org/apache/flink/metrics/util/TestReporter.java
+++ b/flink-core/src/test/java/org/apache/flink/metrics/util/TestReporter.java
@@ -19,10 +19,10 @@
package org.apache.flink.metrics.util;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Metric;
+import org.apache.flink.metrics.groups.AbstractMetricGroup;
import org.apache.flink.metrics.reporter.AbstractReporter;
-import java.util.List;
-
public class TestReporter extends AbstractReporter {
@Override
@@ -32,7 +32,8 @@ public class TestReporter extends AbstractReporter {
public void close() {}
@Override
- public String generateName(String name, List<String> scope) {
- return name;
- }
+ public void notifyOfAddedMetric(Metric metric, String metricName, AbstractMetricGroup group) {}
+
+ @Override
+ public void notifyOfRemovedMetric(Metric metric, String metricName, AbstractMetricGroup group) {}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
index 31b3ba2..2682584 100644
--- a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
@@ -31,8 +31,9 @@ import org.apache.flink.api.java.tuple.builder.Tuple2Builder;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.util.Collector;
+
import org.junit.Assert;
import org.junit.Test;
@@ -77,7 +78,8 @@ public class CoGroupOperatorCollectionTest implements Serializable {
final HashMap<String, Accumulator<?, ?>> accumulators = new HashMap<String, Accumulator<?, ?>>();
final HashMap<String, Future<Path>> cpTasks = new HashMap<>();
final TaskInfo taskInfo = new TaskInfo("Test UDF", 0, 4, 0);
- final RuntimeContext ctx = new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulators, new DummyMetricGroup());
+ final RuntimeContext ctx = new RuntimeUDFContext(
+ taskInfo, null, executionConfig, cpTasks, accumulators, new UnregisteredMetricsGroup());
{
SumCoGroup udf1 = new SumCoGroup();
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
index 1b627c4..c5a247a 100644
--- a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
@@ -30,7 +30,7 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.util.Collector;
import org.junit.Test;
@@ -169,9 +169,20 @@ public class GroupReduceOperatorTest implements java.io.Serializable {
ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.disableObjectReuse();
- List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskInfo, null, executionConfig, new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup()), executionConfig);
+ List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input,
+ new RuntimeUDFContext(taskInfo, null, executionConfig,
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup()),
+ executionConfig);
+
executionConfig.enableObjectReuse();
- List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskInfo, null, executionConfig, new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup()), executionConfig);
+ List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input,
+ new RuntimeUDFContext(taskInfo, null, executionConfig,
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup()),
+ executionConfig);
Set<Tuple2<String, Integer>> resultSetMutableSafe = new HashSet<Tuple2<String, Integer>>(resultMutableSafe);
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/InnerJoinOperatorBaseTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/InnerJoinOperatorBaseTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/InnerJoinOperatorBaseTest.java
index 1d5668b..89574a8 100644
--- a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/InnerJoinOperatorBaseTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/InnerJoinOperatorBaseTest.java
@@ -18,8 +18,6 @@
package org.apache.flink.api.common.operators.base;
-import static org.junit.Assert.*;
-
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.common.accumulators.Accumulator;
@@ -30,8 +28,9 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.util.Collector;
+
import org.junit.Test;
import java.io.Serializable;
@@ -43,6 +42,9 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.Future;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
@SuppressWarnings({ "unchecked", "serial" })
public class InnerJoinOperatorBaseTest implements Serializable {
@@ -107,10 +109,22 @@ public class InnerJoinOperatorBaseTest implements Serializable {
try {
final TaskInfo taskInfo = new TaskInfo("op", 0, 1, 0);
ExecutionConfig executionConfig = new ExecutionConfig();
+
executionConfig.disableObjectReuse();
- List<Tuple2<Double, String>> resultSafe = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext(taskInfo, null, executionConfig, new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup()), executionConfig);
+ List<Tuple2<Double, String>> resultSafe = base.executeOnCollections(inputData1, inputData2,
+ new RuntimeUDFContext(taskInfo, null, executionConfig,
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup()),
+ executionConfig);
+
executionConfig.enableObjectReuse();
- List<Tuple2<Double, String>> resultRegular = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext(taskInfo, null, executionConfig, new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup()), executionConfig);
+ List<Tuple2<Double, String>> resultRegular = base.executeOnCollections(inputData1, inputData2,
+ new RuntimeUDFContext(taskInfo, null, executionConfig,
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup()),
+ executionConfig);
assertEquals(expected, new HashSet<>(resultSafe));
assertEquals(expected, new HashSet<>(resultRegular));
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
index 4317c03..150854d 100644
--- a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
@@ -30,7 +30,8 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+
import org.junit.Test;
import java.util.ArrayList;
@@ -42,7 +43,9 @@ import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import static java.util.Arrays.asList;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
@SuppressWarnings({"serial", "unchecked"})
public class ReduceOperatorTest implements java.io.Serializable {
@@ -145,10 +148,22 @@ public class ReduceOperatorTest implements java.io.Serializable {
final TaskInfo taskInfo = new TaskInfo(taskName, 0, 1, 0);
ExecutionConfig executionConfig = new ExecutionConfig();
+
executionConfig.disableObjectReuse();
- List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskInfo, null, executionConfig, new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup()), executionConfig);
+ List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input,
+ new RuntimeUDFContext(taskInfo, null, executionConfig,
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup()),
+ executionConfig);
+
executionConfig.enableObjectReuse();
- List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskInfo, null, executionConfig, new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup()), executionConfig);
+ List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input,
+ new RuntimeUDFContext(taskInfo, null, executionConfig,
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup()),
+ executionConfig);
Set<Tuple2<String, Integer>> resultSetMutableSafe = new HashSet<Tuple2<String, Integer>>(resultMutableSafe);
Set<Tuple2<String, Integer>> resultSetRegular = new HashSet<Tuple2<String, Integer>>(resultRegular);
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-metric-reporters/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/ScheduledDropwizardReporter.java
----------------------------------------------------------------------
diff --git a/flink-metric-reporters/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/ScheduledDropwizardReporter.java b/flink-metric-reporters/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/ScheduledDropwizardReporter.java
index 059704d..74fdb85 100644
--- a/flink-metric-reporters/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/ScheduledDropwizardReporter.java
+++ b/flink-metric-reporters/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/ScheduledDropwizardReporter.java
@@ -15,10 +15,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
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.configuration.Configuration;
import org.apache.flink.dropwizard.metrics.CounterWrapper;
@@ -26,19 +29,19 @@ import org.apache.flink.dropwizard.metrics.GaugeWrapper;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Metric;
+import org.apache.flink.metrics.groups.AbstractMetricGroup;
import org.apache.flink.metrics.reporter.MetricReporter;
import org.apache.flink.metrics.reporter.Scheduled;
-import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
/**
* Base class for {@link org.apache.flink.metrics.reporter.MetricReporter} that wraps a
* Dropwizard {@link com.codahale.metrics.Reporter}.
*/
@PublicEvolving
-public abstract class ScheduledDropwizardReporter implements MetricReporter, Scheduled {
- protected MetricRegistry registry;
- protected ScheduledReporter reporter;
+public abstract class ScheduledDropwizardReporter implements MetricReporter, Scheduled, Reporter {
public static final String ARG_HOST = "host";
public static final String ARG_PORT = "port";
@@ -46,25 +49,24 @@ public abstract class ScheduledDropwizardReporter implements MetricReporter, Sch
public static final String ARG_CONVERSION_RATE = "rateConversion";
public static final String ARG_CONVERSION_DURATION = "durationConversion";
- protected ScheduledDropwizardReporter() {
- this.registry = new MetricRegistry();
- }
+ // ------------------------------------------------------------------------
- @Override
- public synchronized void notifyOfAddedMetric(Metric metric, String name) {
- if (metric instanceof Counter) {
- registry.register(name, new CounterWrapper((Counter) metric));
- } else if (metric instanceof Gauge) {
- registry.register(name, new GaugeWrapper((Gauge<?>) metric));
- }
- }
+ protected final MetricRegistry registry;
- @Override
- public synchronized void notifyOfRemovedMetric(Metric metric, String name) {
- registry.remove(name);
+ protected ScheduledReporter reporter;
+
+ private final Map<Gauge<?>, String> gauges = new HashMap<>();
+ private final Map<Counter, String> counters = new HashMap<>();
+
+ // ------------------------------------------------------------------------
+
+ protected ScheduledDropwizardReporter() {
+ this.registry = new MetricRegistry();
}
- public abstract ScheduledReporter getReporter(Configuration config);
+ // ------------------------------------------------------------------------
+ // life cycle
+ // ------------------------------------------------------------------------
@Override
public void open(Configuration config) {
@@ -76,24 +78,60 @@ public abstract class ScheduledDropwizardReporter implements MetricReporter, Sch
this.reporter.stop();
}
+ // ------------------------------------------------------------------------
+ // adding / removing metrics
+ // ------------------------------------------------------------------------
+
@Override
- public String generateName(String name, List<String> scope) {
- StringBuilder sb = new StringBuilder();
- for (String s : scope) {
- sb.append(s);
- sb.append('.');
+ public void notifyOfAddedMetric(Metric metric, String metricName, AbstractMetricGroup group) {
+ final String fullName = group.getScopeString() + '.' + metricName;
+
+ synchronized (this) {
+ if (metric instanceof Counter) {
+ counters.put((Counter) metric, fullName);
+ registry.register(fullName, new CounterWrapper((Counter) metric));
+ }
+ else if (metric instanceof Gauge) {
+ gauges.put((Gauge<?>) metric, fullName);
+ registry.register(fullName, GaugeWrapper.fromGauge((Gauge<?>) metric));
+ }
}
- sb.append(name);
- return sb.toString();
}
@Override
- public synchronized void report() {
- this.reporter.report(
- this.registry.getGauges(),
- this.registry.getCounters(),
- this.registry.getHistograms(),
- this.registry.getMeters(),
- this.registry.getTimers());
+ public void notifyOfRemovedMetric(Metric metric, String metricName, AbstractMetricGroup group) {
+ synchronized (this) {
+ String fullName;
+
+ if (metric instanceof Counter) {
+ fullName = counters.remove(metric);
+ } else if (metric instanceof Gauge) {
+ fullName = gauges.remove(metric);
+ } else {
+ fullName = null;
+ }
+
+ if (fullName != null) {
+ registry.remove(fullName);
+ }
+ }
}
+
+ // ------------------------------------------------------------------------
+ // scheduled reporting
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void report() {
+ synchronized (this) {
+ this.reporter.report(
+ this.registry.getGauges(),
+ this.registry.getCounters(),
+ this.registry.getHistograms(),
+ this.registry.getMeters(),
+ this.registry.getTimers());
+ }
+ }
+
+ public abstract ScheduledReporter getReporter(Configuration config);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-metric-reporters/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/GaugeWrapper.java
----------------------------------------------------------------------
diff --git a/flink-metric-reporters/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/GaugeWrapper.java b/flink-metric-reporters/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/GaugeWrapper.java
index fcb629a..655cd60 100644
--- a/flink-metric-reporters/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/GaugeWrapper.java
+++ b/flink-metric-reporters/flink-metrics-dropwizard/src/main/java/org/apache/flink/dropwizard/metrics/GaugeWrapper.java
@@ -15,11 +15,13 @@
* 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.Gauge;
public class GaugeWrapper<T> implements com.codahale.metrics.Gauge<T> {
+
private final Gauge<T> gauge;
public GaugeWrapper(Gauge<T> gauge) {
@@ -30,4 +32,10 @@ public class GaugeWrapper<T> implements com.codahale.metrics.Gauge<T> {
public T getValue() {
return this.gauge.getValue();
}
+
+ public static <T> GaugeWrapper<T> fromGauge(Gauge<?> gauge) {
+ @SuppressWarnings("unchecked")
+ Gauge<T> typedGauge = (Gauge<T>) gauge;
+ return new GaugeWrapper<>(typedGauge);
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-metric-reporters/flink-metrics-ganglia/src/main/java/org/apache/flink/metrics/ganglia/GangliaReporter.java
----------------------------------------------------------------------
diff --git a/flink-metric-reporters/flink-metrics-ganglia/src/main/java/org/apache/flink/metrics/ganglia/GangliaReporter.java b/flink-metric-reporters/flink-metrics-ganglia/src/main/java/org/apache/flink/metrics/ganglia/GangliaReporter.java
new file mode 100644
index 0000000..adf9394
--- /dev/null
+++ b/flink-metric-reporters/flink-metrics-ganglia/src/main/java/org/apache/flink/metrics/ganglia/GangliaReporter.java
@@ -0,0 +1,79 @@
+/*
+ * 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.flink.metrics.ganglia;
+
+import com.codahale.metrics.ScheduledReporter;
+
+import info.ganglia.gmetric4j.gmetric.GMetric;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.dropwizard.ScheduledDropwizardReporter;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+@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";
+ public static final String ARG_MODE_ADDRESSING = "addressingMode";
+
+ @Override
+ public ScheduledReporter getReporter(Configuration config) {
+
+ try {
+ String host = config.getString(ARG_HOST, null);
+ int port = config.getInteger(ARG_PORT, -1);
+ if (host == null || host.length() == 0 || port < 1) {
+ throw new IllegalArgumentException("Invalid host/port configuration. Host: " + host + " Port: " + port);
+ }
+ String addressingMode = config.getString(ARG_MODE_ADDRESSING, "MULTICAST");
+ int ttl = config.getInteger(ARG_TTL, -1);
+ GMetric gMetric = new GMetric(host, port, GMetric.UDPAddressingMode.valueOf(addressingMode), ttl);
+
+ String prefix = config.getString(ARG_PREFIX, null);
+ String conversionRate = config.getString(ARG_CONVERSION_RATE, null);
+ String conversionDuration = config.getString(ARG_CONVERSION_DURATION, null);
+ int dMax = config.getInteger(ARG_DMAX, 0);
+ int tMax = config.getInteger(ARG_TMAX, 60);
+
+ com.codahale.metrics.ganglia.GangliaReporter.Builder builder =
+ com.codahale.metrics.ganglia.GangliaReporter.forRegistry(registry);
+
+ if (prefix != null) {
+ builder.prefixedWith(prefix);
+ }
+ if (conversionRate != null) {
+ builder.convertRatesTo(TimeUnit.valueOf(conversionRate));
+ }
+ if (conversionDuration != null) {
+ builder.convertDurationsTo(TimeUnit.valueOf(conversionDuration));
+ }
+ builder.withDMax(dMax);
+ builder.withTMax(tMax);
+
+ return builder.build(gMetric);
+ } catch (IOException e) {
+ throw new RuntimeException("Error while instantiating GangliaReporter.", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-metric-reporters/flink-metrics-ganglia/src/main/java/org/apache/flink/metrics/graphite/GangliaReporter.java
----------------------------------------------------------------------
diff --git a/flink-metric-reporters/flink-metrics-ganglia/src/main/java/org/apache/flink/metrics/graphite/GangliaReporter.java b/flink-metric-reporters/flink-metrics-ganglia/src/main/java/org/apache/flink/metrics/graphite/GangliaReporter.java
deleted file mode 100644
index a1dafc9..0000000
--- a/flink-metric-reporters/flink-metrics-ganglia/src/main/java/org/apache/flink/metrics/graphite/GangliaReporter.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * 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.flink.metrics.graphite;
-
-import com.codahale.metrics.ScheduledReporter;
-import info.ganglia.gmetric4j.gmetric.GMetric;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.dropwizard.ScheduledDropwizardReporter;
-
-import java.io.IOException;
-import java.util.concurrent.TimeUnit;
-
-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";
- public static final String ARG_MODE_ADDRESSING = "addressingMode";
-
- @Override
- public ScheduledReporter getReporter(Configuration config) {
-
- try {
- String host = config.getString(ARG_HOST, null);
- int port = config.getInteger(ARG_PORT, -1);
- if (host == null || host.length() == 0 || port < 1) {
- throw new IllegalArgumentException("Invalid host/port configuration. Host: " + host + " Port: " + port);
- }
- String addressingMode = config.getString(ARG_MODE_ADDRESSING, "MULTICAST");
- int ttl = config.getInteger(ARG_TTL, -1);
- GMetric gMetric = new GMetric(host, port, GMetric.UDPAddressingMode.valueOf(addressingMode), ttl);
-
- String prefix = config.getString(ARG_PREFIX, null);
- String conversionRate = config.getString(ARG_CONVERSION_RATE, null);
- String conversionDuration = config.getString(ARG_CONVERSION_DURATION, null);
- int dMax = config.getInteger(ARG_DMAX, 0);
- int tMax = config.getInteger(ARG_TMAX, 60);
-
- com.codahale.metrics.ganglia.GangliaReporter.Builder builder =
- com.codahale.metrics.ganglia.GangliaReporter.forRegistry(registry);
-
- if (prefix != null) {
- builder.prefixedWith(prefix);
- }
- if (conversionRate != null) {
- builder.convertRatesTo(TimeUnit.valueOf(conversionRate));
- }
- if (conversionDuration != null) {
- builder.convertDurationsTo(TimeUnit.valueOf(conversionDuration));
- }
- builder.withDMax(dMax);
- builder.withTMax(tMax);
-
- return builder.build(gMetric);
- } catch (IOException e) {
- throw new RuntimeException("Error while instantiating GangliaReporter.", e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-metric-reporters/flink-metrics-graphite/src/main/java/org/apache/flink/metrics/graphite/GraphiteReporter.java
----------------------------------------------------------------------
diff --git a/flink-metric-reporters/flink-metrics-graphite/src/main/java/org/apache/flink/metrics/graphite/GraphiteReporter.java b/flink-metric-reporters/flink-metrics-graphite/src/main/java/org/apache/flink/metrics/graphite/GraphiteReporter.java
index b28d7a4..16be830 100644
--- a/flink-metric-reporters/flink-metrics-graphite/src/main/java/org/apache/flink/metrics/graphite/GraphiteReporter.java
+++ b/flink-metric-reporters/flink-metrics-graphite/src/main/java/org/apache/flink/metrics/graphite/GraphiteReporter.java
@@ -15,17 +15,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics.graphite;
import com.codahale.metrics.ScheduledReporter;
import com.codahale.metrics.graphite.Graphite;
+
+import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.dropwizard.ScheduledDropwizardReporter;
-import java.util.List;
import java.util.concurrent.TimeUnit;
+@PublicEvolving
public class GraphiteReporter extends ScheduledDropwizardReporter {
+
@Override
public ScheduledReporter getReporter(Configuration config) {
String host = config.getString(ARG_HOST, null);
@@ -56,15 +60,4 @@ public class GraphiteReporter extends ScheduledDropwizardReporter {
return builder.build(new Graphite(host, port));
}
-
- @Override
- public String generateName(String name, List<String> scope) {
- StringBuilder sb = new StringBuilder();
- for (String s : scope) {
- sb.append(s.replace(".", "_").replace("\"", ""));
- sb.append(".");
- }
- sb.append(name);
- return sb.toString();
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-metric-reporters/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporter.java
----------------------------------------------------------------------
diff --git a/flink-metric-reporters/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporter.java b/flink-metric-reporters/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporter.java
index e57001f..124b21d 100644
--- a/flink-metric-reporters/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporter.java
+++ b/flink-metric-reporters/flink-metrics-statsd/src/main/java/org/apache/flink/metrics/statsd/StatsDReporter.java
@@ -15,13 +15,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics.statsd;
+import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.reporter.AbstractReporter;
import org.apache.flink.metrics.reporter.Scheduled;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -30,9 +33,7 @@ import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.SocketException;
-import java.util.List;
import java.util.Map;
-import java.util.concurrent.TimeUnit;
/**
* Largely based on the StatsDReporter class by ReadyTalk
@@ -40,20 +41,19 @@ import java.util.concurrent.TimeUnit;
*
* Ported since it was not present in maven central.
*/
+@PublicEvolving
public class StatsDReporter extends AbstractReporter implements Scheduled {
+
private static final Logger LOG = LoggerFactory.getLogger(StatsDReporter.class);
public static final String ARG_HOST = "host";
public static final String ARG_PORT = "port";
- public static final String ARG_CONVERSION_RATE = "rateConversion";
- public static final String ARG_CONVERSION_DURATION = "durationConversion";
+// public static final String ARG_CONVERSION_RATE = "rateConversion";
+// public static final String ARG_CONVERSION_DURATION = "durationConversion";
private DatagramSocket socket;
private InetSocketAddress address;
- private double durationFactor;
- private double rateFactor;
-
@Override
public void open(Configuration config) {
String host = config.getString(ARG_HOST, null);
@@ -63,16 +63,17 @@ public class StatsDReporter extends AbstractReporter implements Scheduled {
throw new IllegalArgumentException("Invalid host/port configuration. Host: " + host + " Port: " + port);
}
- String conversionRate = config.getString(ARG_CONVERSION_RATE, "SECONDS");
- String conversionDuration = config.getString(ARG_CONVERSION_DURATION, "MILLISECONDS");
-
this.address = new InetSocketAddress(host, port);
- this.rateFactor = TimeUnit.valueOf(conversionRate).toSeconds(1);
- this.durationFactor = 1.0 / TimeUnit.valueOf(conversionDuration).toNanos(1);
+
+// String conversionRate = config.getString(ARG_CONVERSION_RATE, "SECONDS");
+// String conversionDuration = config.getString(ARG_CONVERSION_DURATION, "MILLISECONDS");
+// this.rateFactor = TimeUnit.valueOf(conversionRate).toSeconds(1);
+// this.durationFactor = 1.0 / TimeUnit.valueOf(conversionDuration).toNanos(1);
+
try {
this.socket = new DatagramSocket(0);
} catch (SocketException e) {
- throw new RuntimeException("Failure while creating socket. ", e);
+ throw new RuntimeException("Could not create datagram socket. ", e);
}
}
@@ -83,50 +84,40 @@ public class StatsDReporter extends AbstractReporter implements Scheduled {
}
}
- @Override
- public String generateName(String name, List<String> scope) {
- StringBuilder sb = new StringBuilder();
- for (String s : scope) {
- sb.append(s);
- sb.append('.');
- }
- sb.append(name);
- return sb.toString();
- }
-
- public void send(final String name, final double value) {
- send(name, "" + value);
- }
-
- public void send(final String name, final String value) {
- try {
- String formatted = String.format("%s:%s|g", name, value);
- byte[] data = formatted.getBytes();
- socket.send(new DatagramPacket(data, data.length, this.address));
- } catch (IOException e) {
- LOG.error("unable to send packet to statsd at '{}:{}'", address.getHostName(), address.getPort());
- }
- }
+ // ------------------------------------------------------------------------
@Override
public void report() {
- for (Map.Entry<String, Gauge<?>> entry : gauges.entrySet()) {
- reportGauge(entry.getKey(), entry.getValue());
+ for (Map.Entry<Gauge<?>, String> entry : gauges.entrySet()) {
+ reportGauge(entry.getValue(), entry.getKey());
}
- for (Map.Entry<String, Counter> entry : counters.entrySet()) {
- reportCounter(entry.getKey(), entry.getValue());
+ for (Map.Entry<Counter, String> entry : counters.entrySet()) {
+ reportCounter(entry.getValue(), entry.getKey());
}
}
+ // ------------------------------------------------------------------------
+
private void reportCounter(final String name, final Counter counter) {
- send(name, counter.getCount());
+ send(name, String.valueOf(counter.getCount()));
}
private void reportGauge(final String name, final Gauge<?> gauge) {
- final String value = gauge.getValue().toString();
+ Object value = gauge.getValue();
if (value != null) {
- send((name), value);
+ send(name, value.toString());
+ }
+ }
+
+ private void send(final String name, final String value) {
+ try {
+ String formatted = String.format("%s:%s|g", name, value);
+ byte[] data = formatted.getBytes();
+ socket.send(new DatagramPacket(data, data.length, this.address));
+ }
+ catch (IOException e) {
+ LOG.error("unable to send packet to statsd at '{}:{}'", address.getHostName(), address.getPort());
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index d711c47..ecf8d1a 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -1092,7 +1092,8 @@ class TaskManager(
val taskMetricGroup = taskManagerMetricGroup
.addTaskForJob(
tdd.getJobID, jobName,
- tdd.getVertexID, tdd.getExecutionId, tdd.getIndexInSubtaskGroup, tdd.getTaskName)
+ tdd.getVertexID, tdd.getExecutionId, tdd.getTaskName,
+ tdd.getIndexInSubtaskGroup, tdd.getAttemptNumber)
val task = new Task(
tdd,
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/drivers/TestTaskContext.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/drivers/TestTaskContext.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/drivers/TestTaskContext.java
index 15ad353..89cde95 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/drivers/TestTaskContext.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/drivers/TestTaskContext.java
@@ -26,7 +26,7 @@ import org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.memory.MemoryType;
import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.memory.MemoryManager;
@@ -227,6 +227,6 @@ public class TestTaskContext<S, T> implements TaskContext<S, T> {
@Override
public MetricGroup getMetricGroup() {
- return new DummyMetricGroup();
+ return new UnregisteredMetricsGroup();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java
index 6e9b817..766123f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/BinaryOperatorTestBase.java
@@ -27,14 +27,14 @@ import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
import org.apache.flink.api.java.typeutils.runtime.RuntimeSerializerFactory;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.runtime.operators.Driver;
-import org.apache.flink.runtime.operators.TaskContext;
import org.apache.flink.runtime.operators.ResettableDriver;
+import org.apache.flink.runtime.operators.TaskContext;
import org.apache.flink.runtime.operators.sort.UnilateralSortMerger;
import org.apache.flink.runtime.operators.util.TaskConfig;
import org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo;
@@ -371,7 +371,7 @@ public class BinaryOperatorTestBase<S extends Function, IN, OUT> extends TestLog
@Override
public MetricGroup getMetricGroup() {
- return new DummyMetricGroup();
+ return new UnregisteredMetricsGroup();
}
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
index eb2a3a7..3a69fab 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
@@ -18,40 +18,41 @@
package org.apache.flink.runtime.operators.testutils;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.LinkedList;
-import java.util.List;
-
import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.util.DummyMetricGroup;
-import org.apache.flink.runtime.operators.Driver;
-import org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo;
-import org.apache.flink.util.TestLogger;
-import org.junit.Assert;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.api.common.functions.util.FunctionUtils;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
-import org.apache.flink.runtime.testutils.recordutils.RecordComparator;
-import org.apache.flink.runtime.testutils.recordutils.RecordSerializerFactory;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.memory.MemoryManager;
-import org.apache.flink.runtime.operators.TaskContext;
+import org.apache.flink.runtime.operators.Driver;
import org.apache.flink.runtime.operators.ResettableDriver;
+import org.apache.flink.runtime.operators.TaskContext;
import org.apache.flink.runtime.operators.sort.UnilateralSortMerger;
import org.apache.flink.runtime.operators.util.TaskConfig;
+import org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo;
+import org.apache.flink.runtime.testutils.recordutils.RecordComparator;
+import org.apache.flink.runtime.testutils.recordutils.RecordSerializerFactory;
import org.apache.flink.types.Record;
import org.apache.flink.util.Collector;
import org.apache.flink.util.MutableObjectIterator;
+import org.apache.flink.util.TestLogger;
+
import org.junit.After;
+import org.junit.Assert;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
@RunWith(Parameterized.class)
public class DriverTestBase<S extends Function> extends TestLogger implements TaskContext<S, Record> {
@@ -368,7 +369,7 @@ public class DriverTestBase<S extends Function> extends TestLogger implements Ta
@Override
public MetricGroup getMetricGroup() {
- return new DummyMetricGroup();
+ return new UnregisteredMetricsGroup();
}
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java
index 9b54383..78fb422 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DummyEnvironment.java
@@ -18,17 +18,12 @@
package org.apache.flink.runtime.operators.testutils;
-import java.util.Collections;
-import java.util.Map;
-import java.util.concurrent.Future;
-
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.metrics.groups.TaskMetricGroup;
-import org.apache.flink.metrics.util.DummyTaskMetricGroup;
import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.execution.Environment;
@@ -42,12 +37,17 @@ import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.runtime.state.StateHandle;
import org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.Future;
+
public class DummyEnvironment implements Environment {
- private final TaskInfo taskInfo;
private final JobID jobId = new JobID();
private final JobVertexID jobVertexId = new JobVertexID();
+ private final ExecutionAttemptID executionId = new ExecutionAttemptID();
private final ExecutionConfig executionConfig = new ExecutionConfig();
+ private final TaskInfo taskInfo;
public DummyEnvironment(String taskName, int numSubTasks, int subTaskIndex) {
this.taskInfo = new TaskInfo(taskName, subTaskIndex, numSubTasks, 0);
@@ -70,12 +70,12 @@ public class DummyEnvironment implements Environment {
@Override
public ExecutionAttemptID getExecutionId() {
- return null;
+ return executionId;
}
@Override
public Configuration getTaskConfiguration() {
- return null;
+ return new Configuration();
}
@Override
@@ -85,12 +85,12 @@ public class DummyEnvironment implements Environment {
@Override
public TaskMetricGroup getMetricGroup() {
- return new DummyTaskMetricGroup();
+ return new UnregisteredTaskMetricsGroup();
}
@Override
public Configuration getJobConfiguration() {
- return null;
+ return new Configuration();
}
@Override
@@ -115,7 +115,7 @@ public class DummyEnvironment implements Environment {
@Override
public ClassLoader getUserClassLoader() {
- return null;
+ return getClass().getClassLoader();
}
@Override
@@ -134,12 +134,10 @@ public class DummyEnvironment implements Environment {
}
@Override
- public void acknowledgeCheckpoint(long checkpointId) {
- }
+ public void acknowledgeCheckpoint(long checkpointId) {}
@Override
- public void acknowledgeCheckpoint(long checkpointId, StateHandle<?> state) {
- }
+ public void acknowledgeCheckpoint(long checkpointId, StateHandle<?> state) {}
@Override
public ResultPartitionWriter getWriter(int index) {
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
index b774b48..0220149 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
@@ -25,7 +25,6 @@ import org.apache.flink.configuration.UnmodifiableConfiguration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.metrics.groups.TaskMetricGroup;
-import org.apache.flink.metrics.util.DummyTaskMetricGroup;
import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.execution.Environment;
@@ -216,7 +215,7 @@ public class MockEnvironment implements Environment {
@Override
public TaskMetricGroup getMetricGroup() {
- return new DummyTaskMetricGroup();
+ return new UnregisteredTaskMetricsGroup();
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/UnregisteredTaskMetricsGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/UnregisteredTaskMetricsGroup.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/UnregisteredTaskMetricsGroup.java
new file mode 100644
index 0000000..a2edce2
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/UnregisteredTaskMetricsGroup.java
@@ -0,0 +1,68 @@
+/*
+ * 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.flink.runtime.operators.testutils;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Metric;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.MetricRegistry;
+import org.apache.flink.metrics.groups.JobMetricGroup;
+import org.apache.flink.metrics.groups.TaskManagerMetricGroup;
+import org.apache.flink.metrics.groups.TaskMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+import java.util.UUID;
+
+public class UnregisteredTaskMetricsGroup extends TaskMetricGroup {
+
+ private static final MetricRegistry EMPTY_REGISTRY = new MetricRegistry(new Configuration());
+
+
+ public UnregisteredTaskMetricsGroup() {
+ super(EMPTY_REGISTRY, new DummyJobMetricGroup(),
+ new JobVertexID(), new ExecutionAttemptID(), "testtask", 0, 0);
+ }
+
+ @Override
+ protected void addMetric(String name, Metric metric) {}
+
+ @Override
+ public MetricGroup addGroup(String name) {
+ return new UnregisteredMetricsGroup();
+ }
+
+ // ------------------------------------------------------------------------
+
+ private static class DummyTaskManagerMetricsGroup extends TaskManagerMetricGroup {
+
+ public DummyTaskManagerMetricsGroup() {
+ super(EMPTY_REGISTRY, "localhost", UUID.randomUUID().toString());
+ }
+ }
+
+ private static class DummyJobMetricGroup extends JobMetricGroup {
+
+ public DummyJobMetricGroup() {
+ super(EMPTY_REGISTRY, new DummyTaskManagerMetricsGroup(), new JobID(), "testjob");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AccumulatingAlignedProcessingTimeWindowOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AccumulatingAlignedProcessingTimeWindowOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AccumulatingAlignedProcessingTimeWindowOperatorTest.java
index 812507f..eb087c6 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AccumulatingAlignedProcessingTimeWindowOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AccumulatingAlignedProcessingTimeWindowOperatorTest.java
@@ -29,14 +29,14 @@ import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.api.java.ClosureCleaner;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.util.DummyTaskMetricGroup;
import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.operators.testutils.UnregisteredTaskMetricsGroup;
+import org.apache.flink.runtime.state.AbstractStateBackend;
+import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.Output;
-import org.apache.flink.runtime.state.AbstractStateBackend;
-import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.runtime.operators.Triggerable;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -46,6 +46,7 @@ import org.apache.flink.util.Collector;
import org.junit.After;
import org.junit.Test;
+
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -61,8 +62,15 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import static org.mockito.Mockito.*;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
@SuppressWarnings({"serial", "SynchronizationOnLocalVariableOrMethodParameter"})
public class AccumulatingAlignedProcessingTimeWindowOperatorTest {
@@ -795,7 +803,7 @@ public class AccumulatingAlignedProcessingTimeWindowOperatorTest {
final Environment env = mock(Environment.class);
when(env.getTaskInfo()).thenReturn(new TaskInfo("Test task name", 0, 1, 0));
when(env.getUserClassLoader()).thenReturn(AggregatingAlignedProcessingTimeWindowOperatorTest.class.getClassLoader());
- when(env.getMetricGroup()).thenReturn(new DummyTaskMetricGroup());
+ when(env.getMetricGroup()).thenReturn(new UnregisteredTaskMetricsGroup());
when(task.getEnvironment()).thenReturn(env);
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AggregatingAlignedProcessingTimeWindowOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AggregatingAlignedProcessingTimeWindowOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AggregatingAlignedProcessingTimeWindowOperatorTest.java
index c89ac50..af46513 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AggregatingAlignedProcessingTimeWindowOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AggregatingAlignedProcessingTimeWindowOperatorTest.java
@@ -19,7 +19,6 @@
package org.apache.flink.streaming.runtime.operators.windowing;
import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
@@ -34,12 +33,12 @@ import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.util.DummyTaskMetricGroup;
import org.apache.flink.runtime.execution.Environment;
-import org.apache.flink.streaming.api.graph.StreamConfig;
-import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.runtime.operators.testutils.DummyEnvironment;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.runtime.operators.Triggerable;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.StreamTask;
@@ -47,6 +46,7 @@ import org.apache.flink.streaming.runtime.tasks.StreamTaskState;
import org.junit.After;
import org.junit.Test;
+
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -964,11 +964,7 @@ public class AggregatingAlignedProcessingTimeWindowOperatorTest {
when(task.getName()).thenReturn("Test task name");
when(task.getExecutionConfig()).thenReturn(new ExecutionConfig());
- final Environment env = mock(Environment.class);
- when(env.getTaskInfo()).thenReturn(new TaskInfo("Test task name", 0, 1, 0));
- when(env.getUserClassLoader()).thenReturn(AggregatingAlignedProcessingTimeWindowOperatorTest.class.getClassLoader());
- when(env.getMetricGroup()).thenReturn(new DummyTaskMetricGroup());
-
+ final Environment env = new DummyEnvironment("Test task name", 1, 0);
when(task.getEnvironment()).thenReturn(env);
try {
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
index b2d0196..f8c36de 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
@@ -26,7 +26,6 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.metrics.groups.TaskMetricGroup;
-import org.apache.flink.metrics.util.DummyTaskMetricGroup;
import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.event.AbstractEvent;
@@ -44,6 +43,7 @@ import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.testutils.UnregisteredTaskMetricsGroup;
import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider;
import org.apache.flink.runtime.plugable.DeserializationDelegate;
import org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate;
@@ -308,7 +308,7 @@ public class StreamMockEnvironment implements Environment {
@Override
public TaskMetricGroup getMetricGroup() {
- return new DummyTaskMetricGroup();
+ return new UnregisteredTaskMetricsGroup();
}
}
[4/7] flink git commit: [FLINK-1502] [core] Cleanups, robustness,
and performance improvements in the metrics system
Posted by se...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/groups/scope/ScopeFormats.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/groups/scope/ScopeFormats.java b/flink-core/src/main/java/org/apache/flink/metrics/groups/scope/ScopeFormats.java
new file mode 100644
index 0000000..1451637
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/metrics/groups/scope/ScopeFormats.java
@@ -0,0 +1,105 @@
+/*
+ * 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.flink.metrics.groups.scope;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.OperatorScopeFormat;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.TaskManagerJobScopeFormat;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.TaskManagerScopeFormat;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.TaskScopeFormat;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A container for component scope formats.
+ */
+@Internal
+public class ScopeFormats {
+
+ private final TaskManagerScopeFormat taskManagerFormat;
+ private final TaskManagerJobScopeFormat taskManagerJobFormat;
+ private final TaskScopeFormat taskFormat;
+ private final OperatorScopeFormat operatorFormat;
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * Creates all default scope formats.
+ */
+ public ScopeFormats() {
+ this.taskManagerFormat = new TaskManagerScopeFormat(ScopeFormat.DEFAULT_SCOPE_TASKMANAGER_COMPONENT);
+
+ this.taskManagerJobFormat = new TaskManagerJobScopeFormat(
+ ScopeFormat.DEFAULT_SCOPE_TASKMANAGER_JOB_GROUP, this.taskManagerFormat);
+
+ this.taskFormat = new TaskScopeFormat(
+ ScopeFormat.DEFAULT_SCOPE_TASK_GROUP, this.taskManagerJobFormat);
+
+ this.operatorFormat = new OperatorScopeFormat(
+ ScopeFormat.DEFAULT_SCOPE_OPERATOR_GROUP, this.taskFormat);
+ }
+
+ /**
+ * Creates all scope formats, based on the given scope format strings.
+ */
+ public ScopeFormats(
+ String taskManagerFormat,
+ String taskManagerJobFormat,
+ String taskFormat,
+ String operatorFormat)
+ {
+ this.taskManagerFormat = new TaskManagerScopeFormat(taskManagerFormat);
+ this.taskManagerJobFormat = new TaskManagerJobScopeFormat(taskManagerJobFormat, this.taskManagerFormat);
+ this.taskFormat = new TaskScopeFormat(taskFormat, this.taskManagerJobFormat);
+ this.operatorFormat = new OperatorScopeFormat(operatorFormat, this.taskFormat);
+ }
+
+ /**
+ * Creates a {@code ScopeFormats} with the given scope formats.
+ */
+ public ScopeFormats(
+ TaskManagerScopeFormat taskManagerFormat,
+ TaskManagerJobScopeFormat taskManagerJobFormat,
+ TaskScopeFormat taskFormat,
+ OperatorScopeFormat operatorFormat)
+ {
+ this.taskManagerFormat = checkNotNull(taskManagerFormat);
+ this.taskManagerJobFormat = checkNotNull(taskManagerJobFormat);
+ this.taskFormat = checkNotNull(taskFormat);
+ this.operatorFormat = checkNotNull(operatorFormat);
+ }
+
+ // ------------------------------------------------------------------------
+
+ public TaskManagerScopeFormat getTaskManagerFormat() {
+ return this.taskManagerFormat;
+ }
+
+ public TaskManagerJobScopeFormat getJobFormat() {
+ return this.taskManagerJobFormat;
+ }
+
+ public TaskScopeFormat getTaskFormat() {
+ return this.taskFormat;
+ }
+
+ public OperatorScopeFormat getOperatorFormat() {
+ return this.operatorFormat;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/reporter/AbstractReporter.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/reporter/AbstractReporter.java b/flink-core/src/main/java/org/apache/flink/metrics/reporter/AbstractReporter.java
index 271e91a..f2e78bf 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/reporter/AbstractReporter.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/reporter/AbstractReporter.java
@@ -18,33 +18,45 @@
package org.apache.flink.metrics.reporter;
+import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Metric;
+import org.apache.flink.metrics.groups.AbstractMetricGroup;
+import java.util.HashMap;
import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
+/**
+ * Base interface for custom metric reporters.
+ */
+@PublicEvolving
public abstract class AbstractReporter implements MetricReporter {
-
- protected Map<String, Gauge<?>> gauges = new ConcurrentHashMap<>();
- protected Map<String, Counter> counters = new ConcurrentHashMap<>();
+
+ protected final Map<Gauge<?>, String> gauges = new HashMap<>();
+ protected final Map<Counter, String> counters = new HashMap<>();
@Override
- public void notifyOfAddedMetric(Metric metric, String name) {
- if (metric instanceof Counter) {
- counters.put(name, (Counter) metric);
- } else if (metric instanceof Gauge) {
- gauges.put(name, (Gauge<?>) metric);
+ public void notifyOfAddedMetric(Metric metric, String metricName, AbstractMetricGroup group) {
+ final String name = group.getScopeString() + '.' + metricName;
+
+ synchronized (this) {
+ if (metric instanceof Counter) {
+ counters.put((Counter) metric, name);
+ } else if (metric instanceof Gauge) {
+ gauges.put((Gauge<?>) metric, name);
+ }
}
}
@Override
- public void notifyOfRemovedMetric(Metric metric, String name) {
- if (metric instanceof Counter) {
- counters.remove(name);
- } else if (metric instanceof Gauge) {
- gauges.remove(name);
+ public void notifyOfRemovedMetric(Metric metric, String metricName, AbstractMetricGroup group) {
+ synchronized (this) {
+ if (metric instanceof Counter) {
+ counters.remove(metric);
+ } else if (metric instanceof Gauge) {
+ gauges.remove(metric);
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/reporter/JMXReporter.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/reporter/JMXReporter.java b/flink-core/src/main/java/org/apache/flink/metrics/reporter/JMXReporter.java
index 69fa11a..71c80de 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/reporter/JMXReporter.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/reporter/JMXReporter.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics.reporter;
import org.apache.flink.annotation.Internal;
@@ -22,46 +23,76 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Metric;
+
+import org.apache.flink.metrics.groups.AbstractMetricGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.management.InstanceAlreadyExistsException;
import javax.management.InstanceNotFoundException;
-import javax.management.MBeanRegistrationException;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.NotCompliantMBeanException;
import javax.management.ObjectName;
import java.lang.management.ManagementFactory;
-import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
/**
- * {@link org.apache.flink.metrics.reporter.MetricReporter} that exports {@link org.apache.flink.metrics.Metric}s via JMX.
+ * {@link MetricReporter} that exports {@link Metric Metrics} via JMX.
*
* 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
*/
@Internal
public class JMXReporter implements MetricReporter {
- private static final Logger LOG = LoggerFactory.getLogger(JMXReporter.class);
-
- private MBeanServer mBeanServer;
private static final String PREFIX = "org.apache.flink.metrics:";
private static final String KEY_PREFIX = "key";
+ private static final Logger LOG = LoggerFactory.getLogger(JMXReporter.class);
+
+ // ------------------------------------------------------------------------
+
+ /** 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 */
+ private final Map<Metric, ObjectName> registeredMetrics;
+
+ /**
+ * Creates a new JMXReporter
+ */
public JMXReporter() {
this.mBeanServer = ManagementFactory.getPlatformMBeanServer();
+ this.registeredMetrics = new HashMap<>();
}
+ // ------------------------------------------------------------------------
+ // life cycle
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void open(Configuration config) {}
+
+ @Override
+ public void close() {}
+
+ // ------------------------------------------------------------------------
+ // adding / removing metrics
+ // ------------------------------------------------------------------------
+
@Override
- public void notifyOfAddedMetric(Metric metric, String name) {
+ public void notifyOfAddedMetric(Metric metric, String metricName, AbstractMetricGroup group) {
+ final String name = generateJmxName(metricName, group.getScopeComponents());
+
AbstractBean jmxMetric;
ObjectName jmxName;
try {
jmxName = new ObjectName(name);
} catch (MalformedObjectNameException e) {
- throw new IllegalArgumentException("Metric name did not conform to JMX ObjectName rules: " + name, e);
+ LOG.error("Metric name did not conform to JMX ObjectName rules: " + name, e);
+ return;
}
if (metric instanceof Gauge) {
@@ -69,68 +100,137 @@ public class JMXReporter implements MetricReporter {
} else if (metric instanceof Counter) {
jmxMetric = new JmxCounter((Counter) metric);
} else {
- throw new IllegalArgumentException("Unknown metric type: " + metric.getClass());
+ LOG.error("Unknown metric type: " + metric.getClass().getName());
+ return;
}
try {
- mBeanServer.registerMBean(jmxMetric, jmxName);
- } catch (NotCompliantMBeanException e) { //implementation error on our side
+ synchronized (this) {
+ mBeanServer.registerMBean(jmxMetric, jmxName);
+ registeredMetrics.put(metric, jmxName);
+ }
+ } catch (NotCompliantMBeanException e) {
+ // implementation error on our side
LOG.error("Metric did not comply with JMX MBean naming rules.", e);
} catch (InstanceAlreadyExistsException e) {
LOG.error("A metric with the name " + jmxName + " was already registered.", e);
- } catch (MBeanRegistrationException e) {
- LOG.error("Failed to register metric.", e);
+ } catch (Throwable t) {
+ LOG.error("Failed to register metric", t);
}
}
@Override
- public void notifyOfRemovedMetric(Metric metric, String name) {
+ public void notifyOfRemovedMetric(Metric metric, String metricName, AbstractMetricGroup group) {
try {
- mBeanServer.unregisterMBean(new ObjectName(name));
- } catch (MBeanRegistrationException e) {
- LOG.error("Un-registering metric failed.", e);
- } catch (MalformedObjectNameException e) {
- LOG.error("Un-registering metric failed due to invalid name.", e);
+ synchronized (this) {
+ final ObjectName jmxName = registeredMetrics.remove(metric);
+
+ // remove the metric if it is known. if it is not known, ignore the request
+ if (jmxName != null) {
+ mBeanServer.unregisterMBean(jmxName);
+ }
+ }
} catch (InstanceNotFoundException e) {
- //alright then
+ // alright then
+ } catch (Throwable t) {
+ // never propagate exceptions - the metrics reporter should not affect the stability
+ // of the running system
+ LOG.error("Un-registering metric failed", t);
}
}
- @Override
- public void open(Configuration config) {
- }
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
- @Override
- public void close() {
- }
+ static String generateJmxName(String metricName, String[] scopeComponents) {
+ final StringBuilder nameBuilder = new StringBuilder(128);
+ nameBuilder.append(PREFIX);
- @Override
- public String generateName(String name, List<String> origin) {
- StringBuilder fullName = new StringBuilder();
-
- fullName.append(PREFIX);
- for (int x = 0; x < origin.size(); x++) {
- fullName.append(KEY_PREFIX);
- fullName.append(x);
- fullName.append("=");
- String value = origin.get(x);
- value = value.replaceAll("\"", "");
- value = value.replaceAll(" ", "_");
- value = value.replaceAll("[,=;:?'*]", "-");
- fullName.append(value);
- fullName.append(",");
+ for (int x = 0; x < scopeComponents.length; x++) {
+ // write keyX=
+ nameBuilder.append(KEY_PREFIX);
+ nameBuilder.append(x);
+ nameBuilder.append("=");
+
+ // write scopeName component
+ nameBuilder.append(replaceInvalidChars(scopeComponents[x]));
+ nameBuilder.append(",");
}
- fullName.append("name=").append(name);
- return fullName.toString();
- }
+ // write the name
+ nameBuilder.append("name=").append(replaceInvalidChars(metricName));
- public interface MetricMBean {
+ return nameBuilder.toString();
}
+
+ /**
+ * 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).
+ *
+ * <p>Replacements:
+ *
+ * <ul>
+ * <li>{@code "} is removed</li>
+ * <li>{@code space} is replaced by {@code _} (underscore)</li>
+ * <li>{@code , = ; : ? ' *} are replaced by {@code -} (hyphen)</li>
+ * </ul>
+ */
+ static String replaceInvalidChars(String str) {
+ char[] chars = null;
+ final int strLen = str.length();
+ int pos = 0;
+
+ for (int i = 0; i < strLen; i++) {
+ final char c = str.charAt(i);
+ switch (c) {
+ case '"':
+ // remove character by not moving cursor
+ if (chars == null) {
+ chars = str.toCharArray();
+ }
+ break;
- private abstract static class AbstractBean implements MetricMBean {
+ case ' ':
+ if (chars == null) {
+ chars = str.toCharArray();
+ }
+ chars[pos++] = '_';
+ break;
+
+ case ',':
+ case '=':
+ case ';':
+ case ':':
+ case '?':
+ case '\'':
+ case '*':
+ if (chars == null) {
+ chars = str.toCharArray();
+ }
+ chars[pos++] = '-';
+ break;
+
+ default:
+ if (chars != null) {
+ chars[pos] = c;
+ }
+ pos++;
+ }
+ }
+
+ return chars == null ? str : new String(chars, 0, pos);
}
+ // ------------------------------------------------------------------------
+ // Interfaces and base classes for JMX beans
+ // ------------------------------------------------------------------------
+
+ public interface MetricMBean {}
+
+ private abstract static class AbstractBean implements MetricMBean {}
+
public interface JmxCounterMBean extends MetricMBean {
long getCount();
}
@@ -153,7 +253,7 @@ public class JMXReporter implements MetricReporter {
}
private static class JmxGauge extends AbstractBean implements JmxGaugeMBean {
-
+
private final Gauge<?> gauge;
public JmxGauge(Gauge<?> gauge) {
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/reporter/MetricReporter.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/reporter/MetricReporter.java b/flink-core/src/main/java/org/apache/flink/metrics/reporter/MetricReporter.java
index f87f5d3..9458246 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/reporter/MetricReporter.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/reporter/MetricReporter.java
@@ -15,28 +15,32 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics.reporter;
-import com.codahale.metrics.Reporter;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Metric;
-
-import java.util.List;
+import org.apache.flink.metrics.groups.AbstractMetricGroup;
/**
- * Reporters are used to export {@link org.apache.flink.metrics.Metric}s to an external backend.
- * <p>
- * Reporters are instantiated generically and must have a no-argument constructor.
+ * Reporters are used to export {@link Metric Metrics} to an external backend.
+ *
+ * <p>Reporters are instantiated via reflection and must be public, non-abstract, and have a
+ * public no-argument constructor.
*/
@PublicEvolving
-public interface MetricReporter extends Reporter {
-
+public interface MetricReporter {
+
+ // ------------------------------------------------------------------------
+ // life cycle
+ // ------------------------------------------------------------------------
+
/**
* 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.
- * <p>
- * This method is always called first on a newly instantiated reporter.
+ *
+ * <p>This method is always called first on a newly instantiated reporter.
*
* @param config The configuration with all parameters.
*/
@@ -47,28 +51,25 @@ public interface MetricReporter extends Reporter {
*/
void close();
- /**
- * Called when a new {@link org.apache.flink.metrics.Metric} was added.
- *
- * @param metric metric that was added
- * @param name name of the metric
- */
- void notifyOfAddedMetric(Metric metric, String name);
+ // ------------------------------------------------------------------------
+ // adding / removing metrics
+ // ------------------------------------------------------------------------
/**
- * Called when a {@link org.apache.flink.metrics.Metric} was removed.
+ * Called when a new {@link Metric} was added.
*
- * @param metric metric that was removed
- * @param name name of the metric
+ * @param metric the metric that was added
+ * @param metricName the name of the metric
+ * @param group the group that contains the metric
*/
- void notifyOfRemovedMetric(Metric metric, String name);
+ void notifyOfAddedMetric(Metric metric, String metricName, AbstractMetricGroup group);
/**
- * Generates the reported name of a metric based on it's hierarchy/scope and associated name.
+ * Called when a {@link Metric} was should be removed.
*
- * @param name name of the metric
- * @param scope hierarchy/scope of the metric
- * @return reported name
+ * @param metric the metric that should be removed
+ * @param metricName the name of the metric
+ * @param group the group that contains the metric
*/
- String generateName(String name, List<String> scope);
+ void notifyOfRemovedMetric(Metric metric, String metricName, AbstractMetricGroup group);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/reporter/Scheduled.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/reporter/Scheduled.java b/flink-core/src/main/java/org/apache/flink/metrics/reporter/Scheduled.java
index 3638f7a..cf1fc52 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/reporter/Scheduled.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/reporter/Scheduled.java
@@ -15,18 +15,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics.reporter;
import org.apache.flink.annotation.PublicEvolving;
/**
- * Marker interface for reporters that actively send out data periodically.
+ * Interface for reporters that actively send out data periodically.
*/
@PublicEvolving
public interface Scheduled {
+
/**
- * Report the current measurements.
- * This method is called in regular intervals
+ * Report the current measurements. This method is called periodically by the
+ * metrics registry that uses the reoprter.
*/
void report();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/api/common/functions/util/RuntimeUDFContextTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/functions/util/RuntimeUDFContextTest.java b/flink-core/src/test/java/org/apache/flink/api/common/functions/util/RuntimeUDFContextTest.java
index 83c88cc..8f00cd5 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/functions/util/RuntimeUDFContextTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/functions/util/RuntimeUDFContextTest.java
@@ -31,7 +31,8 @@ import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.BroadcastVariableInitializer;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+
import org.junit.Test;
@@ -42,7 +43,11 @@ public class RuntimeUDFContextTest {
@Test
public void testBroadcastVariableNotFound() {
try {
- RuntimeUDFContext ctx = new RuntimeUDFContext(taskInfo, getClass().getClassLoader(), new ExecutionConfig(), new HashMap<String, Future<Path>>(),new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup());
+ RuntimeUDFContext ctx = new RuntimeUDFContext(
+ taskInfo, getClass().getClassLoader(), new ExecutionConfig(),
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup());
try {
ctx.getBroadcastVariable("some name");
@@ -72,7 +77,11 @@ public class RuntimeUDFContextTest {
@Test
public void testBroadcastVariableSimple() {
try {
- RuntimeUDFContext ctx = new RuntimeUDFContext(taskInfo, getClass().getClassLoader(), new ExecutionConfig(), new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup());
+ RuntimeUDFContext ctx = new RuntimeUDFContext(
+ taskInfo, getClass().getClassLoader(), new ExecutionConfig(),
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup());
ctx.setBroadcastVariable("name1", Arrays.asList(1, 2, 3, 4));
ctx.setBroadcastVariable("name2", Arrays.asList(1.0, 2.0, 3.0, 4.0));
@@ -106,7 +115,11 @@ public class RuntimeUDFContextTest {
@Test
public void testBroadcastVariableWithInitializer() {
try {
- RuntimeUDFContext ctx = new RuntimeUDFContext(taskInfo, getClass().getClassLoader(), new ExecutionConfig(), new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup());
+ RuntimeUDFContext ctx = new RuntimeUDFContext(
+ taskInfo, getClass().getClassLoader(), new ExecutionConfig(),
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup());
ctx.setBroadcastVariable("name", Arrays.asList(1, 2, 3, 4));
@@ -131,7 +144,11 @@ public class RuntimeUDFContextTest {
@Test
public void testResetBroadcastVariableWithInitializer() {
try {
- RuntimeUDFContext ctx = new RuntimeUDFContext(taskInfo, getClass().getClassLoader(), new ExecutionConfig(), new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup());
+ RuntimeUDFContext ctx = new RuntimeUDFContext(
+ taskInfo, getClass().getClassLoader(), new ExecutionConfig(),
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup());
ctx.setBroadcastVariable("name", Arrays.asList(1, 2, 3, 4));
@@ -154,7 +171,11 @@ public class RuntimeUDFContextTest {
@Test
public void testBroadcastVariableWithInitializerAndMismatch() {
try {
- RuntimeUDFContext ctx = new RuntimeUDFContext(taskInfo, getClass().getClassLoader(), new ExecutionConfig(), new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup());
+ RuntimeUDFContext ctx = new RuntimeUDFContext(
+ taskInfo, getClass().getClassLoader(), new ExecutionConfig(),
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup());
ctx.setBroadcastVariable("name", Arrays.asList(1, 2, 3, 4));
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/api/common/io/RichInputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/io/RichInputFormatTest.java b/flink-core/src/test/java/org/apache/flink/api/common/io/RichInputFormatTest.java
index 554820e..ae0f8e5 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/io/RichInputFormatTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/io/RichInputFormatTest.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.api.common.io;
import java.util.HashMap;
@@ -27,11 +26,13 @@ import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.util.RuntimeUDFContext;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.types.Value;
-import org.junit.Assert;
+
import org.junit.Test;
+import static org.junit.Assert.assertEquals;
+
/**
* Tests runtime context access from inside an RichInputFormat class
*/
@@ -41,9 +42,14 @@ public class RichInputFormatTest {
public void testCheckRuntimeContextAccess() {
final SerializedInputFormat<Value> inputFormat = new SerializedInputFormat<Value>();
final TaskInfo taskInfo = new TaskInfo("test name", 1, 3, 0);
- inputFormat.setRuntimeContext(new RuntimeUDFContext(taskInfo, getClass().getClassLoader(), new ExecutionConfig(), new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup()));
+ inputFormat.setRuntimeContext(
+ new RuntimeUDFContext(
+ taskInfo, getClass().getClassLoader(), new ExecutionConfig(),
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup()));
- Assert.assertEquals(inputFormat.getRuntimeContext().getIndexOfThisSubtask(), 1);
- Assert.assertEquals(inputFormat.getRuntimeContext().getNumberOfParallelSubtasks(),3);
+ assertEquals(inputFormat.getRuntimeContext().getIndexOfThisSubtask(), 1);
+ assertEquals(inputFormat.getRuntimeContext().getNumberOfParallelSubtasks(),3);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/api/common/io/RichOutputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/io/RichOutputFormatTest.java b/flink-core/src/test/java/org/apache/flink/api/common/io/RichOutputFormatTest.java
index 09db3a9..296af11 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/io/RichOutputFormatTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/io/RichOutputFormatTest.java
@@ -16,11 +16,9 @@
* limitations under the License.
*/
-
package org.apache.flink.api.common.io;
import java.util.HashMap;
-import java.util.Map;
import java.util.concurrent.Future;
import org.apache.flink.api.common.ExecutionConfig;
@@ -28,11 +26,13 @@ import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.util.RuntimeUDFContext;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.types.Value;
-import org.junit.Assert;
+
import org.junit.Test;
+import static org.junit.Assert.assertEquals;
+
/**
* Tests runtime context access from inside an RichOutputFormat class
*/
@@ -42,9 +42,14 @@ public class RichOutputFormatTest {
public void testCheckRuntimeContextAccess() {
final SerializedOutputFormat<Value> inputFormat = new SerializedOutputFormat<Value>();
final TaskInfo taskInfo = new TaskInfo("test name", 1, 3, 0);
- inputFormat.setRuntimeContext(new RuntimeUDFContext(taskInfo, getClass().getClassLoader(), new ExecutionConfig(), new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup()));
+
+ inputFormat.setRuntimeContext(new RuntimeUDFContext(
+ taskInfo, getClass().getClassLoader(), new ExecutionConfig(),
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup()));
- Assert.assertEquals(inputFormat.getRuntimeContext().getIndexOfThisSubtask(), 1);
- Assert.assertEquals(inputFormat.getRuntimeContext().getNumberOfParallelSubtasks(),3);
+ assertEquals(inputFormat.getRuntimeContext().getIndexOfThisSubtask(), 1);
+ assertEquals(inputFormat.getRuntimeContext().getNumberOfParallelSubtasks(),3);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/api/common/operators/GenericDataSinkBaseTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/operators/GenericDataSinkBaseTest.java b/flink-core/src/test/java/org/apache/flink/api/common/operators/GenericDataSinkBaseTest.java
index 7c905c1..71bb102 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/operators/GenericDataSinkBaseTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/operators/GenericDataSinkBaseTest.java
@@ -28,8 +28,9 @@ import org.apache.flink.api.common.operators.util.TestNonRichInputFormat;
import org.apache.flink.api.common.operators.util.TestRichOutputFormat;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.types.Nothing;
+
import org.junit.Test;
import java.util.HashMap;
@@ -95,13 +96,20 @@ public class GenericDataSinkBaseTest implements java.io.Serializable {
final TaskInfo taskInfo = new TaskInfo("test_sink", 0, 1, 0);
executionConfig.disableObjectReuse();
in.reset();
- sink.executeOnCollections(asList(TestIOData.NAMES), new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap, new DummyMetricGroup()), executionConfig);
- assertEquals(out.output, asList(TestIOData.RICH_NAMES));
+
+ sink.executeOnCollections(asList(TestIOData.NAMES), new RuntimeUDFContext(
+ taskInfo, null, executionConfig, cpTasks, accumulatorMap, new UnregisteredMetricsGroup()),
+ executionConfig);
+
+ assertEquals(out.output, asList(TestIOData.RICH_NAMES));
executionConfig.enableObjectReuse();
out.clear();
in.reset();
- sink.executeOnCollections(asList(TestIOData.NAMES), new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap, new DummyMetricGroup()), executionConfig);
+
+ sink.executeOnCollections(asList(TestIOData.NAMES), new RuntimeUDFContext(
+ taskInfo, null, executionConfig, cpTasks, accumulatorMap, new UnregisteredMetricsGroup()),
+ executionConfig);
assertEquals(out.output, asList(TestIOData.RICH_NAMES));
} catch(Exception e){
e.printStackTrace();
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/api/common/operators/GenericDataSourceBaseTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/operators/GenericDataSourceBaseTest.java b/flink-core/src/test/java/org/apache/flink/api/common/operators/GenericDataSourceBaseTest.java
index c360c62..2dabe48 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/operators/GenericDataSourceBaseTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/operators/GenericDataSourceBaseTest.java
@@ -27,7 +27,8 @@ import org.apache.flink.api.common.operators.util.TestNonRichInputFormat;
import org.apache.flink.api.common.operators.util.TestRichInputFormat;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.junit.Test;
import java.util.HashMap;
@@ -84,7 +85,11 @@ public class GenericDataSourceBaseTest implements java.io.Serializable {
executionConfig.disableObjectReuse();
assertEquals(false, in.hasBeenClosed());
assertEquals(false, in.hasBeenOpened());
- List<String> resultMutableSafe = source.executeOnCollections(new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap, new DummyMetricGroup()), executionConfig);
+
+ List<String> resultMutableSafe = source.executeOnCollections(
+ new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap,
+ new UnregisteredMetricsGroup()), executionConfig);
+
assertEquals(true, in.hasBeenClosed());
assertEquals(true, in.hasBeenOpened());
@@ -92,13 +97,18 @@ public class GenericDataSourceBaseTest implements java.io.Serializable {
executionConfig.enableObjectReuse();
assertEquals(false, in.hasBeenClosed());
assertEquals(false, in.hasBeenOpened());
- List<String> resultRegular = source.executeOnCollections(new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap, new DummyMetricGroup()), executionConfig);
+
+ List<String> resultRegular = source.executeOnCollections(
+ new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap,
+ new UnregisteredMetricsGroup()), executionConfig);
+
assertEquals(true, in.hasBeenClosed());
assertEquals(true, in.hasBeenOpened());
assertEquals(asList(TestIOData.RICH_NAMES), resultMutableSafe);
assertEquals(asList(TestIOData.RICH_NAMES), resultRegular);
- } catch(Exception e){
+ }
+ catch(Exception e){
e.printStackTrace();
fail(e.getMessage());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/FlatMapOperatorCollectionTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/FlatMapOperatorCollectionTest.java b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/FlatMapOperatorCollectionTest.java
index 9447efd..f125c4b 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/FlatMapOperatorCollectionTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/FlatMapOperatorCollectionTest.java
@@ -29,8 +29,9 @@ import org.apache.flink.api.common.operators.UnaryOperatorInformation;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.util.Collector;
+
import org.junit.Assert;
import org.junit.Test;
@@ -79,7 +80,11 @@ public class FlatMapOperatorCollectionTest implements Serializable {
final TaskInfo taskInfo = new TaskInfo("Test UDF", 0, 4, 0);
// run on collections
final List<String> result = getTestFlatMapOperator(udf)
- .executeOnCollections(input, new RuntimeUDFContext(taskInfo, null, executionConfig, new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup()), executionConfig);
+ .executeOnCollections(input,
+ new RuntimeUDFContext(
+ taskInfo, null, executionConfig, new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(), new UnregisteredMetricsGroup()),
+ executionConfig);
Assert.assertEquals(input.size(), result.size());
Assert.assertEquals(input, result);
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/InnerJoinOperatorBaseTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/InnerJoinOperatorBaseTest.java b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/InnerJoinOperatorBaseTest.java
index a610a4d..8befcb9 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/InnerJoinOperatorBaseTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/InnerJoinOperatorBaseTest.java
@@ -30,8 +30,9 @@ import org.apache.flink.api.common.operators.BinaryOperatorInformation;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.util.Collector;
+
import org.junit.Test;
import java.io.Serializable;
@@ -125,10 +126,18 @@ public class InnerJoinOperatorBaseTest implements Serializable {
final HashMap<String, Future<Path>> cpTasks = new HashMap<>();
ExecutionConfig executionConfig = new ExecutionConfig();
+
executionConfig.disableObjectReuse();
- List<Integer> resultSafe = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap, new DummyMetricGroup()), executionConfig);
+ List<Integer> resultSafe = base.executeOnCollections(inputData1, inputData2,
+ new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks,
+ accumulatorMap, new UnregisteredMetricsGroup()),
+ executionConfig);
+
executionConfig.enableObjectReuse();
- List<Integer> resultRegular = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap, new DummyMetricGroup()), executionConfig);
+ List<Integer> resultRegular = base.executeOnCollections(inputData1, inputData2,
+ new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks,
+ accumulatorMap, new UnregisteredMetricsGroup()),
+ executionConfig);
assertEquals(expected, resultSafe);
assertEquals(expected, resultRegular);
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/MapOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/MapOperatorTest.java b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/MapOperatorTest.java
index 7ecdefa..d79e2a5 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/MapOperatorTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/MapOperatorTest.java
@@ -18,15 +18,6 @@
package org.apache.flink.api.common.operators.base;
-import static org.junit.Assert.*;
-import static java.util.Arrays.asList;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicBoolean;
-
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.common.accumulators.Accumulator;
@@ -35,13 +26,24 @@ import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.functions.util.RuntimeUDFContext;
import org.apache.flink.api.common.operators.UnaryOperatorInformation;
-import org.apache.flink.api.common.operators.base.MapOperatorBase;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+
import org.junit.Test;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static java.util.Arrays.asList;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
@SuppressWarnings("serial")
public class MapOperatorTest implements java.io.Serializable {
@@ -113,9 +115,17 @@ public class MapOperatorTest implements java.io.Serializable {
final TaskInfo taskInfo = new TaskInfo(taskName, 0, 1, 0);
ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.disableObjectReuse();
- List<Integer> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap, new DummyMetricGroup()), executionConfig);
+
+ List<Integer> resultMutableSafe = op.executeOnCollections(input,
+ new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks,
+ accumulatorMap, new UnregisteredMetricsGroup()),
+ executionConfig);
+
executionConfig.enableObjectReuse();
- List<Integer> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks, accumulatorMap, new DummyMetricGroup()), executionConfig);
+ List<Integer> resultRegular = op.executeOnCollections(input,
+ new RuntimeUDFContext(taskInfo, null, executionConfig, cpTasks,
+ accumulatorMap, new UnregisteredMetricsGroup()),
+ executionConfig);
assertEquals(asList(1, 2, 3, 4, 5, 6), resultMutableSafe);
assertEquals(asList(1, 2, 3, 4, 5, 6), resultRegular);
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/PartitionMapOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/PartitionMapOperatorTest.java b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/PartitionMapOperatorTest.java
index 5012718..83c194a 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/PartitionMapOperatorTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/PartitionMapOperatorTest.java
@@ -31,7 +31,7 @@ import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.metrics.util.DummyMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.util.Collector;
import org.apache.flink.api.common.functions.MapPartitionFunction;
import org.apache.flink.api.common.functions.RichMapPartitionFunction;
@@ -40,6 +40,7 @@ import org.apache.flink.api.common.functions.util.RuntimeUDFContext;
import org.apache.flink.api.common.operators.UnaryOperatorInformation;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.configuration.Configuration;
+
import org.junit.Test;
@SuppressWarnings("serial")
@@ -86,9 +87,21 @@ public class PartitionMapOperatorTest implements java.io.Serializable {
ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.disableObjectReuse();
- List<Integer> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskInfo, null, executionConfig, new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup()), executionConfig);
+
+ List<Integer> resultMutableSafe = op.executeOnCollections(input,
+ new RuntimeUDFContext(taskInfo, null, executionConfig,
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup()),
+ executionConfig);
+
executionConfig.enableObjectReuse();
- List<Integer> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskInfo, null, executionConfig, new HashMap<String, Future<Path>>(), new HashMap<String, Accumulator<?, ?>>(), new DummyMetricGroup()), executionConfig);
+ List<Integer> resultRegular = op.executeOnCollections(input,
+ new RuntimeUDFContext(taskInfo, null, executionConfig,
+ new HashMap<String, Future<Path>>(),
+ new HashMap<String, Accumulator<?, ?>>(),
+ new UnregisteredMetricsGroup()),
+ executionConfig);
assertEquals(asList(1, 2, 3, 4, 5, 6), resultMutableSafe);
assertEquals(asList(1, 2, 3, 4, 5, 6), resultRegular);
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/MetricRegistryTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/MetricRegistryTest.java b/flink-core/src/test/java/org/apache/flink/metrics/MetricRegistryTest.java
index 32cc11c..f8e0bf5 100644
--- a/flink-core/src/test/java/org/apache/flink/metrics/MetricRegistryTest.java
+++ b/flink-core/src/test/java/org/apache/flink/metrics/MetricRegistryTest.java
@@ -15,22 +15,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.groups.JobMetricGroup;
-import org.apache.flink.metrics.groups.OperatorMetricGroup;
-import org.apache.flink.metrics.groups.Scope;
+import org.apache.flink.metrics.groups.AbstractMetricGroup;
import org.apache.flink.metrics.groups.TaskManagerMetricGroup;
-import org.apache.flink.metrics.groups.TaskMetricGroup;
+import org.apache.flink.metrics.groups.scope.ScopeFormats;
import org.apache.flink.metrics.reporter.Scheduled;
import org.apache.flink.metrics.util.TestReporter;
+
import org.junit.Assert;
import org.junit.Test;
-import java.util.List;
+import static org.junit.Assert.*;
public class MetricRegistryTest {
+
/**
* Verifies that the reporter class argument is correctly used to instantiate and open the reporter.
*/
@@ -118,45 +119,11 @@ public class MetricRegistryTest {
}
/**
- * Verifies that groups are correctly created, nesting works, and names are properly forwarded to generate names.
- */
- @Test
- public void testMetricGroupGeneration() {
- Configuration config = new Configuration();
-
- config.setString(MetricRegistry.KEY_METRICS_REPORTER_CLASS, TestReporter4.class.getName());
-
- MetricRegistry registry = new MetricRegistry(config);
-
- MetricGroup root = new TaskManagerMetricGroup(registry, "host", "id");
- root.counter("rootCounter");
- root.addGroup("top").counter("topCounter");
- }
-
- protected static class TestReporter4 extends TestReporter {
- @Override
- public String generateName(String name, List<String> scope) {
- if (name.compareTo("rootCounter") == 0) {
- Assert.assertEquals("host", scope.get(0));
- return "success";
- } else if (name.compareTo("topCounter") == 0) {
- Assert.assertEquals("host", scope.get(0));
- Assert.assertEquals("taskmanager", scope.get(1));
- return "success";
- } else {
- Assert.fail();
- return null;
- }
- }
- }
-
- /**
* Verifies that reporters implementing the Listener interface are notified when Metrics are added or removed.
*/
@Test
public void testListener() {
Configuration config = new Configuration();
-
config.setString(MetricRegistry.KEY_METRICS_REPORTER_CLASS, TestReporter6.class.getName());
MetricRegistry registry = new MetricRegistry(config);
@@ -165,8 +132,8 @@ public class MetricRegistryTest {
root.counter("rootCounter");
root.close();
- Assert.assertTrue(TestReporter6.addCalled);
- Assert.assertTrue(TestReporter6.removeCalled);
+ assertTrue(TestReporter6.addCalled);
+ assertTrue(TestReporter6.removeCalled);
}
protected static class TestReporter6 extends TestReporter {
@@ -174,22 +141,22 @@ public class MetricRegistryTest {
public static boolean removeCalled = false;
@Override
- public void notifyOfAddedMetric(Metric metric, String name) {
+ public void notifyOfAddedMetric(Metric metric, String metricName, AbstractMetricGroup group) {
addCalled = true;
- Assert.assertTrue(metric instanceof Counter);
- Assert.assertEquals("rootCounter", name);
+ assertTrue(metric instanceof Counter);
+ assertEquals("rootCounter", metricName);
}
@Override
- public void notifyOfRemovedMetric(Metric metric, String name) {
+ public void notifyOfRemovedMetric(Metric metric, String metricName, AbstractMetricGroup group) {
removeCalled = true;
Assert.assertTrue(metric instanceof Counter);
- Assert.assertEquals("rootCounter", name);
+ Assert.assertEquals("rootCounter", metricName);
}
}
/**
- * Verifies that the scope configuration is properly extracted.
+ * Verifies that the scopeName configuration is properly extracted.
*/
@Test
public void testScopeConfig() {
@@ -200,18 +167,11 @@ public class MetricRegistryTest {
config.setString(MetricRegistry.KEY_METRICS_SCOPE_NAMING_TASK, "C");
config.setString(MetricRegistry.KEY_METRICS_SCOPE_NAMING_OPERATOR, "D");
- Scope.ScopeFormat scopeConfig = new MetricRegistry(config).getScopeConfig();
-
- Assert.assertEquals("A", scopeConfig.getTaskManagerFormat());
- Assert.assertEquals("B", scopeConfig.getJobFormat());
- Assert.assertEquals("C", scopeConfig.getTaskFormat());
- Assert.assertEquals("D", scopeConfig.getOperatorFormat());
-
- Scope.ScopeFormat emptyScopeConfig = new MetricRegistry(new Configuration()).getScopeConfig();
+ ScopeFormats scopeConfig = MetricRegistry.createScopeConfig(config);
- Assert.assertEquals(TaskManagerMetricGroup.DEFAULT_SCOPE_TM, emptyScopeConfig.getTaskManagerFormat());
- Assert.assertEquals(JobMetricGroup.DEFAULT_SCOPE_JOB, emptyScopeConfig.getJobFormat());
- Assert.assertEquals(TaskMetricGroup.DEFAULT_SCOPE_TASK, emptyScopeConfig.getTaskFormat());
- Assert.assertEquals(OperatorMetricGroup.DEFAULT_SCOPE_OPERATOR, emptyScopeConfig.getOperatorFormat());
+ assertEquals("A", scopeConfig.getTaskManagerFormat().format());
+ assertEquals("B", scopeConfig.getJobFormat().format());
+ assertEquals("C", scopeConfig.getTaskFormat().format());
+ assertEquals("D", scopeConfig.getOperatorFormat().format());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/groups/JobGroupTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/groups/JobGroupTest.java b/flink-core/src/test/java/org/apache/flink/metrics/groups/JobGroupTest.java
index cea11fb..e820762 100644
--- a/flink-core/src/test/java/org/apache/flink/metrics/groups/JobGroupTest.java
+++ b/flink-core/src/test/java/org/apache/flink/metrics/groups/JobGroupTest.java
@@ -15,64 +15,77 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics.groups;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.MetricRegistry;
-import org.apache.flink.util.AbstractID;
-import org.junit.Test;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.TaskManagerJobScopeFormat;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.TaskManagerScopeFormat;
-import java.util.List;
+import org.junit.Test;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
public class JobGroupTest {
+
@Test
public void testGenerateScopeDefault() {
MetricRegistry registry = new MetricRegistry(new Configuration());
- TaskMetricGroup tmGroup = new TaskManagerMetricGroup(registry, "host", "id")
- .addTaskForJob(new JobID(), "job", new AbstractID(), new AbstractID(), 0, "task");
- JobMetricGroup jmGroup = tmGroup.parent();
+ TaskManagerMetricGroup tmGroup = new TaskManagerMetricGroup(registry, "theHostName", "test-tm-id");
+ JobMetricGroup jmGroup = new JobMetricGroup(registry, tmGroup, new JobID(), "myJobName");
+
+ assertArrayEquals(
+ new String[] { "theHostName", "taskmanager", "test-tm-id", "myJobName"},
+ jmGroup.getScopeComponents());
- List<String> scope = jmGroup.generateScope();
- assertEquals(4, scope.size());
- assertEquals("job", scope.get(3));
+ assertEquals(
+ "theHostName.taskmanager.test-tm-id.myJobName",
+ jmGroup.getScopeString());
}
@Test
- public void testGenerateScopeWildcard() {
+ public void testGenerateScopeCustom() {
MetricRegistry registry = new MetricRegistry(new Configuration());
- TaskMetricGroup tmGroup = new TaskManagerMetricGroup(registry, "host", "id")
- .addTaskForJob(new JobID(), "job", new AbstractID(), new AbstractID(), 0, "task");
- JobMetricGroup jmGroup = tmGroup.parent();
+ TaskManagerScopeFormat tmFormat = new TaskManagerScopeFormat("abc");
+ TaskManagerJobScopeFormat jmFormat = new TaskManagerJobScopeFormat("some-constant.<job_name>", tmFormat);
- Scope.ScopeFormat format = new Scope.ScopeFormat();
- format.setJobFormat(Scope.concat(Scope.SCOPE_WILDCARD, "superjob", JobMetricGroup.SCOPE_JOB_NAME));
+ JobID jid = new JobID();
- List<String> scope = jmGroup.generateScope(format);
- assertEquals(5, scope.size());
- assertEquals("superjob", scope.get(3));
- assertEquals("job", scope.get(4));
+ TaskManagerMetricGroup tmGroup = new TaskManagerMetricGroup(registry, "theHostName", "test-tm-id");
+ JobMetricGroup jmGroup = new JobMetricGroup(registry, tmGroup, jmFormat, jid, "myJobName");
+
+ assertArrayEquals(
+ new String[] { "some-constant", "myJobName" },
+ jmGroup.getScopeComponents());
+
+ assertEquals(
+ "some-constant.myJobName",
+ jmGroup.getScopeString());
}
@Test
- public void testGenerateScopeCustom() {
+ public void testGenerateScopeCustomWildcard() {
MetricRegistry registry = new MetricRegistry(new Configuration());
- TaskMetricGroup tmGroup = new TaskManagerMetricGroup(registry, "host", "id")
- .addTaskForJob(new JobID(), "job", new AbstractID(), new AbstractID(), 0, "task");
- JobMetricGroup jmGroup = tmGroup.parent();
+ TaskManagerScopeFormat tmFormat = new TaskManagerScopeFormat("peter.<tm_id>");
+ TaskManagerJobScopeFormat jmFormat = new TaskManagerJobScopeFormat("*.some-constant.<job_id>", tmFormat);
+
+ JobID jid = new JobID();
+
+ TaskManagerMetricGroup tmGroup = new TaskManagerMetricGroup(registry, tmFormat, "theHostName", "test-tm-id");
+ JobMetricGroup jmGroup = new JobMetricGroup(registry, tmGroup, jmFormat, jid, "myJobName");
- Scope.ScopeFormat format = new Scope.ScopeFormat();
- format.setJobFormat(Scope.concat(TaskManagerMetricGroup.SCOPE_TM_HOST, "superjob", JobMetricGroup.SCOPE_JOB_NAME));
+ assertArrayEquals(
+ new String[] { "peter", "test-tm-id", "some-constant", jid.toString() },
+ jmGroup.getScopeComponents());
- List<String> scope = jmGroup.generateScope(format);
- assertEquals(3, scope.size());
- assertEquals("host", scope.get(0));
- assertEquals("superjob", scope.get(1));
- assertEquals("job", scope.get(2));
+ assertEquals(
+ "peter.test-tm-id.some-constant." + jid,
+ jmGroup.getScopeString());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/groups/MetricGroupRegistrationTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/groups/MetricGroupRegistrationTest.java b/flink-core/src/test/java/org/apache/flink/metrics/groups/MetricGroupRegistrationTest.java
index cb962ea..5645b94 100644
--- a/flink-core/src/test/java/org/apache/flink/metrics/groups/MetricGroupRegistrationTest.java
+++ b/flink-core/src/test/java/org/apache/flink/metrics/groups/MetricGroupRegistrationTest.java
@@ -62,13 +62,10 @@ public class MetricGroupRegistrationTest {
public static String lastPassedName;
@Override
- public void notifyOfAddedMetric(Metric metric, String name) {
+ public void notifyOfAddedMetric(Metric metric, String metricName, AbstractMetricGroup group) {
lastPassedMetric = metric;
- lastPassedName = name;
+ lastPassedName = metricName;
}
-
- @Override
- public void notifyOfRemovedMetric(Metric metric, String name) {}
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/groups/MetricGroupTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/groups/MetricGroupTest.java b/flink-core/src/test/java/org/apache/flink/metrics/groups/MetricGroupTest.java
index 31ab2a7..2849bab 100644
--- a/flink-core/src/test/java/org/apache/flink/metrics/groups/MetricGroupTest.java
+++ b/flink-core/src/test/java/org/apache/flink/metrics/groups/MetricGroupTest.java
@@ -24,7 +24,6 @@ import org.apache.flink.metrics.Metric;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.MetricRegistry;
-import org.apache.flink.metrics.util.DummyMetricGroup;
import org.junit.Test;
import static org.junit.Assert.*;
@@ -37,7 +36,8 @@ public class MetricGroupTest {
@Test
public void sameGroupOnNameCollision() {
- GenericMetricGroup group = new GenericMetricGroup(registry, new DummyMetricGroup(registry), "somegroup");
+ GenericMetricGroup group = new GenericMetricGroup(
+ registry, new DummyAbstractMetricGroup(registry), "somegroup");
String groupName = "sometestname";
MetricGroup subgroup1 = group.addGroup(groupName);
@@ -51,7 +51,7 @@ public class MetricGroupTest {
@Test
public void closedGroupDoesNotRegisterMetrics() {
GenericMetricGroup group = new GenericMetricGroup(
- exceptionOnRegister, new DummyMetricGroup(exceptionOnRegister), "testgroup");
+ exceptionOnRegister, new DummyAbstractMetricGroup(exceptionOnRegister), "testgroup");
assertFalse(group.isClosed());
group.close();
@@ -68,7 +68,7 @@ public class MetricGroupTest {
@Test
public void closedGroupCreatesClosedGroups() {
GenericMetricGroup group = new GenericMetricGroup(exceptionOnRegister,
- new DummyMetricGroup(exceptionOnRegister), "testgroup");
+ new DummyAbstractMetricGroup(exceptionOnRegister), "testgroup");
assertFalse(group.isClosed());
group.close();
@@ -82,7 +82,7 @@ public class MetricGroupTest {
public void tolerateMetricNameCollisions() {
final String name = "abctestname";
GenericMetricGroup group = new GenericMetricGroup(
- registry, new DummyMetricGroup(registry), "testgroup");
+ registry, new DummyAbstractMetricGroup(registry), "testgroup");
assertNotNull(group.counter(name));
assertNotNull(group.counter(name));
@@ -91,7 +91,8 @@ public class MetricGroupTest {
@Test
public void tolerateMetricAndGroupNameCollisions() {
final String name = "abctestname";
- GenericMetricGroup group = new GenericMetricGroup(registry, new DummyMetricGroup(registry), "testgroup");
+ GenericMetricGroup group = new GenericMetricGroup(
+ registry, new DummyAbstractMetricGroup(registry), "testgroup");
assertNotNull(group.addGroup(name));
assertNotNull(group.counter(name));
@@ -100,7 +101,7 @@ public class MetricGroupTest {
@Test(expected = IllegalArgumentException.class)
public void exceptionOnIllegalName() {
GenericMetricGroup group = new GenericMetricGroup(
- exceptionOnRegister, new DummyMetricGroup(exceptionOnRegister), "testgroup");
+ exceptionOnRegister, new DummyAbstractMetricGroup(exceptionOnRegister), "testgroup");
group.counter("�berC�unter");
}
@@ -122,4 +123,21 @@ public class MetricGroupTest {
fail("Metric should never be un-registered");
}
}
+
+ // ------------------------------------------------------------------------
+
+ private static class DummyAbstractMetricGroup extends AbstractMetricGroup {
+
+ public DummyAbstractMetricGroup(MetricRegistry registry) {
+ super(registry, new String[0]);
+ }
+
+ @Override
+ protected void addMetric(String name, Metric metric) {}
+
+ @Override
+ public MetricGroup addGroup(String name) {
+ return new DummyAbstractMetricGroup(registry);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/groups/OperatorGroupTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/groups/OperatorGroupTest.java b/flink-core/src/test/java/org/apache/flink/metrics/groups/OperatorGroupTest.java
index 6202834..cb5e082 100644
--- a/flink-core/src/test/java/org/apache/flink/metrics/groups/OperatorGroupTest.java
+++ b/flink-core/src/test/java/org/apache/flink/metrics/groups/OperatorGroupTest.java
@@ -15,73 +15,37 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics.groups;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.MetricRegistry;
import org.apache.flink.util.AbstractID;
-import org.junit.Test;
-import java.util.List;
+import org.junit.Test;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
public class OperatorGroupTest {
- @Test
- public void testGenerateScopeDefault() {
- MetricRegistry registry = new MetricRegistry(new Configuration());
-
- OperatorMetricGroup operator = new TaskManagerMetricGroup(registry, "host", "id")
- .addTaskForJob(new JobID(), "job", new AbstractID(), new AbstractID(), 0, "task")
- .addOperator("operator");
-
- List<String> scope = operator.generateScope();
- assertEquals(6, scope.size());
- assertEquals("host", scope.get(0));
- assertEquals("taskmanager", scope.get(1));
- assertEquals("id", scope.get(2));
- assertEquals("job", scope.get(3));
- assertEquals("operator", scope.get(4));
- assertEquals("0", scope.get(5));
- }
@Test
- public void testGenerateScopeWildcard() {
+ public void testGenerateScopeDefault() {
MetricRegistry registry = new MetricRegistry(new Configuration());
- OperatorMetricGroup operator = new TaskManagerMetricGroup(registry, "host", "id")
- .addTaskForJob(new JobID(), "job", new AbstractID(), new AbstractID(), 0, "task")
- .addOperator("operator");
- Scope.ScopeFormat format = new Scope.ScopeFormat();
- format.setOperatorFormat(Scope.concat(Scope.SCOPE_WILDCARD, "op", OperatorMetricGroup.SCOPE_OPERATOR_NAME));
-
- List<String> scope = operator.generateScope(format);
- assertEquals(7, scope.size());
- assertEquals("host", scope.get(0));
- assertEquals("taskmanager", scope.get(1));
- assertEquals("id", scope.get(2));
- assertEquals("job", scope.get(3));
- assertEquals("task", scope.get(4));
- assertEquals("op", scope.get(5));
- assertEquals("operator", scope.get(6));
- }
-
- @Test
- public void testGenerateScopeCustom() {
- MetricRegistry registry = new MetricRegistry(new Configuration());
- OperatorMetricGroup operator = new TaskManagerMetricGroup(registry, "host", "id")
- .addTaskForJob(new JobID(), "job", new AbstractID(), new AbstractID(), 0, "task")
- .addOperator("operator");
+ TaskManagerMetricGroup tmGroup = new TaskManagerMetricGroup(registry, "theHostName", "test-tm-id");
+ JobMetricGroup jmGroup = new JobMetricGroup(registry, tmGroup, new JobID(), "myJobName");
+ TaskMetricGroup taskGroup = new TaskMetricGroup(
+ registry, jmGroup, new AbstractID(), new AbstractID(), "aTaskName", 11, 0);
+ OperatorMetricGroup opGroup = new OperatorMetricGroup(registry, taskGroup, "myOpName");
- Scope.ScopeFormat format = new Scope.ScopeFormat();
- format.setOperatorFormat(Scope.concat("jobs", JobMetricGroup.SCOPE_JOB_NAME, "op", OperatorMetricGroup.SCOPE_OPERATOR_NAME));
+ assertArrayEquals(
+ new String[] { "theHostName", "taskmanager", "test-tm-id", "myJobName", "myOpName", "11" },
+ opGroup.getScopeComponents());
- List<String> scope = operator.generateScope(format);
- assertEquals(4, scope.size());
- assertEquals("jobs", scope.get(0));
- assertEquals("job", scope.get(1));
- assertEquals("op", scope.get(2));
- assertEquals("operator", scope.get(3));
+ assertEquals(
+ "theHostName.taskmanager.test-tm-id.myJobName.myOpName.11",
+ opGroup.getScopeString());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/groups/TaskGroupTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/groups/TaskGroupTest.java b/flink-core/src/test/java/org/apache/flink/metrics/groups/TaskGroupTest.java
index 69faee7..4a492d2 100644
--- a/flink-core/src/test/java/org/apache/flink/metrics/groups/TaskGroupTest.java
+++ b/flink-core/src/test/java/org/apache/flink/metrics/groups/TaskGroupTest.java
@@ -15,16 +15,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics.groups;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.MetricRegistry;
+import org.apache.flink.metrics.groups.scope.ScopeFormat;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.TaskManagerJobScopeFormat;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.TaskManagerScopeFormat;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.TaskScopeFormat;
import org.apache.flink.util.AbstractID;
-import org.junit.Test;
-import java.util.List;
+import org.junit.Test;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
public class TaskGroupTest {
@@ -36,48 +41,74 @@ public class TaskGroupTest {
@Test
public void testGenerateScopeDefault() {
MetricRegistry registry = new MetricRegistry(new Configuration());
-
- TaskMetricGroup operator = new TaskManagerMetricGroup(registry, "host", "id")
- .addTaskForJob(new JobID(), "job", new AbstractID(), new AbstractID(), 0, "task");
- List<String> scope = operator.generateScope();
- assertEquals(5, scope.size());
- assertEquals("task", scope.get(4));
+ AbstractID vertexId = new AbstractID();
+ AbstractID executionId = new AbstractID();
+
+ TaskManagerMetricGroup tmGroup = new TaskManagerMetricGroup(registry, "theHostName", "test-tm-id");
+ JobMetricGroup jmGroup = new JobMetricGroup(registry, tmGroup, new JobID(), "myJobName");
+ TaskMetricGroup taskGroup = new TaskMetricGroup(registry, jmGroup, vertexId, executionId, "aTaskName", 13, 2);
+
+ assertArrayEquals(
+ new String[] { "theHostName", "taskmanager", "test-tm-id", "myJobName", "aTaskName", "13"},
+ taskGroup.getScopeComponents());
+
+ assertEquals(
+ "theHostName.taskmanager.test-tm-id.myJobName.aTaskName.13",
+ taskGroup.getScopeString());
}
@Test
- public void testGenerateScopeWilcard() {
+ public void testGenerateScopeCustom() {
MetricRegistry registry = new MetricRegistry(new Configuration());
- TaskMetricGroup operator = new TaskManagerMetricGroup(registry, "host", "id")
- .addTaskForJob(new JobID(), "job", new AbstractID(), new AbstractID(), 0, "task");
-
- Scope.ScopeFormat format = new Scope.ScopeFormat();
- format.setTaskFormat(Scope.concat(Scope.SCOPE_WILDCARD, "supertask", TaskMetricGroup.SCOPE_TASK_NAME));
-
- List<String> scope = operator.generateScope(format);
- assertEquals(6, scope.size());
- assertEquals("host", scope.get(0));
- assertEquals("taskmanager", scope.get(1));
- assertEquals("id", scope.get(2));
- assertEquals("job", scope.get(3));
- assertEquals("supertask", scope.get(4));
- assertEquals("task", scope.get(5));
+
+ TaskManagerScopeFormat tmFormat = new TaskManagerScopeFormat("abc");
+ TaskManagerJobScopeFormat jmFormat = new TaskManagerJobScopeFormat("def", tmFormat);
+ TaskScopeFormat taskFormat = new TaskScopeFormat("<tm_id>.<job_id>.<task_id>.<task_attempt_id>", jmFormat);
+
+ JobID jid = new JobID();
+ AbstractID vertexId = new AbstractID();
+ AbstractID executionId = new AbstractID();
+
+ TaskManagerMetricGroup tmGroup = new TaskManagerMetricGroup(registry, "theHostName", "test-tm-id");
+ JobMetricGroup jmGroup = new JobMetricGroup(registry, tmGroup, jid, "myJobName");
+ TaskMetricGroup taskGroup = new TaskMetricGroup(
+ registry, jmGroup, taskFormat, vertexId, executionId, "aTaskName", 13, 2);
+
+ assertArrayEquals(
+ new String[] { "test-tm-id", jid.toString(), vertexId.toString(), executionId.toString() },
+ taskGroup.getScopeComponents());
+
+ assertEquals(
+ String.format("test-tm-id.%s.%s.%s", jid, vertexId, executionId),
+ taskGroup.getScopeString());
}
@Test
- public void testGenerateScopeCustom() {
+ public void testGenerateScopeWilcard() {
MetricRegistry registry = new MetricRegistry(new Configuration());
- TaskMetricGroup operator = new TaskManagerMetricGroup(registry, "host", "id")
- .addTaskForJob(new JobID(), "job", new AbstractID(), new AbstractID(), 0, "task");
-
- Scope.ScopeFormat format = new Scope.ScopeFormat();
- format.setTaskFormat(Scope.concat(TaskManagerMetricGroup.SCOPE_TM_HOST, JobMetricGroup.SCOPE_JOB_NAME, "supertask", TaskMetricGroup.SCOPE_TASK_NAME));
-
- List<String> scope = operator.generateScope(format);
- assertEquals(4, scope.size());
- assertEquals("host", scope.get(0));
- assertEquals("job", scope.get(1));
- assertEquals("supertask", scope.get(2));
- assertEquals("task", scope.get(3));
+
+ TaskManagerScopeFormat tmFormat = new TaskManagerScopeFormat(
+ ScopeFormat.DEFAULT_SCOPE_TASKMANAGER_GROUP);
+ TaskManagerJobScopeFormat jmFormat = new TaskManagerJobScopeFormat(
+ ScopeFormat.DEFAULT_SCOPE_TASKMANAGER_JOB_GROUP, tmFormat);
+
+ TaskScopeFormat format = new TaskScopeFormat("*.<task_attempt_id>.<subtask_index>", jmFormat);
+
+ AbstractID executionId = new AbstractID();
+
+ TaskManagerMetricGroup tmGroup = new TaskManagerMetricGroup(registry, "theHostName", "test-tm-id");
+ JobMetricGroup jmGroup = new JobMetricGroup(registry, tmGroup, new JobID(), "myJobName");
+
+ TaskMetricGroup taskGroup = new TaskMetricGroup(
+ registry, jmGroup, format, new AbstractID(), executionId, "aTaskName", 13, 1);
+
+ assertArrayEquals(
+ new String[] { "theHostName", "taskmanager", "test-tm-id", "myJobName", executionId.toString(), "13" },
+ taskGroup.getScopeComponents());
+
+ assertEquals(
+ "theHostName.taskmanager.test-tm-id.myJobName." + executionId + ".13",
+ taskGroup.getScopeString());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/groups/TaskManagerGroupTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/groups/TaskManagerGroupTest.java b/flink-core/src/test/java/org/apache/flink/metrics/groups/TaskManagerGroupTest.java
index 6b9a5fc..9adc1be 100644
--- a/flink-core/src/test/java/org/apache/flink/metrics/groups/TaskManagerGroupTest.java
+++ b/flink-core/src/test/java/org/apache/flink/metrics/groups/TaskManagerGroupTest.java
@@ -15,15 +15,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics.groups;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.MetricRegistry;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.TaskManagerScopeFormat;
import org.apache.flink.util.AbstractID;
-import org.junit.Test;
-import java.util.List;
+import org.junit.Test;
import static org.junit.Assert.*;
@@ -55,9 +56,9 @@ public class TaskManagerGroupTest {
final AbstractID execution13 = new AbstractID();
final AbstractID execution21 = new AbstractID();
- TaskMetricGroup tmGroup11 = group.addTaskForJob(jid1, jobName1, vertex11, execution11, 17, "test");
- TaskMetricGroup tmGroup12 = group.addTaskForJob(jid1, jobName1, vertex12, execution12, 13, "test");
- TaskMetricGroup tmGroup21 = group.addTaskForJob(jid2, jobName2, vertex21, execution21, 7, "test");
+ TaskMetricGroup tmGroup11 = group.addTaskForJob(jid1, jobName1, vertex11, execution11, "test", 17, 0);
+ TaskMetricGroup tmGroup12 = group.addTaskForJob(jid1, jobName1, vertex12, execution12, "test", 13, 1);
+ TaskMetricGroup tmGroup21 = group.addTaskForJob(jid2, jobName2, vertex21, execution21, "test", 7, 2);
assertEquals(2, group.numRegisteredJobMetricGroups());
assertFalse(tmGroup11.parent().isClosed());
@@ -77,7 +78,7 @@ public class TaskManagerGroupTest {
assertEquals(1, group.numRegisteredJobMetricGroups());
// add one more to job one
- TaskMetricGroup tmGroup13 = group.addTaskForJob(jid1, jobName1, vertex13, execution13, 0, "test");
+ TaskMetricGroup tmGroup13 = group.addTaskForJob(jid1, jobName1, vertex13, execution13, "test", 0, 0);
tmGroup12.close();
tmGroup13.close();
@@ -108,9 +109,9 @@ public class TaskManagerGroupTest {
final AbstractID execution12 = new AbstractID();
final AbstractID execution21 = new AbstractID();
- TaskMetricGroup tmGroup11 = group.addTaskForJob(jid1, jobName1, vertex11, execution11, 17, "test");
- TaskMetricGroup tmGroup12 = group.addTaskForJob(jid1, jobName1, vertex12, execution12, 13, "test");
- TaskMetricGroup tmGroup21 = group.addTaskForJob(jid2, jobName2, vertex21, execution21, 7, "test");
+ TaskMetricGroup tmGroup11 = group.addTaskForJob(jid1, jobName1, vertex11, execution11, "test", 17, 1);
+ TaskMetricGroup tmGroup12 = group.addTaskForJob(jid1, jobName1, vertex12, execution12, "test", 13, 2);
+ TaskMetricGroup tmGroup21 = group.addTaskForJob(jid2, jobName2, vertex21, execution21, "test", 7, 1);
group.close();
@@ -120,48 +121,25 @@ public class TaskManagerGroupTest {
}
// ------------------------------------------------------------------------
- // scope tests
+ // scope name tests
// ------------------------------------------------------------------------
@Test
public void testGenerateScopeDefault() {
MetricRegistry registry = new MetricRegistry(new Configuration());
- TaskManagerMetricGroup operator = new TaskManagerMetricGroup(registry, "host", "id");
-
- List<String> scope = operator.generateScope();
- assertEquals(3, scope.size());
- assertEquals("host", scope.get(0));
- assertEquals("taskmanager", scope.get(1));
- assertEquals("id", scope.get(2));
- }
-
- @Test
- public void testGenerateScopeWildcard() {
- MetricRegistry registry = new MetricRegistry(new Configuration());
- TaskManagerMetricGroup operator = new TaskManagerMetricGroup(registry, "host", "id");
+ TaskManagerMetricGroup group = new TaskManagerMetricGroup(registry, "localhost", "id");
- Scope.ScopeFormat format = new Scope.ScopeFormat();
- format.setTaskManagerFormat(Scope.concat(Scope.SCOPE_WILDCARD, "superhost", TaskManagerMetricGroup.SCOPE_TM_HOST));
-
- List<String> scope = operator.generateScope(format);
- assertEquals(2, scope.size());
- assertEquals("superhost", scope.get(0));
- assertEquals("host", scope.get(1));
+ assertArrayEquals(new String[] { "localhost", "taskmanager", "id" }, group.getScopeComponents());
+ assertEquals("localhost.taskmanager.id", group.getScopeString());
}
@Test
public void testGenerateScopeCustom() {
MetricRegistry registry = new MetricRegistry(new Configuration());
- TaskManagerMetricGroup operator = new TaskManagerMetricGroup(registry, "host", "id");
-
- Scope.ScopeFormat format = new Scope.ScopeFormat();
- format.setTaskManagerFormat(Scope.concat("h", TaskManagerMetricGroup.SCOPE_TM_HOST, "t", TaskManagerMetricGroup.SCOPE_TM_ID));
+ TaskManagerScopeFormat format = new TaskManagerScopeFormat("constant.<host>.foo.<host>");
+ TaskManagerMetricGroup group = new TaskManagerMetricGroup(registry, format, "host", "id");
- List<String> scope = operator.generateScope(format);
- assertEquals(4, scope.size());
- assertEquals("h", scope.get(0));
- assertEquals("host", scope.get(1));
- assertEquals("t", scope.get(2));
- assertEquals("id", scope.get(3));
+ assertArrayEquals(new String[] { "constant", "host", "foo", "host" }, group.getScopeComponents());
+ assertEquals("constant.host.foo.host", group.getScopeString());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/reporter/JMXReporterTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/reporter/JMXReporterTest.java b/flink-core/src/test/java/org/apache/flink/metrics/reporter/JMXReporterTest.java
index 0d683c2..abe1669 100644
--- a/flink-core/src/test/java/org/apache/flink/metrics/reporter/JMXReporterTest.java
+++ b/flink-core/src/test/java/org/apache/flink/metrics/reporter/JMXReporterTest.java
@@ -15,29 +15,40 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics.reporter;
-import org.junit.Assert;
import org.junit.Test;
-import java.util.ArrayList;
-import java.util.List;
+import static org.junit.Assert.*;
public class JMXReporterTest {
+
+ @Test
+ public void testReplaceInvalidChars() {
+ assertEquals("", JMXReporter.replaceInvalidChars(""));
+ assertEquals("abc", JMXReporter.replaceInvalidChars("abc"));
+ assertEquals("abc", JMXReporter.replaceInvalidChars("abc\""));
+ assertEquals("abc", JMXReporter.replaceInvalidChars("\"abc"));
+ assertEquals("abc", JMXReporter.replaceInvalidChars("\"abc\""));
+ assertEquals("abc", JMXReporter.replaceInvalidChars("\"a\"b\"c\""));
+ assertEquals("", JMXReporter.replaceInvalidChars("\"\"\"\""));
+ assertEquals("____", JMXReporter.replaceInvalidChars(" "));
+ assertEquals("ab_-(c)-", JMXReporter.replaceInvalidChars("\"ab ;(c)'"));
+ assertEquals("a_b_c", JMXReporter.replaceInvalidChars("a b c"));
+ assertEquals("a_b_c_", JMXReporter.replaceInvalidChars("a b c "));
+ assertEquals("a-b-c-", JMXReporter.replaceInvalidChars("a;b'c*"));
+ assertEquals("a------b------c", JMXReporter.replaceInvalidChars("a,=;:?'b,=;:?'c"));
+ }
+
/**
* Verifies that the JMXReporter properly generates the JMX name.
*/
@Test
public void testGenerateName() {
- String name = "metric";
-
- List<String> scope = new ArrayList<>();
- scope.add("value0");
- scope.add("value1");
- scope.add("\"value2 (test),=;:?'");
-
- String jmxName = new JMXReporter().generateName(name, scope);
+ String[] scope = { "value0", "value1", "\"value2 (test),=;:?'" };
+ String jmxName = JMXReporter.generateJmxName("TestMetric", scope);
- Assert.assertEquals("org.apache.flink.metrics:key0=value0,key1=value1,key2=value2_(test)------,name=metric", jmxName);
+ assertEquals("org.apache.flink.metrics:key0=value0,key1=value1,key2=value2_(test)------,name=TestMetric", jmxName);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyJobMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyJobMetricGroup.java b/flink-core/src/test/java/org/apache/flink/metrics/util/DummyJobMetricGroup.java
deleted file mode 100644
index ab78288..0000000
--- a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyJobMetricGroup.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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.flink.metrics.util;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.metrics.Metric;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.groups.JobMetricGroup;
-import org.apache.flink.util.AbstractID;
-
-public class DummyJobMetricGroup extends JobMetricGroup {
-
- public DummyJobMetricGroup() {
- super(new DummyMetricRegistry(), new DummyTaskManagerMetricGroup(), new JobID(), "job");
- }
-
- @Override
- public DummyTaskMetricGroup addTask(AbstractID id, AbstractID attemptID, int subtaskIndex, String name) {
- return new DummyTaskMetricGroup();
- }
-
- @Override
- public void removeTaskMetricGroup(AbstractID executionId) {}
-
- @Override
- protected void addMetric(String name, Metric metric) {}
-
- @Override
- public MetricGroup addGroup(String name) {
- return new DummyMetricGroup();
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyMetricGroup.java b/flink-core/src/test/java/org/apache/flink/metrics/util/DummyMetricGroup.java
deleted file mode 100644
index 77ddd17..0000000
--- a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyMetricGroup.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * 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.flink.metrics.util;
-
-import org.apache.flink.metrics.Metric;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.MetricRegistry;
-import org.apache.flink.metrics.groups.AbstractMetricGroup;
-import org.apache.flink.metrics.groups.Scope;
-
-import java.util.ArrayList;
-import java.util.List;
-
-public class DummyMetricGroup extends AbstractMetricGroup {
-
- public DummyMetricGroup() {
- this(new DummyMetricRegistry());
- }
-
- public DummyMetricGroup(MetricRegistry registry) {
- super(registry);
- }
-
- @Override
- public List<String> generateScope() {
- return new ArrayList<>();
- }
-
- @Override
- public List<String> generateScope(Scope.ScopeFormat format) {
- return new ArrayList<>();
- }
-
- @Override
- protected void addMetric(String name, Metric metric) {}
-
-
- @Override
- public MetricGroup addGroup(String name) {
- return new DummyMetricGroup();
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyMetricRegistry.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyMetricRegistry.java b/flink-core/src/test/java/org/apache/flink/metrics/util/DummyMetricRegistry.java
deleted file mode 100644
index f8b73a9..0000000
--- a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyMetricRegistry.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * 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.flink.metrics.util;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.MetricRegistry;
-
-public class DummyMetricRegistry extends MetricRegistry {
-
- public DummyMetricRegistry() {
- super(new Configuration());
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyOperatorMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyOperatorMetricGroup.java b/flink-core/src/test/java/org/apache/flink/metrics/util/DummyOperatorMetricGroup.java
deleted file mode 100644
index e271d6e..0000000
--- a/flink-core/src/test/java/org/apache/flink/metrics/util/DummyOperatorMetricGroup.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.flink.metrics.util;
-
-import org.apache.flink.metrics.Metric;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.groups.OperatorMetricGroup;
-
-public class DummyOperatorMetricGroup extends OperatorMetricGroup {
-
- public DummyOperatorMetricGroup() {
- super(new DummyMetricRegistry(), new DummyTaskMetricGroup(), "operator", 0);
- }
-
- @Override
- protected void addMetric(String name, Metric metric) {}
-
- @Override
- public MetricGroup addGroup(String name) {
- return new DummyMetricGroup();
- }
-}
[6/7] flink git commit: [FLINK-1502] [core] Temporarily set no
default metrics reporter (instead of JMX)
Posted by se...@apache.org.
[FLINK-1502] [core] Temporarily set no default metrics reporter (instead of JMX)
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/65115577
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/65115577
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/65115577
Branch: refs/heads/master
Commit: 651155775e74473f0adb3847b282bd4a8b636640
Parents: 7ad8375
Author: Stephan Ewen <se...@apache.org>
Authored: Mon May 30 17:16:57 2016 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon May 30 19:48:51 2016 +0200
----------------------------------------------------------------------
.../src/main/java/org/apache/flink/metrics/MetricRegistry.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/65115577/flink-core/src/main/java/org/apache/flink/metrics/MetricRegistry.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/MetricRegistry.java b/flink-core/src/main/java/org/apache/flink/metrics/MetricRegistry.java
index e5d3477..4efdbab 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/MetricRegistry.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/MetricRegistry.java
@@ -83,7 +83,7 @@ public class MetricRegistry {
final String className = config.getString(KEY_METRICS_REPORTER_CLASS, null);
if (className == null) {
- this.reporter = new JMXReporter();
+ this.reporter = null;
this.timer = null;
}
else {
[5/7] flink git commit: [FLINK-1502] [core] Cleanups, robustness,
and performance improvements in the metrics system
Posted by se...@apache.org.
[FLINK-1502] [core] Cleanups, robustness, and performance improvements in the metrics system
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7ad8375a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7ad8375a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7ad8375a
Branch: refs/heads/master
Commit: 7ad8375a89374bec80571029e9166f1336bdea8e
Parents: d3e3bd5
Author: Stephan Ewen <se...@apache.org>
Authored: Wed May 25 20:34:44 2016 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon May 30 14:54:10 2016 +0200
----------------------------------------------------------------------
flink-contrib/flink-storm/pom.xml | 22 +-
.../flink/storm/wrappers/BoltWrapperTest.java | 13 +-
.../common/operators/CollectionExecutor.java | 32 +-
.../java/org/apache/flink/metrics/Counter.java | 5 +-
.../java/org/apache/flink/metrics/Gauge.java | 3 +-
.../java/org/apache/flink/metrics/Metric.java | 3 +-
.../apache/flink/metrics/MetricRegistry.java | 92 ++---
.../metrics/groups/AbstractMetricGroup.java | 98 +++--
.../metrics/groups/ComponentMetricGroup.java | 109 +----
.../metrics/groups/GenericMetricGroup.java | 45 +--
.../flink/metrics/groups/IOMetricGroup.java | 27 +-
.../flink/metrics/groups/JobMetricGroup.java | 88 ++--
.../groups/NonRegisteringMetricsGroup.java | 87 ----
.../metrics/groups/OperatorMetricGroup.java | 35 +-
.../org/apache/flink/metrics/groups/Scope.java | 119 ------
.../metrics/groups/TaskManagerMetricGroup.java | 68 ++--
.../flink/metrics/groups/TaskMetricGroup.java | 126 ++++--
.../groups/UnregisteredMetricsGroup.java | 73 ++++
.../flink/metrics/groups/scope/ScopeFormat.java | 399 +++++++++++++++++++
.../metrics/groups/scope/ScopeFormats.java | 105 +++++
.../metrics/reporter/AbstractReporter.java | 40 +-
.../flink/metrics/reporter/JMXReporter.java | 194 ++++++---
.../flink/metrics/reporter/MetricReporter.java | 53 +--
.../flink/metrics/reporter/Scheduled.java | 8 +-
.../functions/util/RuntimeUDFContextTest.java | 33 +-
.../api/common/io/RichInputFormatTest.java | 18 +-
.../api/common/io/RichOutputFormatTest.java | 19 +-
.../operators/GenericDataSinkBaseTest.java | 16 +-
.../operators/GenericDataSourceBaseTest.java | 18 +-
.../base/FlatMapOperatorCollectionTest.java | 9 +-
.../base/InnerJoinOperatorBaseTest.java | 15 +-
.../common/operators/base/MapOperatorTest.java | 36 +-
.../base/PartitionMapOperatorTest.java | 19 +-
.../flink/metrics/MetricRegistryTest.java | 78 +---
.../flink/metrics/groups/JobGroupTest.java | 73 ++--
.../groups/MetricGroupRegistrationTest.java | 7 +-
.../flink/metrics/groups/MetricGroupTest.java | 32 +-
.../flink/metrics/groups/OperatorGroupTest.java | 66 +--
.../flink/metrics/groups/TaskGroupTest.java | 103 +++--
.../metrics/groups/TaskManagerGroupTest.java | 58 +--
.../flink/metrics/reporter/JMXReporterTest.java | 35 +-
.../flink/metrics/util/DummyJobMetricGroup.java | 47 ---
.../flink/metrics/util/DummyMetricGroup.java | 57 ---
.../flink/metrics/util/DummyMetricRegistry.java | 29 --
.../metrics/util/DummyOperatorMetricGroup.java | 37 --
.../flink/metrics/util/DummyReporter.java | 47 ---
.../util/DummyTaskManagerMetricGroup.java | 42 --
.../metrics/util/DummyTaskMetricGroup.java | 42 --
.../apache/flink/metrics/util/TestReporter.java | 11 +-
.../base/CoGroupOperatorCollectionTest.java | 6 +-
.../operators/base/GroupReduceOperatorTest.java | 17 +-
.../base/InnerJoinOperatorBaseTest.java | 24 +-
.../operators/base/ReduceOperatorTest.java | 23 +-
.../dropwizard/ScheduledDropwizardReporter.java | 104 +++--
.../flink/dropwizard/metrics/GaugeWrapper.java | 8 +
.../flink/metrics/ganglia/GangliaReporter.java | 79 ++++
.../flink/metrics/graphite/GangliaReporter.java | 73 ----
.../metrics/graphite/GraphiteReporter.java | 17 +-
.../flink/metrics/statsd/StatsDReporter.java | 79 ++--
.../flink/runtime/taskmanager/TaskManager.scala | 3 +-
.../operators/drivers/TestTaskContext.java | 4 +-
.../testutils/BinaryOperatorTestBase.java | 6 +-
.../operators/testutils/DriverTestBase.java | 31 +-
.../operators/testutils/DummyEnvironment.java | 28 +-
.../operators/testutils/MockEnvironment.java | 3 +-
.../testutils/UnregisteredTaskMetricsGroup.java | 68 ++++
...AlignedProcessingTimeWindowOperatorTest.java | 20 +-
...AlignedProcessingTimeWindowOperatorTest.java | 14 +-
.../runtime/tasks/StreamMockEnvironment.java | 4 +-
69 files changed, 1893 insertions(+), 1509 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-contrib/flink-storm/pom.xml
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/pom.xml b/flink-contrib/flink-storm/pom.xml
index a080a03..590f33d 100644
--- a/flink-contrib/flink-storm/pom.xml
+++ b/flink-contrib/flink-storm/pom.xml
@@ -35,6 +35,9 @@ under the License.
<packaging>jar</packaging>
<dependencies>
+
+ <!-- core dependencies -->
+
<dependency>
<!-- Together with the dependency management section in flink-parent, this
pins the Kryo version of transitive dependencies to the Flink Kryo version -->
@@ -49,14 +52,6 @@ under the License.
</dependency>
<dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-core</artifactId>
- <version>${project.version}</version>
- <type>test-jar</type>
- <scope>test</scope>
- </dependency>
-
- <dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>0.9.4</version>
@@ -77,6 +72,17 @@ under the License.
<artifactId>guava</artifactId>
<version>${guava.version}</version>
</dependency>
+
+ <!-- test dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime_2.10</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
<build>
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java
index cb9ac1c..5da12ef 100644
--- a/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java
+++ b/flink-contrib/flink-storm/src/test/java/org/apache/flink/storm/wrappers/BoltWrapperTest.java
@@ -32,10 +32,9 @@ import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.UnmodifiableConfiguration;
-import org.apache.flink.metrics.util.DummyMetricGroup;
-import org.apache.flink.metrics.util.DummyTaskMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.execution.Environment;
-import org.apache.flink.runtime.taskmanager.RuntimeEnvironment;
+import org.apache.flink.runtime.operators.testutils.UnregisteredTaskMetricsGroup;
import org.apache.flink.storm.util.AbstractTest;
import org.apache.flink.storm.util.SplitStreamType;
import org.apache.flink.storm.util.StormConfig;
@@ -144,7 +143,7 @@ public class BoltWrapperTest extends AbstractTest {
final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class);
when(taskContext.getExecutionConfig()).thenReturn(mock(ExecutionConfig.class));
when(taskContext.getTaskName()).thenReturn("name");
- when(taskContext.getMetricGroup()).thenReturn(new DummyMetricGroup());
+ when(taskContext.getMetricGroup()).thenReturn(new UnregisteredMetricsGroup());
final IRichBolt bolt = mock(IRichBolt.class);
@@ -229,7 +228,7 @@ public class BoltWrapperTest extends AbstractTest {
final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class);
when(taskContext.getExecutionConfig()).thenReturn(taskConfig);
when(taskContext.getTaskName()).thenReturn("name");
- when(taskContext.getMetricGroup()).thenReturn(new DummyMetricGroup());
+ when(taskContext.getMetricGroup()).thenReturn(new UnregisteredMetricsGroup());
final SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer();
declarer.declare(new Fields("dummy"));
@@ -294,7 +293,7 @@ public class BoltWrapperTest extends AbstractTest {
final StreamingRuntimeContext taskContext = mock(StreamingRuntimeContext.class);
when(taskContext.getExecutionConfig()).thenReturn(taskConfig);
when(taskContext.getTaskName()).thenReturn("name");
- when(taskContext.getMetricGroup()).thenReturn(new DummyMetricGroup());
+ when(taskContext.getMetricGroup()).thenReturn(new UnregisteredMetricsGroup());
final IRichBolt bolt = mock(IRichBolt.class);
BoltWrapper<Object, Object> wrapper = new BoltWrapper<Object, Object>(bolt);
@@ -367,7 +366,7 @@ public class BoltWrapperTest extends AbstractTest {
Environment env = mock(Environment.class);
when(env.getTaskInfo()).thenReturn(new TaskInfo("Mock Task", 0, 1, 0));
when(env.getUserClassLoader()).thenReturn(BoltWrapperTest.class.getClassLoader());
- when(env.getMetricGroup()).thenReturn(new DummyTaskMetricGroup());
+ when(env.getMetricGroup()).thenReturn(new UnregisteredTaskMetricsGroup());
StreamTask<?, ?> mockTask = mock(StreamTask.class);
when(mockTask.getCheckpointLock()).thenReturn(new Object());
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
index 97f0c5e..913b205 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
@@ -36,7 +36,6 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.JobExecutionResult;
-import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.Plan;
import org.apache.flink.api.common.TaskInfo;
import org.apache.flink.api.common.accumulators.Accumulator;
@@ -62,7 +61,7 @@ import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.fs.local.LocalFileSystem;
import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.groups.NonRegisteringMetricsGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.types.Value;
import org.apache.flink.util.Visitor;
@@ -72,8 +71,6 @@ import org.apache.flink.util.Visitor;
@Internal
public class CollectionExecutor {
- private static final boolean DEFAULT_MUTABLE_OBJECT_SAFE_MODE = true;
-
private final Map<Operator<?>, List<?>> intermediateResults;
private final Map<String, Accumulator<?, ?>> accumulators;
@@ -109,11 +106,6 @@ public class CollectionExecutor {
public JobExecutionResult execute(Plan program) throws Exception {
long startTime = System.currentTimeMillis();
-
- JobID jobID = program.getJobId();
- if (jobID == null) {
- jobID = new JobID();
- }
initCache(program.getCachedFiles());
Collection<? extends GenericDataSinkBase<?>> sinks = program.getDataSinks();
@@ -194,7 +186,7 @@ public class CollectionExecutor {
TaskInfo taskInfo = new TaskInfo(typedSink.getName(), 0, 1, 0);
RuntimeUDFContext ctx;
- MetricGroup metrics = NonRegisteringMetricsGroup.get();
+ MetricGroup metrics = new UnregisteredMetricsGroup();
if (RichOutputFormat.class.isAssignableFrom(typedSink.getUserCodeWrapper().getUserCodeClass())) {
ctx = superStep == 0 ? new RuntimeUDFContext(taskInfo, classLoader, executionConfig, cachedFiles, accumulators, metrics) :
@@ -215,7 +207,7 @@ public class CollectionExecutor {
RuntimeUDFContext ctx;
- MetricGroup metrics = NonRegisteringMetricsGroup.get();
+ MetricGroup metrics = new UnregisteredMetricsGroup();
if (RichInputFormat.class.isAssignableFrom(typedSource.getUserCodeWrapper().getUserCodeClass())) {
ctx = superStep == 0 ? new RuntimeUDFContext(taskInfo, classLoader, executionConfig, cachedFiles, accumulators, metrics) :
new IterationRuntimeUDFContext(taskInfo, classLoader, executionConfig, cachedFiles, accumulators, metrics);
@@ -241,7 +233,7 @@ public class CollectionExecutor {
TaskInfo taskInfo = new TaskInfo(typedOp.getName(), 0, 1, 0);
RuntimeUDFContext ctx;
- MetricGroup metrics = NonRegisteringMetricsGroup.get();
+ MetricGroup metrics = new UnregisteredMetricsGroup();
if (RichFunction.class.isAssignableFrom(typedOp.getUserCodeWrapper().getUserCodeClass())) {
ctx = superStep == 0 ? new RuntimeUDFContext(taskInfo, classLoader, executionConfig, cachedFiles, accumulators, metrics) :
new IterationRuntimeUDFContext(taskInfo, classLoader, executionConfig, cachedFiles, accumulators, metrics);
@@ -253,10 +245,8 @@ public class CollectionExecutor {
} else {
ctx = null;
}
-
- List<OUT> result = typedOp.executeOnCollections(inputData, ctx, executionConfig);
-
- return result;
+
+ return typedOp.executeOnCollections(inputData, ctx, executionConfig);
}
private <IN1, IN2, OUT> List<OUT> executeBinaryOperator(DualInputOperator<?, ?, ?, ?> operator, int superStep) throws Exception {
@@ -283,7 +273,7 @@ public class CollectionExecutor {
TaskInfo taskInfo = new TaskInfo(typedOp.getName(), 0, 1, 0);
RuntimeUDFContext ctx;
- MetricGroup metrics = NonRegisteringMetricsGroup.get();
+ MetricGroup metrics = new UnregisteredMetricsGroup();
if (RichFunction.class.isAssignableFrom(typedOp.getUserCodeWrapper().getUserCodeClass())) {
ctx = superStep == 0 ? new RuntimeUDFContext(taskInfo, classLoader, executionConfig, cachedFiles, accumulators, metrics) :
@@ -296,10 +286,8 @@ public class CollectionExecutor {
} else {
ctx = null;
}
-
- List<OUT> result = typedOp.executeOnCollections(inputData1, inputData2, ctx, executionConfig);
-
- return result;
+
+ return typedOp.executeOnCollections(inputData1, inputData2, ctx, executionConfig);
}
@SuppressWarnings("unchecked")
@@ -448,7 +436,7 @@ public class CollectionExecutor {
solutionMap.put(wrapper, delta);
}
- currentWorkset = (List<?>) execute(iteration.getNextWorkset(), superstep);
+ currentWorkset = execute(iteration.getNextWorkset(), superstep);
if (currentWorkset.isEmpty()) {
break;
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/Counter.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/Counter.java b/flink-core/src/main/java/org/apache/flink/metrics/Counter.java
index 201a613..acc37cf 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/Counter.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/Counter.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 org.apache.flink.annotation.PublicEvolving;
@@ -24,9 +25,9 @@ import org.apache.flink.annotation.PublicEvolving;
*/
@PublicEvolving
public final class Counter implements Metric {
-
+
private long count;
-
+
/**
* Increment the current count by 1.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/Gauge.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/Gauge.java b/flink-core/src/main/java/org/apache/flink/metrics/Gauge.java
index cca105e..aad8deb 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/Gauge.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/Gauge.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 org.apache.flink.annotation.PublicEvolving;
@@ -24,7 +25,7 @@ import org.apache.flink.annotation.PublicEvolving;
*/
@PublicEvolving
public abstract class Gauge<T> implements Metric {
-
+
/**
* Calculates and returns the measured value.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/Metric.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/Metric.java b/flink-core/src/main/java/org/apache/flink/metrics/Metric.java
index 11cfcc6..8054de0 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/Metric.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/Metric.java
@@ -15,12 +15,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics;
import org.apache.flink.annotation.PublicEvolving;
/**
- * Common interface for all metrics.
+ * Common super interface for all metrics.
*/
@PublicEvolving
public interface Metric {
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/MetricRegistry.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/MetricRegistry.java b/flink-core/src/main/java/org/apache/flink/metrics/MetricRegistry.java
index 7e06217..e5d3477 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/MetricRegistry.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/MetricRegistry.java
@@ -21,7 +21,8 @@ package org.apache.flink.metrics;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.groups.AbstractMetricGroup;
-import org.apache.flink.metrics.groups.Scope;
+import org.apache.flink.metrics.groups.scope.ScopeFormat;
+import org.apache.flink.metrics.groups.scope.ScopeFormats;
import org.apache.flink.metrics.reporter.JMXReporter;
import org.apache.flink.metrics.reporter.MetricReporter;
import org.apache.flink.metrics.reporter.Scheduled;
@@ -32,11 +33,6 @@ import org.slf4j.LoggerFactory;
import java.util.TimerTask;
import java.util.concurrent.TimeUnit;
-import static org.apache.flink.metrics.groups.JobMetricGroup.DEFAULT_SCOPE_JOB;
-import static org.apache.flink.metrics.groups.OperatorMetricGroup.DEFAULT_SCOPE_OPERATOR;
-import static org.apache.flink.metrics.groups.TaskManagerMetricGroup.DEFAULT_SCOPE_TM;
-import static org.apache.flink.metrics.groups.TaskMetricGroup.DEFAULT_SCOPE_TASK;
-
/**
* A MetricRegistry keeps track of all registered {@link Metric Metrics}. It serves as the
* connection between {@link MetricGroup MetricGroups} and {@link MetricReporter MetricReporters}.
@@ -52,10 +48,10 @@ public class MetricRegistry {
public static final String KEY_METRICS_REPORTER_ARGUMENTS = "metrics.reporter.arguments";
public static final String KEY_METRICS_REPORTER_INTERVAL = "metrics.reporter.interval";
- public static final String KEY_METRICS_SCOPE_NAMING_TM = "metrics.scope.tm";
- public static final String KEY_METRICS_SCOPE_NAMING_JOB = "metrics.scope.job";
- public static final String KEY_METRICS_SCOPE_NAMING_TASK = "metrics.scope.task";
- public static final String KEY_METRICS_SCOPE_NAMING_OPERATOR = "metrics.scope.operator";
+ public static final String KEY_METRICS_SCOPE_NAMING_TM = "metrics.scopeName.tm";
+ public static final String KEY_METRICS_SCOPE_NAMING_JOB = "metrics.scopeName.job";
+ public static final String KEY_METRICS_SCOPE_NAMING_TASK = "metrics.scopeName.task";
+ public static final String KEY_METRICS_SCOPE_NAMING_OPERATOR = "metrics.scopeName.operator";
// ------------------------------------------------------------------------
// configuration keys
@@ -66,23 +62,22 @@ public class MetricRegistry {
private final MetricReporter reporter;
private final java.util.Timer timer;
- private final Scope.ScopeFormat scopeConfig;
+ private final ScopeFormats scopeFormats;
/**
* Creates a new MetricRegistry and starts the configured reporter.
*/
public MetricRegistry(Configuration config) {
- // first parse the scope formats, these are needed for all reporters
-
- Scope.ScopeFormat scopeFormat;
+ // first parse the scopeName formats, these are needed for all reporters
+ ScopeFormats scopeFormats;
try {
- scopeFormat = createScopeConfig(config);
+ scopeFormats = createScopeConfig(config);
}
catch (Exception e) {
- scopeFormat = createScopeConfig(new Configuration());
- LOG.warn("Failed to parse scope format, using default scope formats", e);
+ LOG.warn("Failed to parse scopeName format, using default scopeName formats", e);
+ scopeFormats = new ScopeFormats();
}
- this.scopeConfig = scopeFormat;
+ this.scopeFormats = scopeFormats;
// second, instantiate any custom configured reporters
@@ -157,8 +152,8 @@ public class MetricRegistry {
}
}
- public Scope.ScopeFormat getScopeConfig() {
- return this.scopeConfig;
+ public ScopeFormats getScopeFormats() {
+ return scopeFormats;
}
// ------------------------------------------------------------------------
@@ -166,35 +161,36 @@ public class MetricRegistry {
// ------------------------------------------------------------------------
/**
- * Registers a new {@link org.apache.flink.metrics.Metric} with this registry.
+ * Registers a new {@link Metric} with this registry.
*
- * @param metric metric to register
- * @param name name of the metric
- * @param parent group that contains the metric
+ * @param metric the metric that was added
+ * @param metricName the name of the metric
+ * @param group the group that contains the metric
*/
- public void register(Metric metric, String name, AbstractMetricGroup parent) {
- String metricName = reporter.generateName(name, parent.generateScope());
- this.reporter.notifyOfAddedMetric(metric, metricName);
+ public void register(Metric metric, String metricName, AbstractMetricGroup group) {
+ if (reporter != null) {
+ reporter.notifyOfAddedMetric(metric, metricName, group);
+ }
}
/**
* Un-registers the given {@link org.apache.flink.metrics.Metric} with this registry.
*
- * @param metric metric to un-register
- * @param name name of the metric
- * @param parent group that contains the metric
+ * @param metric the metric that should be removed
+ * @param metricName the name of the metric
+ * @param group the group that contains the metric
*/
- public void unregister(Metric metric, String name, AbstractMetricGroup parent) {
- String metricName = reporter.generateName(name, parent.generateScope());
-
- this.reporter.notifyOfRemovedMetric(metric, metricName);
+ public void unregister(Metric metric, String metricName, AbstractMetricGroup group) {
+ if (reporter != null) {
+ reporter.notifyOfRemovedMetric(metric, metricName, group);
+ }
}
// ------------------------------------------------------------------------
// Utilities
// ------------------------------------------------------------------------
- private static Configuration createReporterConfig(Configuration config, TimeUnit timeunit, long period) {
+ static Configuration createReporterConfig(Configuration config, TimeUnit timeunit, long period) {
Configuration reporterConfig = new Configuration();
reporterConfig.setLong("period", period);
reporterConfig.setString("timeunit", timeunit.name());
@@ -208,19 +204,17 @@ public class MetricRegistry {
return reporterConfig;
}
- private static Scope.ScopeFormat createScopeConfig(Configuration config) {
- String tmFormat = config.getString(KEY_METRICS_SCOPE_NAMING_TM, DEFAULT_SCOPE_TM);
- String jobFormat = config.getString(KEY_METRICS_SCOPE_NAMING_JOB, DEFAULT_SCOPE_JOB);
- String taskFormat = config.getString(KEY_METRICS_SCOPE_NAMING_TASK, DEFAULT_SCOPE_TASK);
- String operatorFormat = config.getString(KEY_METRICS_SCOPE_NAMING_OPERATOR, DEFAULT_SCOPE_OPERATOR);
-
-
- Scope.ScopeFormat format = new Scope.ScopeFormat();
- format.setTaskManagerFormat(tmFormat);
- format.setJobFormat(jobFormat);
- format.setTaskFormat(taskFormat);
- format.setOperatorFormat(operatorFormat);
- return format;
+ static ScopeFormats createScopeConfig(Configuration config) {
+ String tmFormat = config.getString(
+ KEY_METRICS_SCOPE_NAMING_TM, ScopeFormat.DEFAULT_SCOPE_TASKMANAGER_GROUP);
+ String jobFormat = config.getString(
+ KEY_METRICS_SCOPE_NAMING_JOB, ScopeFormat.DEFAULT_SCOPE_TASKMANAGER_JOB_GROUP);
+ String taskFormat = config.getString(
+ KEY_METRICS_SCOPE_NAMING_TASK, ScopeFormat.DEFAULT_SCOPE_TASK_GROUP);
+ String operatorFormat = config.getString(
+ KEY_METRICS_SCOPE_NAMING_OPERATOR, ScopeFormat.DEFAULT_SCOPE_OPERATOR_GROUP);
+
+ return new ScopeFormats(tmFormat, jobFormat, taskFormat, operatorFormat);
}
// ------------------------------------------------------------------------
@@ -236,7 +230,7 @@ public class MetricRegistry {
* which acts as a fail-safe to stop the timer thread and prevents resource leaks.
*/
private static final class ReporterTask extends TimerTask {
-
+
private final Scheduled reporter;
private ReporterTask(Scheduled reporter) {
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/groups/AbstractMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/groups/AbstractMetricGroup.java b/flink-core/src/main/java/org/apache/flink/metrics/groups/AbstractMetricGroup.java
index cad241d..032fa04 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/groups/AbstractMetricGroup.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/groups/AbstractMetricGroup.java
@@ -25,14 +25,12 @@ import org.apache.flink.metrics.Metric;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.MetricRegistry;
+import org.apache.flink.metrics.groups.scope.ScopeFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -55,31 +53,58 @@ public abstract class AbstractMetricGroup implements MetricGroup {
/** shared logger */
private static final Logger LOG = LoggerFactory.getLogger(MetricGroup.class);
- private static final String METRIC_NAME_REGEX = "[a-zA-Z0-9]*";
-
- /** The pattern that metric and group names have to match */
- private static final Pattern METRIC_NAME_PATTERN = Pattern.compile(METRIC_NAME_REGEX);
-
// ------------------------------------------------------------------------
/** The registry that this metrics group belongs to */
protected final MetricRegistry registry;
/** All metrics that are directly contained in this group */
- protected final Map<String, Metric> metrics = new HashMap<>();
+ private final Map<String, Metric> metrics = new HashMap<>();
/** All metric subgroups of this group */
- protected final Map<String, MetricGroup> groups = new HashMap<>();
+ private final Map<String, MetricGroup> groups = new HashMap<>();
+
+ /** The metrics scope represented by this group.
+ * For example ["host-7", "taskmanager-2", "window_word_count", "my-mapper" ]. */
+ private final String[] scopeComponents;
+
+ /** The metrics scope represented by this group, as a concatenated string, lazily computed.
+ * For example: "host-7.taskmanager-2.window_word_count.my-mapper" */
+ private String scopeString;
/** Flag indicating whether this group has been closed */
private volatile boolean closed;
// ------------------------------------------------------------------------
-
- public AbstractMetricGroup(MetricRegistry registry) {
+
+ public AbstractMetricGroup(MetricRegistry registry, String[] scope) {
this.registry = checkNotNull(registry);
+ this.scopeComponents = checkNotNull(scope);
}
+ /**
+ * Gets the scope as an array of the scope components, for example
+ * {@code ["host-7", "taskmanager-2", "window_word_count", "my-mapper"]}
+ *
+ * @see #getScopeString()
+ */
+ public String[] getScopeComponents() {
+ return scopeComponents;
+ }
+
+ /**
+ * Gets the scope as a single delimited string, for example
+ * {@code "host-7.taskmanager-2.window_word_count.my-mapper"}
+ *
+ * @see #getScopeComponents()
+ */
+ public String getScopeString() {
+ if (scopeString == null) {
+ scopeString = ScopeFormat.concat(scopeComponents);
+ }
+ return scopeString;
+ }
+
// ------------------------------------------------------------------------
// Closing
// ------------------------------------------------------------------------
@@ -111,25 +136,6 @@ public abstract class AbstractMetricGroup implements MetricGroup {
}
// -----------------------------------------------------------------------------------------------------------------
- // Scope
- // -----------------------------------------------------------------------------------------------------------------
-
- /**
- * Generates the full scope based on the default/configured format that applies to all metrics within this group.
- *
- * @return generated scope
- */
- public abstract List<String> generateScope();
-
- /**
- * Generates the full scope based on the given format that applies to all metrics within this group.
- *
- * @param format format string
- * @return generated scope
- */
- public abstract List<String> generateScope(Scope.ScopeFormat format);
-
- // -----------------------------------------------------------------------------------------------------------------
// Metrics
// -----------------------------------------------------------------------------------------------------------------
@@ -164,11 +170,8 @@ public abstract class AbstractMetricGroup implements MetricGroup {
* @param metric the metric to register
*/
protected void addMetric(String name, Metric metric) {
- Matcher nameMatcher = METRIC_NAME_PATTERN.matcher(name);
- if (!nameMatcher.matches()) {
- throw new IllegalArgumentException("Metric names may not contain special characters or spaces. " +
- "Allowed is: " + METRIC_NAME_REGEX);
- }
+ // early reject names that will later cause issues
+ checkAllowedCharacters(name);
// add the metric only if the group is still open
synchronized (this) {
@@ -185,7 +188,7 @@ public abstract class AbstractMetricGroup implements MetricGroup {
// we warn here, rather than failing, because metrics are tools that should not fail the
// program when used incorrectly
LOG.warn("Name collision: Adding a metric with the same name as a metric subgroup: '" +
- name + "'. Metric might not get properly reported. (" + generateScope() + ')');
+ name + "'. Metric might not get properly reported. (" + scopeString + ')');
}
registry.register(metric, name, this);
@@ -197,7 +200,7 @@ public abstract class AbstractMetricGroup implements MetricGroup {
// we warn here, rather than failing, because metrics are tools that should not fail the
// program when used incorrectly
LOG.warn("Name collision: Group already contains a Metric with the name '" +
- name + "'. Metric will not be reported. (" + generateScope() + ')');
+ name + "'. Metric will not be reported. (" + scopeString + ')');
}
}
}
@@ -221,7 +224,7 @@ public abstract class AbstractMetricGroup implements MetricGroup {
// program when used incorrectly
if (metrics.containsKey(name)) {
LOG.warn("Name collision: Adding a metric subgroup with the same name as an existing metric: '" +
- name + "'. Metric might not get properly reported. (" + generateScope() + ')');
+ name + "'. Metric might not get properly reported. (" + scopeString + ')');
}
MetricGroup newGroup = new GenericMetricGroup(registry, this, name);
@@ -243,4 +246,21 @@ public abstract class AbstractMetricGroup implements MetricGroup {
}
}
}
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ /**
+ * Fast implementation to check if a string has only alphanumeric characters.
+ * Compared to a regular expression, this is about an order of magnitude faster.
+ */
+ private static void checkAllowedCharacters(String name) {
+ for (int i = 0; i < name.length(); i++) {
+ final char c = name.charAt(i);
+ if (c < 0x30 || (c >= 0x3a && c <= 0x40) || (c > 0x5a && c <= 0x60) || c > 0x7a) {
+ throw new IllegalArgumentException("Metric names may only contain [a-zA-Z0-9].");
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/groups/ComponentMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/groups/ComponentMetricGroup.java b/flink-core/src/main/java/org/apache/flink/metrics/groups/ComponentMetricGroup.java
index c68cdc1..518d940 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/groups/ComponentMetricGroup.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/groups/ComponentMetricGroup.java
@@ -21,46 +21,33 @@ package org.apache.flink.metrics.groups;
import org.apache.flink.annotation.Internal;
import org.apache.flink.metrics.MetricRegistry;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import static org.apache.flink.metrics.groups.Scope.SCOPE_WILDCARD;
-
/**
* Abstract {@link org.apache.flink.metrics.MetricGroup} for system components (e.g.,
* TaskManager, Job, Task, Operator).
*
- * <p>The components metric groups contain functionality to define alias names for
- * the component identifier. For example, while Tasks are registered under a Task Attempt ID,
- * the metrics name scope may use the task name instead. Using these aliases makes
- * Metric scope names stable across jobs, recovery attempts, etc.
+ * <p>Usually, the scope of metrics is simply the hierarchy of the containing groups. For example
+ * the Metric {@code "MyMetric"} in group {@code "B"} nested in group {@code "A"} would have a
+ * fully scoped name of {@code "A.B.MyMetric"}, with {@code "A.B"} being the Metric's scope.
+ *
+ * <p>Component groups, however, have configurable scopes. This allow users to include or exclude
+ * certain identifiers from the scope. The scope for metrics belonging to the "Task"
+ * group could for example include the task attempt number (more fine grained identification), or
+ * exclude it (for continuity of the namespace across failure and recovery).
*/
@Internal
public abstract class ComponentMetricGroup extends AbstractMetricGroup {
- private final ComponentMetricGroup parent;
-
- private final String format;
-
- // Map: scope variable -> specific value
- protected final Map<String, String> formats;
-
- // ------------------------------------------------------------------------
-
/**
* Creates a new ComponentMetricGroup.
*
- * @param registry registry to register new metrics with
- * @param parentGroup parent group, may be null
- * @param scopeFormat default format string
+ * @param registry registry to register new metrics with
+ * @param scope the scope of the group
*/
- public ComponentMetricGroup(MetricRegistry registry, ComponentMetricGroup parentGroup, String scopeFormat) {
- super(registry);
- this.formats = new HashMap<>();
- this.parent = parentGroup;
- this.format = scopeFormat;
+ public ComponentMetricGroup(
+ MetricRegistry registry,
+ String[] scope) {
+
+ super(registry, scope);
}
/**
@@ -87,71 +74,5 @@ public abstract class ComponentMetricGroup extends AbstractMetricGroup {
// sub components
// ------------------------------------------------------------------------
- protected ComponentMetricGroup parent() {
- return parent;
- }
-
protected abstract Iterable<? extends ComponentMetricGroup> subComponents();
-
- // ------------------------------------------------------------------------
- // scope format
- // ------------------------------------------------------------------------
-
- protected abstract String getScopeFormat(Scope.ScopeFormat format);
-
- @Override
- public List<String> generateScope() {
- return generateScope(format);
- }
-
- @Override
- public List<String> generateScope(Scope.ScopeFormat format) {
- return generateScope(getScopeFormat(format));
- }
-
- private List<String> generateScope(String format) {
- String[] components = Scope.split(format);
-
- List<String> scope = new ArrayList<>();
- if (components[0].equals(SCOPE_WILDCARD)) {
- if (this.parent != null) {
- scope = this.parent.generateScope();
- }
- this.replaceFormats(components);
- addToList(scope, components, 1);
- } else {
- if (this.parent != null) {
- this.parent.replaceFormats(components);
- }
- this.replaceFormats(components);
- addToList(scope, components, 0);
- }
- return scope;
- }
-
- private void replaceFormats(String[] components) {
- if (this.parent != null) {
- this.parent.replaceFormats(components);
- }
- for (int x = 0; x < components.length; x++) {
- if (components[x].startsWith("<")) {
- if (this.formats.containsKey(components[x])) {
- components[x] = this.formats.get(components[x]);
- }
- }
- }
- }
-
- /**
- * Adds all elements from the given array, starting from the given index, to the given list.
- *
- * @param list destination
- * @param array source
- * @param startIndex array index to start from
- */
- private static void addToList(List<String> list, String[] array, int startIndex) {
- for (int x = startIndex; x < array.length; x++) {
- list.add(array[x]);
- }
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/groups/GenericMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/groups/GenericMetricGroup.java b/flink-core/src/main/java/org/apache/flink/metrics/groups/GenericMetricGroup.java
index eedb0fa..ddcd73b 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/groups/GenericMetricGroup.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/groups/GenericMetricGroup.java
@@ -15,44 +15,35 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics.groups;
import org.apache.flink.annotation.Internal;
import org.apache.flink.metrics.MetricRegistry;
-import java.util.List;
-
/**
- * A simple named {@link org.apache.flink.metrics.MetricGroup} with no special properties.
+ * A simple named {@link org.apache.flink.metrics.MetricGroup} that is used to hold
+ * subgroups of metrics.
*/
@Internal
public class GenericMetricGroup extends AbstractMetricGroup {
-
- private final AbstractMetricGroup parent;
-
- private final String name;
-
- protected GenericMetricGroup(MetricRegistry registry, AbstractMetricGroup parent, int name) {
- this(registry, parent, String.valueOf(name));
- }
-
- protected GenericMetricGroup(MetricRegistry registry, AbstractMetricGroup parent, String name) {
- super(registry);
- this.parent = parent;
- this.name = name;
- }
- @Override
- public List<String> generateScope() {
- List<String> scope = parent.generateScope();
- scope.add(name);
- return scope;
+ public GenericMetricGroup(MetricRegistry registry, AbstractMetricGroup parent, String name) {
+ super(registry, makeScopeComponents(parent, name));
}
- @Override
- public List<String> generateScope(Scope.ScopeFormat format) {
- List<String> scope = parent.generateScope(format);
- scope.add(name);
- return scope;
+ // ------------------------------------------------------------------------
+
+ private static String[] makeScopeComponents(AbstractMetricGroup parent, String name) {
+ if (parent != null) {
+ String[] parentComponents = parent.getScopeComponents();
+ if (parentComponents != null && parentComponents.length > 0) {
+ String[] parts = new String[parentComponents.length + 1];
+ System.arraycopy(parentComponents, 0, parts, 0, parentComponents.length);
+ parts[parts.length - 1] = name;
+ return parts;
+ }
+ }
+ return new String[] { name };
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/groups/IOMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/groups/IOMetricGroup.java b/flink-core/src/main/java/org/apache/flink/metrics/groups/IOMetricGroup.java
index dea5650..b34c844 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/groups/IOMetricGroup.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/groups/IOMetricGroup.java
@@ -15,19 +15,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics.groups;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.MetricRegistry;
-import java.util.List;
-
/**
* Special {@link org.apache.flink.metrics.MetricGroup} that contains shareable pre-defined IO-related metrics.
*/
public class IOMetricGroup extends AbstractMetricGroup {
-
- private final TaskMetricGroup parent;
private final Counter numBytesIn;
private final Counter numBytesOut;
@@ -35,37 +32,27 @@ public class IOMetricGroup extends AbstractMetricGroup {
private final Counter numRecordsOut;
public IOMetricGroup(MetricRegistry registry, TaskMetricGroup parent) {
- super(registry);
- this.parent = parent;
+ super(registry, parent.getScopeComponents());
+
this.numBytesIn = parent.counter("numBytesIn");
this.numBytesOut = parent.counter("numBytesOut");
this.numRecordsIn = parent.counter("numRecordsIn");
this.numRecordsOut = parent.counter("numRecordsOut");
}
- @Override
- public List<String> generateScope() {
- return parent.generateScope();
- }
-
- @Override
- public List<String> generateScope(Scope.ScopeFormat format) {
- return parent.generateScope(format);
- }
-
public Counter getBytesInCounter() {
- return this.numBytesIn;
+ return numBytesIn;
}
public Counter getBytesOutCounter() {
- return this.numBytesOut;
+ return numBytesOut;
}
public Counter getRecordsInCounter() {
- return this.numRecordsIn;
+ return numRecordsIn;
}
public Counter getRecordsOutCounter() {
- return this.numRecordsOut;
+ return numRecordsOut;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/groups/JobMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/groups/JobMetricGroup.java b/flink-core/src/main/java/org/apache/flink/metrics/groups/JobMetricGroup.java
index f4f634a..f816278 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/groups/JobMetricGroup.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/groups/JobMetricGroup.java
@@ -21,57 +21,92 @@ package org.apache.flink.metrics.groups;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.JobID;
import org.apache.flink.metrics.MetricRegistry;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.TaskManagerJobScopeFormat;
import org.apache.flink.util.AbstractID;
+import javax.annotation.Nullable;
import java.util.HashMap;
import java.util.Map;
-import static org.apache.flink.metrics.groups.TaskManagerMetricGroup.DEFAULT_SCOPE_TM;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
- * Special {@link org.apache.flink.metrics.MetricGroup} representing a Job.
+ * Special {@link org.apache.flink.metrics.MetricGroup} representing everything belonging to
+ * a specific job, running on the TaskManager.
*
- * <p>Contains extra logic for adding tasks.
+ * <p>Contains extra logic for adding Tasks ({@link TaskMetricGroup}).
*/
@Internal
public class JobMetricGroup extends ComponentMetricGroup {
- public static final String SCOPE_JOB_DESCRIPTOR = "job";
- public static final String SCOPE_JOB_ID = Scope.format("job_id");
- public static final String SCOPE_JOB_NAME = Scope.format("job_name");
- public static final String DEFAULT_SCOPE_JOB_COMPONENT = Scope.concat(SCOPE_JOB_NAME);
- public static final String DEFAULT_SCOPE_JOB = Scope.concat(DEFAULT_SCOPE_TM, DEFAULT_SCOPE_JOB_COMPONENT);
-
- // ------------------------------------------------------------------------
+ /** The metrics group that contains this group */
+ private final TaskManagerMetricGroup parent;
/** Map from execution attempt ID (task identifier) to task metrics */
private final Map<AbstractID, TaskMetricGroup> tasks = new HashMap<>();
-
+
+ /** The ID of the job represented by this metrics group */
private final JobID jobId;
+ /** The name of the job represented by this metrics group */
+ @Nullable
+ private final String jobName;
+
// ------------------------------------------------------------------------
- public JobMetricGroup(MetricRegistry registry, TaskManagerMetricGroup taskManager, JobID jobId, String jobName) {
- super(registry, taskManager, registry.getScopeConfig().getJobFormat());
+ public JobMetricGroup(
+ MetricRegistry registry,
+ TaskManagerMetricGroup parent,
+ JobID jobId,
+ @Nullable String jobName) {
+ this(registry, checkNotNull(parent), registry.getScopeFormats().getJobFormat(), jobId, jobName);
+ }
+
+ public JobMetricGroup(
+ MetricRegistry registry,
+ TaskManagerMetricGroup parent,
+ TaskManagerJobScopeFormat scopeFormat,
+ JobID jobId,
+ @Nullable String jobName) {
+
+ super(registry, scopeFormat.formatScope(parent, jobId, jobName));
+
+ this.parent = checkNotNull(parent);
this.jobId = checkNotNull(jobId);
- this.formats.put(SCOPE_JOB_ID, jobId.toString());
- this.formats.put(SCOPE_JOB_NAME, checkNotNull(jobName));
+ this.jobName = jobName;
+ }
+
+ public final TaskManagerMetricGroup parent() {
+ return parent;
+ }
+
+ public JobID jobId() {
+ return jobId;
+ }
+
+ @Nullable
+ public String jobName() {
+ return jobName;
}
// ------------------------------------------------------------------------
// adding / removing tasks
// ------------------------------------------------------------------------
- public TaskMetricGroup addTask(AbstractID vertexId, AbstractID executionId, int subtaskIndex, String name) {
- checkNotNull(vertexId);
+ public TaskMetricGroup addTask(
+ AbstractID vertexId,
+ AbstractID executionId,
+ String taskName,
+ int subtaskIndex,
+ int attemptNumber) {
+
checkNotNull(executionId);
- checkNotNull(name);
synchronized (this) {
if (!isClosed()) {
- TaskMetricGroup task = new TaskMetricGroup(registry, this, vertexId, executionId, subtaskIndex, name);
+ TaskMetricGroup task = new TaskMetricGroup(registry, this,
+ vertexId, executionId, taskName, subtaskIndex, attemptNumber);
tasks.put(executionId, task);
return task;
} else {
@@ -82,7 +117,7 @@ public class JobMetricGroup extends ComponentMetricGroup {
public void removeTaskMetricGroup(AbstractID executionId) {
checkNotNull(executionId);
-
+
boolean removeFromParent = false;
synchronized (this) {
if (!isClosed() && tasks.remove(executionId) != null && tasks.isEmpty()) {
@@ -91,25 +126,16 @@ public class JobMetricGroup extends ComponentMetricGroup {
close();
}
}
-
+
// IMPORTANT: removing from the parent must happen while holding the this group's lock,
// because it would violate the "first parent then subgroup" lock acquisition order
if (removeFromParent) {
- ((TaskManagerMetricGroup) parent()).removeJobMetricsGroup(jobId, this);
+ parent.removeJobMetricsGroup(jobId, this);
}
}
- // ------------------------------------------------------------------------
- // component group behavior
- // ------------------------------------------------------------------------
-
@Override
protected Iterable<? extends ComponentMetricGroup> subComponents() {
return tasks.values();
}
-
- @Override
- protected String getScopeFormat(Scope.ScopeFormat format) {
- return format.getJobFormat();
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/groups/NonRegisteringMetricsGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/groups/NonRegisteringMetricsGroup.java b/flink-core/src/main/java/org/apache/flink/metrics/groups/NonRegisteringMetricsGroup.java
deleted file mode 100644
index 1bfcfe3..0000000
--- a/flink-core/src/main/java/org/apache/flink/metrics/groups/NonRegisteringMetricsGroup.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * 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.flink.metrics.groups;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.metrics.Counter;
-import org.apache.flink.metrics.Gauge;
-import org.apache.flink.metrics.MetricGroup;
-
-/**
- * Metrics group that does not register any metrics.
- */
-@Internal
-public class NonRegisteringMetricsGroup implements MetricGroup {
-
- // ------------------------------------------------------------------------
- // singleton
- // ------------------------------------------------------------------------
-
- private static final NonRegisteringMetricsGroup INSTANCE = new NonRegisteringMetricsGroup();
-
- public static NonRegisteringMetricsGroup get() {
- return INSTANCE;
- }
-
- /** Private constructor to prevent instantiation */
- private NonRegisteringMetricsGroup() {}
-
- // ------------------------------------------------------------------------
- // metrics group
- // ------------------------------------------------------------------------
-
- @Override
- public void close() {}
-
- @Override
- public boolean isClosed() {
- return false;
- }
-
- @Override
- public Counter counter(int name) {
- return new Counter();
- }
-
- @Override
- public Counter counter(String name) {
- return new Counter();
- }
-
- @Override
- public <T> Gauge<T> gauge(int name, Gauge<T> gauge) {
- return gauge;
- }
-
- @Override
- public <T> Gauge<T> gauge(String name, Gauge<T> gauge) {
- return gauge;
- }
-
-
- @Override
- public MetricGroup addGroup(int name) {
- return addGroup(String.valueOf(name));
- }
-
- @Override
- public MetricGroup addGroup(String name) {
- return new NonRegisteringMetricsGroup();
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/groups/OperatorMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/groups/OperatorMetricGroup.java b/flink-core/src/main/java/org/apache/flink/metrics/groups/OperatorMetricGroup.java
index 390b55b..6db79ab 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/groups/OperatorMetricGroup.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/groups/OperatorMetricGroup.java
@@ -15,14 +15,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.metrics.groups;
import org.apache.flink.annotation.Internal;
import org.apache.flink.metrics.MetricRegistry;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.OperatorScopeFormat;
import java.util.Collections;
-import static org.apache.flink.metrics.groups.JobMetricGroup.DEFAULT_SCOPE_JOB;
+import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* Special {@link org.apache.flink.metrics.MetricGroup} representing an Operator.
@@ -30,26 +32,29 @@ import static org.apache.flink.metrics.groups.JobMetricGroup.DEFAULT_SCOPE_JOB;
@Internal
public class OperatorMetricGroup extends ComponentMetricGroup {
- public static final String SCOPE_OPERATOR_DESCRIPTOR = "operator";
- public static final String SCOPE_OPERATOR_NAME = Scope.format("operator_name");
- public static final String SCOPE_OPERATOR_SUBTASK_INDEX = Scope.format("subtask_index");
- public static final String DEFAULT_SCOPE_OPERATOR_COMPONENT = Scope.concat(SCOPE_OPERATOR_NAME, SCOPE_OPERATOR_SUBTASK_INDEX);
- public static final String DEFAULT_SCOPE_OPERATOR = Scope.concat(DEFAULT_SCOPE_JOB, DEFAULT_SCOPE_OPERATOR_COMPONENT);
-
- protected OperatorMetricGroup(MetricRegistry registry, TaskMetricGroup task, String name, int subTaskIndex) {
- super(registry, task, registry.getScopeConfig().getOperatorFormat());
+ /** The task metric group that contains this operator metric groups */
+ private final TaskMetricGroup parent;
- this.formats.put(SCOPE_OPERATOR_NAME, name);
- this.formats.put(SCOPE_OPERATOR_SUBTASK_INDEX, String.valueOf(subTaskIndex));
+ public OperatorMetricGroup(MetricRegistry registry, TaskMetricGroup parent, String operatorName) {
+ this(registry, parent, registry.getScopeFormats().getOperatorFormat(), operatorName);
}
- // ------------------------------------------------------------------------
+ public OperatorMetricGroup(
+ MetricRegistry registry,
+ TaskMetricGroup parent,
+ OperatorScopeFormat scopeFormat,
+ String operatorName) {
- @Override
- protected String getScopeFormat(Scope.ScopeFormat format) {
- return format.getOperatorFormat();
+ super(registry, scopeFormat.formatScope(parent, operatorName));
+ this.parent = checkNotNull(parent);
}
+ // ------------------------------------------------------------------------
+
+ public final TaskMetricGroup parent() {
+ return parent;
+ }
+
@Override
protected Iterable<? extends ComponentMetricGroup> subComponents() {
return Collections.emptyList();
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/groups/Scope.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/groups/Scope.java b/flink-core/src/main/java/org/apache/flink/metrics/groups/Scope.java
deleted file mode 100644
index 83013e2..0000000
--- a/flink-core/src/main/java/org/apache/flink/metrics/groups/Scope.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/*
- * 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.flink.metrics.groups;
-
-import org.apache.flink.annotation.Internal;
-
-/**
- * This class provides utility-functions for handling scopes.
- */
-@Internal
-public class Scope {
- public static final String SCOPE_WILDCARD = "*";
-
- private static final String SCOPE_PREFIX = "<";
- private static final String SCOPE_SUFFIX = ">";
- private static final String SCOPE_SPLIT = ".";
-
- private Scope() {
- }
-
- /**
- * Modifies the given string to resemble a scope variable.
- *
- * @param scope string to format
- * @return formatted string
- */
- public static String format(String scope) {
- return SCOPE_PREFIX + scope + SCOPE_SUFFIX;
- }
-
- /**
- * Joins the given components into a single scope.
- *
- * @param components components to join
- * @return joined scoped
- */
- public static String concat(String... components) {
- StringBuilder sb = new StringBuilder();
- sb.append(components[0]);
- for (int x = 1; x < components.length; x++) {
- sb.append(SCOPE_SPLIT);
- sb.append(components[x]);
- }
- return sb.toString();
- }
-
- /**
- * Splits the given scope into it's individual components.
- *
- * @param scope scope to split
- * @return array of components
- */
- public static String[] split(String scope) {
- return scope.split("\\" + SCOPE_SPLIT);
- }
-
- /**
- * Simple container for component scope format strings.
- */
- public static class ScopeFormat {
-
- private String operatorFormat = OperatorMetricGroup.DEFAULT_SCOPE_OPERATOR;
- private String taskFormat = TaskMetricGroup.DEFAULT_SCOPE_TASK;
- private String jobFormat = JobMetricGroup.DEFAULT_SCOPE_JOB;
- private String taskManagerFormat = TaskManagerMetricGroup.DEFAULT_SCOPE_TM;
-
-
- public ScopeFormat setOperatorFormat(String format) {
- this.operatorFormat = format;
- return this;
- }
-
- public ScopeFormat setTaskFormat(String format) {
- this.taskFormat = format;
- return this;
- }
-
- public ScopeFormat setJobFormat(String format) {
- this.jobFormat = format;
- return this;
- }
-
- public ScopeFormat setTaskManagerFormat(String format) {
- this.taskManagerFormat = format;
- return this;
- }
-
- public String getOperatorFormat() {
- return this.operatorFormat;
- }
-
- public String getTaskFormat() {
- return this.taskFormat;
- }
-
- public String getJobFormat() {
- return this.jobFormat;
- }
-
- public String getTaskManagerFormat() {
- return this.taskManagerFormat;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/groups/TaskManagerMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/groups/TaskManagerMetricGroup.java b/flink-core/src/main/java/org/apache/flink/metrics/groups/TaskManagerMetricGroup.java
index bfb9362..3cb3936 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/groups/TaskManagerMetricGroup.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/groups/TaskManagerMetricGroup.java
@@ -21,13 +21,12 @@ package org.apache.flink.metrics.groups;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.JobID;
import org.apache.flink.metrics.MetricRegistry;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.TaskManagerScopeFormat;
import org.apache.flink.util.AbstractID;
import java.util.HashMap;
import java.util.Map;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
/**
* Special {@link org.apache.flink.metrics.MetricGroup} representing a TaskManager.
*
@@ -37,22 +36,33 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
@Internal
public class TaskManagerMetricGroup extends ComponentMetricGroup {
- public static final String SCOPE_HOST_DESCRIPTOR = "host";
- public static final String SCOPE_TM_DESCRIPTOR = "taskmanager";
- public static final String SCOPE_TM_HOST = Scope.format("host");
- public static final String SCOPE_TM_ID = Scope.format("tm_id");
- public static final String DEFAULT_SCOPE_TM_COMPONENT = Scope.concat(SCOPE_TM_HOST, "taskmanager", SCOPE_TM_ID);
- public static final String DEFAULT_SCOPE_TM = DEFAULT_SCOPE_TM_COMPONENT;
-
- // ------------------------------------------------------------------------
-
private final Map<JobID, JobMetricGroup> jobs = new HashMap<>();
- public TaskManagerMetricGroup(MetricRegistry registry, String host, String taskManagerId) {
- super(registry, null, registry.getScopeConfig().getTaskManagerFormat());
+ private final String hostname;
+
+ private final String taskManagerId;
- this.formats.put(SCOPE_TM_HOST, checkNotNull(host));
- this.formats.put(SCOPE_TM_ID, checkNotNull(taskManagerId));
+
+ public TaskManagerMetricGroup(MetricRegistry registry, String hostname, String taskManagerId) {
+ this(registry, registry.getScopeFormats().getTaskManagerFormat(), hostname, taskManagerId);
+ }
+
+ public TaskManagerMetricGroup(
+ MetricRegistry registry,
+ TaskManagerScopeFormat scopeFormat,
+ String hostname, String taskManagerId) {
+
+ super(registry, scopeFormat.formatScope(hostname, taskManagerId));
+ this.hostname = hostname;
+ this.taskManagerId = taskManagerId;
+ }
+
+ public String hostname() {
+ return hostname;
+ }
+
+ public String taskManagerId() {
+ return taskManagerId;
}
// ------------------------------------------------------------------------
@@ -64,9 +74,10 @@ public class TaskManagerMetricGroup extends ComponentMetricGroup {
String jobName,
AbstractID vertexID,
AbstractID executionId,
+ String taskName,
int subtaskIndex,
- String taskName) {
-
+ int attemptNumber) {
+
// we cannot strictly lock both our map modification and the job group modification
// because it might lead to a deadlock
while (true) {
@@ -80,28 +91,30 @@ public class TaskManagerMetricGroup extends ComponentMetricGroup {
jobs.put(jobId, currentJobGroup);
}
}
-
+
// try to add another task. this may fail if we found a pre-existing job metrics
// group and it is closed concurrently
- TaskMetricGroup taskGroup = currentJobGroup.addTask(vertexID, executionId, subtaskIndex, taskName);
+ TaskMetricGroup taskGroup = currentJobGroup.addTask(
+ vertexID, executionId, taskName, subtaskIndex, attemptNumber);
+
if (taskGroup != null) {
// successfully added the next task
return taskGroup;
}
-
+
// else fall through the loop
}
}
-
+
public void removeJobMetricsGroup(JobID jobId, JobMetricGroup group) {
if (jobId == null || group == null || !group.isClosed()) {
return;
}
-
+
synchronized (this) {
// optimistically remove the currently contained group, and check later if it was correct
JobMetricGroup containedGroup = jobs.remove(jobId);
-
+
// check if another group was actually contained, and restore that one
if (containedGroup != null && containedGroup != group) {
jobs.put(jobId, containedGroup);
@@ -113,15 +126,6 @@ public class TaskManagerMetricGroup extends ComponentMetricGroup {
return jobs.size();
}
- // ------------------------------------------------------------------------
- // component group behavior
- // ------------------------------------------------------------------------
-
- @Override
- protected String getScopeFormat(Scope.ScopeFormat format) {
- return format.getTaskManagerFormat();
- }
-
@Override
protected Iterable<? extends ComponentMetricGroup> subComponents() {
return jobs.values();
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/groups/TaskMetricGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/groups/TaskMetricGroup.java b/flink-core/src/main/java/org/apache/flink/metrics/groups/TaskMetricGroup.java
index 316c84f..784578b 100644
--- a/flink-core/src/main/java/org/apache/flink/metrics/groups/TaskMetricGroup.java
+++ b/flink-core/src/main/java/org/apache/flink/metrics/groups/TaskMetricGroup.java
@@ -20,12 +20,13 @@ package org.apache.flink.metrics.groups;
import org.apache.flink.annotation.Internal;
import org.apache.flink.metrics.MetricRegistry;
+import org.apache.flink.metrics.groups.scope.ScopeFormat.TaskScopeFormat;
import org.apache.flink.util.AbstractID;
+import javax.annotation.Nullable;
import java.util.HashMap;
import java.util.Map;
-import static org.apache.flink.metrics.groups.JobMetricGroup.DEFAULT_SCOPE_JOB;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -35,47 +36,109 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*/
@Internal
public class TaskMetricGroup extends ComponentMetricGroup {
-
- public static final String SCOPE_TASK_DESCRIPTOR = "task";
- public static final String SCOPE_TASK_ID = Scope.format("task_id");
- public static final String SCOPE_TASK_NAME = Scope.format("task_name");
- public static final String SCOPE_TASK_ATTEMPT = Scope.format("task_attempt");
- public static final String SCOPE_TASK_SUBTASK_INDEX = Scope.format("subtask_index");
- public static final String DEFAULT_SCOPE_TASK_COMPONENT = SCOPE_TASK_NAME;
- public static final String DEFAULT_SCOPE_TASK = Scope.concat(DEFAULT_SCOPE_JOB, DEFAULT_SCOPE_TASK_COMPONENT);
+ /** The job metrics group containing this task metrics group */
+ private final JobMetricGroup parent;
private final Map<String, OperatorMetricGroup> operators = new HashMap<>();
private final IOMetricGroup ioMetrics;
-
+
+ /** The execution Id uniquely identifying the executed task represented by this metrics group */
private final AbstractID executionId;
+
+ @Nullable
+ private final AbstractID vertexId;
+ @Nullable
+ private final String taskName;
+
private final int subtaskIndex;
- protected TaskMetricGroup(
+ private final int attemptNumber;
+
+ // ------------------------------------------------------------------------
+
+ public TaskMetricGroup(
MetricRegistry registry,
JobMetricGroup parent,
- AbstractID taskId,
+ @Nullable AbstractID vertexId,
AbstractID executionId,
+ @Nullable String taskName,
int subtaskIndex,
- String name) {
+ int attemptNumber) {
+
+ this(registry, parent, registry.getScopeFormats().getTaskFormat(),
+ vertexId, executionId, taskName, subtaskIndex, attemptNumber);
+ }
- super(registry, parent, registry.getScopeConfig().getTaskFormat());
+ public TaskMetricGroup(
+ MetricRegistry registry,
+ JobMetricGroup parent,
+ TaskScopeFormat scopeFormat,
+ @Nullable AbstractID vertexId,
+ AbstractID executionId,
+ @Nullable String taskName,
+ int subtaskIndex,
+ int attemptNumber) {
- this.executionId = executionId;
+ super(registry, scopeFormat.formatScope(
+ parent, vertexId, executionId, taskName, subtaskIndex, attemptNumber));
+
+ this.parent = checkNotNull(parent);
+ this.executionId = checkNotNull(executionId);
+ this.vertexId = vertexId;
+ this.taskName = taskName;
this.subtaskIndex = subtaskIndex;
-
- this.formats.put(SCOPE_TASK_ID, taskId.toString());
- this.formats.put(SCOPE_TASK_ATTEMPT, executionId.toString());
- this.formats.put(SCOPE_TASK_NAME, checkNotNull(name));
- this.formats.put(SCOPE_TASK_SUBTASK_INDEX, String.valueOf(subtaskIndex));
+ this.attemptNumber = attemptNumber;
this.ioMetrics = new IOMetricGroup(registry, this);
}
+ // ------------------------------------------------------------------------
+ // properties
+ // ------------------------------------------------------------------------
+
+ public final JobMetricGroup parent() {
+ return parent;
+ }
+
+ public AbstractID executionId() {
+ return executionId;
+ }
+
+ @Nullable
+ public AbstractID vertexId() {
+ return vertexId;
+ }
+
+ @Nullable
+ public String taskName() {
+ return taskName;
+ }
+
+ public int subtaskIndex() {
+ return subtaskIndex;
+ }
+
+ public int attemptNumber() {
+ return attemptNumber;
+ }
+
+ /**
+ * Returns the IOMetricGroup for this task.
+ *
+ * @return IOMetricGroup for this task.
+ */
+ public IOMetricGroup getIOMetricGroup() {
+ return ioMetrics;
+ }
+
+ // ------------------------------------------------------------------------
+ // operators and cleanup
+ // ------------------------------------------------------------------------
public OperatorMetricGroup addOperator(String name) {
- OperatorMetricGroup operator = new OperatorMetricGroup(this.registry, this, name, this.subtaskIndex);
+ OperatorMetricGroup operator = new OperatorMetricGroup(this.registry, this, name);
synchronized (this) {
OperatorMetricGroup previous = operators.put(name, operator);
@@ -93,27 +156,10 @@ public class TaskMetricGroup extends ComponentMetricGroup {
@Override
public void close() {
super.close();
- parent().removeTaskMetricGroup(executionId);
- }
-
- /**
- * Returns the IOMetricGroup for this task.
- *
- * @return IOMetricGroup for this task.
- */
- public IOMetricGroup getIOMetricGroup() {
- return this.ioMetrics;
- }
-
- @Override
- protected JobMetricGroup parent() {
- return (JobMetricGroup) super.parent();
+ parent.removeTaskMetricGroup(executionId);
}
- @Override
- protected String getScopeFormat(Scope.ScopeFormat format) {
- return format.getTaskFormat();
- }
+ // ------------------------------------------------------------------------
@Override
protected Iterable<? extends ComponentMetricGroup> subComponents() {
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/groups/UnregisteredMetricsGroup.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/groups/UnregisteredMetricsGroup.java b/flink-core/src/main/java/org/apache/flink/metrics/groups/UnregisteredMetricsGroup.java
new file mode 100644
index 0000000..961bcce
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/metrics/groups/UnregisteredMetricsGroup.java
@@ -0,0 +1,73 @@
+/*
+ * 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.flink.metrics.groups;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.MetricGroup;
+
+/**
+ * A special {@link MetricGroup} that does not register any metrics at the metrics registry
+ * and any reporters.
+ *
+ * <p>This metrics group appears always closed ({@link #isClosed()}).
+ */
+@Internal
+public class UnregisteredMetricsGroup implements MetricGroup {
+
+ @Override
+ public void close() {}
+
+ @Override
+ public boolean isClosed() {
+ return true;
+ }
+
+ @Override
+ public Counter counter(int name) {
+ return new Counter();
+ }
+
+ @Override
+ public Counter counter(String name) {
+ return new Counter();
+ }
+
+ @Override
+ public <T> Gauge<T> gauge(int name, Gauge<T> gauge) {
+ return gauge;
+ }
+
+ @Override
+ public <T> Gauge<T> gauge(String name, Gauge<T> gauge) {
+ return gauge;
+ }
+
+
+ @Override
+ public MetricGroup addGroup(int name) {
+ return addGroup(String.valueOf(name));
+ }
+
+ @Override
+ public MetricGroup addGroup(String name) {
+ return new UnregisteredMetricsGroup();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ad8375a/flink-core/src/main/java/org/apache/flink/metrics/groups/scope/ScopeFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/metrics/groups/scope/ScopeFormat.java b/flink-core/src/main/java/org/apache/flink/metrics/groups/scope/ScopeFormat.java
new file mode 100644
index 0000000..9637f65
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/metrics/groups/scope/ScopeFormat.java
@@ -0,0 +1,399 @@
+/*
+ * 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.flink.metrics.groups.scope;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.metrics.groups.JobMetricGroup;
+import org.apache.flink.metrics.groups.TaskManagerMetricGroup;
+import org.apache.flink.metrics.groups.TaskMetricGroup;
+import org.apache.flink.util.AbstractID;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This class represents the format after which the "scope" (or namespace) of the various
+ * component metric groups is built. Component metric groups
+ * ({@link org.apache.flink.metrics.groups.ComponentMetricGroup}), are for example
+ * "TaskManager", "Task", or "Operator".
+ *
+ * <p>User defined scope formats allow users to include or exclude
+ * certain identifiers from the scope. The scope for metrics belonging to the "Task"
+ * group could for example include the task attempt number (more fine grained identification), or
+ * exclude it (continuity of the namespace across failure and recovery).
+ */
+public abstract class ScopeFormat {
+
+ // ------------------------------------------------------------------------
+ // Scope Format Special Characters
+ // ------------------------------------------------------------------------
+
+ /**
+ * If the scope format starts with this character, then the parent components scope
+ * format will be used as a prefix.
+ *
+ * <p>For example, if the {@link JobMetricGroup} format is {@code "*.<job_name>"}, and the
+ * {@link TaskManagerMetricGroup} format is {@code "<host>"}, then the job's metrics
+ * will have {@code "<host>.<job_name>"} as their scope.
+ */
+ public static final String SCOPE_INHERIT_PARENT = "*";
+
+ public static final String SCOPE_SEPARATOR = ".";
+
+ private static final String SCOPE_VARIABLE_PREFIX = "<";
+ private static final String SCOPE_VARIABLE_SUFFIX = ">";
+
+ // ------------------------------------------------------------------------
+ // Scope Variables
+ // ------------------------------------------------------------------------
+
+ // ----- Task Manager ----
+
+ public static final String SCOPE_TASKMANAGER_HOST = asVariable("host");
+ public static final String SCOPE_TASKMANAGER_ID = asVariable("tm_id");
+
+ /** The default scope format of the TaskManager component: {@code "<host>.taskmanager.<tm_id>"} */
+ public static final String DEFAULT_SCOPE_TASKMANAGER_COMPONENT =
+ concat(SCOPE_TASKMANAGER_HOST, "taskmanager", SCOPE_TASKMANAGER_ID);
+
+ /** The default scope format of TaskManager metrics: {@code "<host>.taskmanager.<tm_id>"} */
+ public static final String DEFAULT_SCOPE_TASKMANAGER_GROUP = DEFAULT_SCOPE_TASKMANAGER_COMPONENT;
+
+ // ----- Job on Task Manager ----
+
+ public static final String SCOPE_JOB_ID = asVariable("job_id");
+ public static final String SCOPE_JOB_NAME = asVariable("job_name");
+
+ /** The default scope format for the job component: {@code "<job_name>"} */
+ public static final String DEFAULT_SCOPE_TASKMANAGER_JOB_COMPONENT = SCOPE_JOB_NAME;
+
+ /** The default scope format for all job metrics: {@code "<host>.taskmanager.<tm_id>.<job_name>"} */
+ public static final String DEFAULT_SCOPE_TASKMANAGER_JOB_GROUP =
+ concat(DEFAULT_SCOPE_TASKMANAGER_COMPONENT, DEFAULT_SCOPE_TASKMANAGER_JOB_COMPONENT);
+
+ // ----- Task ----
+
+ public static final String SCOPE_TASK_VERTEX_ID = asVariable("task_id");
+ public static final String SCOPE_TASK_NAME = asVariable("task_name");
+ public static final String SCOPE_TASK_ATTEMPT_ID = asVariable("task_attempt_id");
+ public static final String SCOPE_TASK_ATTEMPT_NUM = asVariable("task_attempt_num");
+ public static final String SCOPE_TASK_SUBTASK_INDEX = asVariable("subtask_index");
+
+ /** Default scope of the task component: {@code "<task_name>.<subtask_index>"} */
+ public static final String DEFAULT_SCOPE_TASK_COMPONENT =
+ concat(SCOPE_TASK_NAME, SCOPE_TASK_SUBTASK_INDEX);
+
+ /** The default scope format for all task metrics:
+ * {@code "<host>.taskmanager.<tm_id>.<job_name>.<task_name>.<subtask_index>"} */
+ public static final String DEFAULT_SCOPE_TASK_GROUP =
+ concat(DEFAULT_SCOPE_TASKMANAGER_JOB_GROUP, DEFAULT_SCOPE_TASK_COMPONENT);
+
+ // ----- Operator ----
+
+ public static final String SCOPE_OPERATOR_NAME = asVariable("operator_name");
+
+ /** The default scope added by the operator component: "<operator_name>.<subtask_index>" */
+ public static final String DEFAULT_SCOPE_OPERATOR_COMPONENT =
+ concat(SCOPE_OPERATOR_NAME, SCOPE_TASK_SUBTASK_INDEX);
+
+ /** The default scope format for all operator metrics:
+ * {@code "<host>.taskmanager.<tm_id>.<job_name>.<operator_name>.<subtask_index>"} */
+ public static final String DEFAULT_SCOPE_OPERATOR_GROUP =
+ concat(DEFAULT_SCOPE_TASKMANAGER_JOB_GROUP, DEFAULT_SCOPE_OPERATOR_COMPONENT);
+
+ // ------------------------------------------------------------------------
+ // Formatters form the individual component types
+ // ------------------------------------------------------------------------
+
+ /**
+ * The scope format for the {@link TaskManagerMetricGroup}.
+ */
+ public static class TaskManagerScopeFormat extends ScopeFormat {
+
+ public TaskManagerScopeFormat(String format) {
+ super(format, null, new String[] {
+ SCOPE_TASKMANAGER_HOST,
+ SCOPE_TASKMANAGER_ID
+ });
+ }
+
+ public String[] formatScope(String hostname, String taskManagerId) {
+ final String[] template = copyTemplate();
+ final String[] values = { hostname, taskManagerId };
+ return bindVariables(template, values);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * The scope format for the {@link JobMetricGroup}.
+ */
+ public static class TaskManagerJobScopeFormat extends ScopeFormat {
+
+ public TaskManagerJobScopeFormat(String format, TaskManagerScopeFormat parentFormat) {
+ super(format, parentFormat, new String[] {
+ SCOPE_TASKMANAGER_HOST,
+ SCOPE_TASKMANAGER_ID,
+ SCOPE_JOB_ID,
+ SCOPE_JOB_NAME
+ });
+ }
+
+ public String[] formatScope(TaskManagerMetricGroup parent, JobID jid, String jobName) {
+ final String[] template = copyTemplate();
+ final String[] values = {
+ parent.hostname(),
+ parent.taskManagerId(),
+ valueOrNull(jid),
+ valueOrNull(jobName)
+ };
+ return bindVariables(template, values);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * The scope format for the {@link TaskMetricGroup}.
+ */
+ public static class TaskScopeFormat extends ScopeFormat {
+
+ public TaskScopeFormat(String format, TaskManagerJobScopeFormat parentFormat) {
+ super(format, parentFormat, new String[] {
+ SCOPE_TASKMANAGER_HOST,
+ SCOPE_TASKMANAGER_ID,
+ SCOPE_JOB_ID,
+ SCOPE_JOB_NAME,
+ SCOPE_TASK_VERTEX_ID,
+ SCOPE_TASK_ATTEMPT_ID,
+ SCOPE_TASK_NAME,
+ SCOPE_TASK_SUBTASK_INDEX,
+ SCOPE_TASK_ATTEMPT_NUM
+ });
+ }
+
+ public String[] formatScope(
+ JobMetricGroup parent,
+ AbstractID vertexId, AbstractID attemptId,
+ String taskName, int subtask, int attemptNumber) {
+
+ final String[] template = copyTemplate();
+ final String[] values = {
+ parent.parent().hostname(),
+ parent.parent().taskManagerId(),
+ valueOrNull(parent.jobId()),
+ valueOrNull(parent.jobName()),
+ valueOrNull(vertexId),
+ valueOrNull(attemptId),
+ valueOrNull(taskName),
+ String.valueOf(subtask),
+ String.valueOf(attemptNumber)
+ };
+ return bindVariables(template, values);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+
+ /**
+ * The scope format for the {@link org.apache.flink.metrics.groups.OperatorMetricGroup}.
+ */
+ public static class OperatorScopeFormat extends ScopeFormat {
+
+ public OperatorScopeFormat(String format, TaskScopeFormat parentFormat) {
+ super(format, parentFormat, new String[] {
+ SCOPE_TASKMANAGER_HOST,
+ SCOPE_TASKMANAGER_ID,
+ SCOPE_JOB_ID,
+ SCOPE_JOB_NAME,
+ SCOPE_TASK_VERTEX_ID,
+ SCOPE_TASK_ATTEMPT_ID,
+ SCOPE_TASK_NAME,
+ SCOPE_TASK_SUBTASK_INDEX,
+ SCOPE_TASK_ATTEMPT_NUM,
+ SCOPE_OPERATOR_NAME
+ });
+ }
+
+ public String[] formatScope(TaskMetricGroup parent, String operatorName) {
+
+ final String[] template = copyTemplate();
+ final String[] values = {
+ parent.parent().parent().hostname(),
+ parent.parent().parent().taskManagerId(),
+ valueOrNull(parent.parent().jobId()),
+ valueOrNull(parent.parent().jobName()),
+ valueOrNull(parent.vertexId()),
+ valueOrNull(parent.executionId()),
+ valueOrNull(parent.taskName()),
+ String.valueOf(parent.subtaskIndex()),
+ String.valueOf(parent.attemptNumber()),
+ valueOrNull(operatorName)
+ };
+ return bindVariables(template, values);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Scope Format Base
+ // ------------------------------------------------------------------------
+
+ /** The scope format */
+ private final String format;
+
+ /** The format, split into components */
+ private final String[] template;
+
+ private final int[] templatePos;
+
+ private final int[] valuePos;
+
+ // ------------------------------------------------------------------------
+
+ protected ScopeFormat(String format, ScopeFormat parent, String[] variables) {
+ checkNotNull(format, "format is null");
+
+ final String[] rawComponents = format.split("\\" + SCOPE_SEPARATOR);
+
+ // compute the template array
+ final boolean parentAsPrefix = rawComponents.length > 0 && rawComponents[0].equals(SCOPE_INHERIT_PARENT);
+ if (parentAsPrefix) {
+ if (parent == null) {
+ throw new IllegalArgumentException("Component scope format requires parent prefix (starts with '"
+ + SCOPE_INHERIT_PARENT + "'), but this component has no parent (is root component).");
+ }
+
+ this.format = format.length() > 2 ? format.substring(2) : "<empty>";
+
+ String[] parentTemplate = parent.template;
+ int parentLen = parentTemplate.length;
+
+ this.template = new String[parentLen + rawComponents.length - 1];
+ System.arraycopy(parentTemplate, 0, this.template, 0, parentLen);
+ System.arraycopy(rawComponents, 1, this.template, parentLen, rawComponents.length - 1);
+ }
+ else {
+ this.format = format.isEmpty() ? "<empty>" : format;
+ this.template = rawComponents;
+ }
+
+ // --- compute the replacement matrix ---
+ // a bit of clumsy Java collections code ;-)
+
+ HashMap<String, Integer> varToValuePos = arrayToMap(variables);
+ List<Integer> templatePos = new ArrayList<>();
+ List<Integer> valuePos = new ArrayList<>();
+
+ for (int i = 0; i < template.length; i++) {
+ final String component = template[i];
+
+ // check if that is a variable
+ if (component != null && component.length() >= 3 &&
+ component.charAt(0) == '<' && component.charAt(component.length() - 1) == '>') {
+
+ // this is a variable
+ Integer replacementPos = varToValuePos.get(component);
+ if (replacementPos != null) {
+ templatePos.add(i);
+ valuePos.add(replacementPos);
+ }
+ }
+ }
+
+ this.templatePos = integerListToArray(templatePos);
+ this.valuePos = integerListToArray(valuePos);
+ }
+
+ // ------------------------------------------------------------------------
+
+ public String format() {
+ return format;
+ }
+
+ protected final String[] copyTemplate() {
+ String[] copy = new String[template.length];
+ System.arraycopy(template, 0, copy, 0, template.length);
+ return copy;
+ }
+
+ protected final String[] bindVariables(String[] template, String[] values) {
+ final int len = templatePos.length;
+ for (int i = 0; i < len; i++) {
+ template[templatePos[i]] = values[valuePos[i]];
+ }
+ return template;
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public String toString() {
+ return "ScopeFormat '" + format + '\'';
+ }
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ /**
+ * Formats the given string to resemble a scope variable.
+ *
+ * @param scope The string to format
+ * @return The formatted string
+ */
+ public static String asVariable(String scope) {
+ return SCOPE_VARIABLE_PREFIX + scope + SCOPE_VARIABLE_SUFFIX;
+ }
+
+ public static String concat(String... components) {
+ StringBuilder sb = new StringBuilder();
+ sb.append(components[0]);
+ for (int x = 1; x < components.length; x++) {
+ sb.append(SCOPE_SEPARATOR);
+ sb.append(components[x]);
+ }
+ return sb.toString();
+ }
+
+ static String valueOrNull(Object value) {
+ return (value == null || (value instanceof String && ((String) value).isEmpty())) ?
+ "null" : value.toString();
+ }
+
+ static HashMap<String, Integer> arrayToMap(String[] array) {
+ HashMap<String, Integer> map = new HashMap<>(array.length);
+ for (int i = 0; i < array.length; i++) {
+ map.put(array[i], i);
+ }
+ return map;
+ }
+
+ private static int[] integerListToArray(List<Integer> list) {
+ int[] array = new int[list.size()];
+ int pos = 0;
+ for (Integer i : list) {
+ array[pos++] = i;
+ }
+ return array;
+ }
+}
[7/7] flink git commit: [hotfix] [yarn tests] Fix deadlock between
YARN Session CLI tests and Surefire
Posted by se...@apache.org.
[hotfix] [yarn tests] Fix deadlock between YARN Session CLI tests and Surefire
The Surefire Plugin uses stdin to communicate with forked JVMs for tests.
The YARN Session CLI tests also try to read the stdin stream. The tests deadlock since
Surefire never releases the stdin locks during the lifetime of a test.
This change adds a parameter whether the YARN Session CLI should try to read
user console input, and sets this to false in the integration tests.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/da23ee38
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/da23ee38
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/da23ee38
Branch: refs/heads/master
Commit: da23ee38e5b36ddf26a6a5a807efbbbcbfe1d517
Parents: 6511557
Author: Stephan Ewen <se...@apache.org>
Authored: Mon May 30 19:49:25 2016 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon May 30 19:49:25 2016 +0200
----------------------------------------------------------------------
.../flink/client/FlinkYarnSessionCli.java | 24 +++++++++++++-------
.../flink/client/cli/CliFrontendParser.java | 2 +-
.../flink/yarn/FlinkYarnSessionCliTest.java | 2 +-
.../org/apache/flink/yarn/YarnTestBase.java | 2 +-
.../src/test/resources/log4j-test.properties | 3 ++-
5 files changed, 21 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/da23ee38/flink-clients/src/main/java/org/apache/flink/client/FlinkYarnSessionCli.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/FlinkYarnSessionCli.java b/flink-clients/src/main/java/org/apache/flink/client/FlinkYarnSessionCli.java
index 91f8df2..bb61ffb 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/FlinkYarnSessionCli.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/FlinkYarnSessionCli.java
@@ -82,6 +82,8 @@ public class FlinkYarnSessionCli {
*/
private final Option DYNAMIC_PROPERTIES;
+ private final boolean acceptInteractiveInput;
+
//------------------------------------ Internal fields -------------------------
private AbstractFlinkYarnCluster yarnCluster = null;
private boolean detachedMode = false;
@@ -89,7 +91,9 @@ public class FlinkYarnSessionCli {
/** Default yarn application name. */
private String defaultApplicationName = null;
- public FlinkYarnSessionCli(String shortPrefix, String longPrefix) {
+ public FlinkYarnSessionCli(String shortPrefix, String longPrefix, boolean acceptInteractiveInput) {
+ this.acceptInteractiveInput = acceptInteractiveInput;
+
QUERY = new Option(shortPrefix + "q", longPrefix + "query", false, "Display available YARN resources (memory, cores)");
QUEUE = new Option(shortPrefix + "qu", longPrefix + "queue", true, "Specify YARN queue.");
SHIP_PATH = new Option(shortPrefix + "t", longPrefix + "ship", true, "Ship files in the specified directory (t for transfer)");
@@ -292,7 +296,7 @@ public class FlinkYarnSessionCli {
propertiesFile.setReadable(true, false); // readable for all.
}
- public static void runInteractiveCli(AbstractFlinkYarnCluster yarnCluster) {
+ public static void runInteractiveCli(AbstractFlinkYarnCluster yarnCluster, boolean readConsoleInput) {
final String HELP = "Available commands:\n" +
"help - show these commands\n" +
"stop - stop the YARN session";
@@ -304,6 +308,8 @@ public class FlinkYarnSessionCli {
// ------------------ check if there are updates by the cluster -----------
GetClusterStatusResponse status = yarnCluster.getClusterStatus();
+ LOG.debug("Received status message: {}", status);
+
if (status != null && numTaskmanagers != status.numRegisteredTaskManagers()) {
System.err.println("Number of connected TaskManagers changed to " +
status.numRegisteredTaskManagers() + ". " +
@@ -324,15 +330,16 @@ public class FlinkYarnSessionCli {
yarnCluster.shutdown(true);
}
- // wait until CLIENT_POLLING_INTERVALL is over or the user entered something.
+ // wait until CLIENT_POLLING_INTERVAL is over or the user entered something.
long startTime = System.currentTimeMillis();
while ((System.currentTimeMillis() - startTime) < CLIENT_POLLING_INTERVALL * 1000
- && !in.ready()) {
+ && (!readConsoleInput || !in.ready()))
+ {
Thread.sleep(200);
}
//------------- handle interactive command by user. ----------------------
-
- if (in.ready()) {
+
+ if (readConsoleInput && in.ready()) {
String command = in.readLine();
switch (command) {
case "quit":
@@ -347,6 +354,7 @@ public class FlinkYarnSessionCli {
break;
}
}
+
if (yarnCluster.hasBeenStopped()) {
LOG.info("Stopping interactive command line interface, YARN cluster has been stopped.");
break;
@@ -358,7 +366,7 @@ public class FlinkYarnSessionCli {
}
public static void main(String[] args) {
- FlinkYarnSessionCli cli = new FlinkYarnSessionCli("", ""); // no prefix for the YARN session
+ FlinkYarnSessionCli cli = new FlinkYarnSessionCli("", "", true); // no prefix for the YARN session
System.exit(cli.run(args));
}
@@ -458,7 +466,7 @@ public class FlinkYarnSessionCli {
"Please also note that the temporary files of the YARN session in {} will not be removed.",
flinkYarnClient.getSessionFilesDir());
} else {
- runInteractiveCli(yarnCluster);
+ runInteractiveCli(yarnCluster, acceptInteractiveInput);
if (!yarnCluster.hasBeenStopped()) {
LOG.info("Command Line Interface requested session shutdown");
http://git-wip-us.apache.org/repos/asf/flink/blob/da23ee38/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
index 2ac53d2..b75952e 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
@@ -35,7 +35,7 @@ public class CliFrontendParser {
/** command line interface of the YARN session, with a special initialization here
* to prefix all options with y/yarn. */
- private static final FlinkYarnSessionCli yarnSessionCLi = new FlinkYarnSessionCli("y", "yarn");
+ private static final FlinkYarnSessionCli yarnSessionCLi = new FlinkYarnSessionCli("y", "yarn", true);
static final Option HELP_OPTION = new Option("h", "help", false,
http://git-wip-us.apache.org/repos/asf/flink/blob/da23ee38/flink-yarn-tests/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
index 30116af..7197b64 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
@@ -53,7 +53,7 @@ public class FlinkYarnSessionCliTest {
map.put("FLINK_CONF_DIR", tmpFolder.getAbsolutePath());
TestBaseUtils.setEnv(map);
Options options = new Options();
- FlinkYarnSessionCli cli = new FlinkYarnSessionCli("", "");
+ FlinkYarnSessionCli cli = new FlinkYarnSessionCli("", "", false);
cli.getYARNSessionCLIOptions(options);
CommandLineParser parser = new PosixParser();
http://git-wip-us.apache.org/repos/asf/flink/blob/da23ee38/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
index fc1e5bc..03ab647 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java
@@ -562,7 +562,7 @@ public abstract class YarnTestBase extends TestLogger {
public void run() {
switch(type) {
case YARN_SESSION:
- yCli = new FlinkYarnSessionCli("", "");
+ yCli = new FlinkYarnSessionCli("", "", false);
returnValue = yCli.run(args);
break;
case CLI_FRONTEND:
http://git-wip-us.apache.org/repos/asf/flink/blob/da23ee38/flink-yarn-tests/src/test/resources/log4j-test.properties
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/resources/log4j-test.properties b/flink-yarn-tests/src/test/resources/log4j-test.properties
index ebe0d37..e94ca26 100644
--- a/flink-yarn-tests/src/test/resources/log4j-test.properties
+++ b/flink-yarn-tests/src/test/resources/log4j-test.properties
@@ -30,6 +30,7 @@ log4j.logger.org.jboss.netty.channel.DefaultChannelPipeline=ERROR, console
log4j.logger.org.apache.flink.yarn.YARNSessionFIFOITCase=INFO
log4j.logger.org.apache.flink.yarn.YARNSessionCapacitySchedulerITCase=INFO
log4j.logger.org.apache.flink.yarn.YarnHighAvailability=INFO
-log4j.logger.org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch=INFO
+log4j.logger.org.apache.hadoop=OFF
log4j.logger.org.apache.flink.runtime.leaderelection=INFO
log4j.logger.org.apache.flink.runtime.leaderretrieval=INFO
+
[2/7] flink git commit: [FLINK-2829] [runtime] Clarify error message
when Flink cannot deploy task to slot
Posted by se...@apache.org.
[FLINK-2829] [runtime] Clarify error message when Flink cannot deploy task to slot
This closes #1993
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/77afe28e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/77afe28e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/77afe28e
Branch: refs/heads/master
Commit: 77afe28eb1d5cf454f7e64be20420051c30596a2
Parents: 035f629
Author: Joshi <re...@gmail.com>
Authored: Mon May 16 14:14:23 2016 -0700
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon May 30 14:41:34 2016 +0200
----------------------------------------------------------------------
.../java/org/apache/flink/runtime/executiongraph/Execution.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/77afe28e/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index a14248e..4ba58d3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -405,8 +405,8 @@ public class Execution implements Serializable {
}
else {
if (!(success.equals(Messages.getAcknowledge()))) {
- markFailed(new Exception("Failed to deploy the task to slot " + slot +
- ": Response was not of type Acknowledge"));
+ markFailed(new Exception("Failed to deploy the task to slot. Response was not of type 'Acknowledge', but was " + success
+ + "\nSlot Details: " + slot));
}
}
}