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 2015/12/31 15:33:28 UTC
[05/14] flink git commit: [hotfix] [web-dashboard] Various dashboard
code cleanups
[hotfix] [web-dashboard] Various dashboard code cleanups
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6f2d9d79
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6f2d9d79
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6f2d9d79
Branch: refs/heads/master
Commit: 6f2d9d79b85a759501ee267cc08a73dea03f7775
Parents: 72f9a6c
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Dec 28 02:31:07 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Dec 31 00:45:27 2015 +0100
----------------------------------------------------------------------
.../program/OptimizerPlanEnvironment.java | 11 +-
.../flink/configuration/ConfigConstants.java | 4 +-
.../runtime/webmonitor/HttpRequestHandler.java | 165 ++++++++++-----
.../webmonitor/PipelineErrorHandler.java | 52 ++---
.../webmonitor/RuntimeMonitorHandler.java | 13 +-
.../runtime/webmonitor/WebMonitorConfig.java | 6 +
.../runtime/webmonitor/WebRuntimeMonitor.java | 170 +++++++--------
.../handlers/ClusterOverviewHandler.java | 8 +-
.../handlers/ConstantTextHandler.java | 10 +-
.../handlers/CurrentJobIdsHandler.java | 20 +-
.../handlers/CurrentJobsOverviewHandler.java | 4 +-
.../handlers/DashboardConfigHandler.java | 4 +-
.../handlers/JarAccessDeniedHandler.java | 4 +-
.../webmonitor/handlers/JarActionHandler.java | 59 ++----
.../webmonitor/handlers/JarDeleteHandler.java | 2 +-
.../webmonitor/handlers/JarListHandler.java | 25 ++-
.../webmonitor/handlers/JarPlanHandler.java | 5 +-
.../webmonitor/handlers/JarUploadHandler.java | 39 ++--
.../handlers/JobAccumulatorsHandler.java | 4 +-
.../handlers/JobCancellationHandler.java | 6 +-
.../handlers/JobCheckpointsHandler.java | 3 +-
.../webmonitor/handlers/JobConfigHandler.java | 4 +-
.../webmonitor/handlers/JobDetailsHandler.java | 4 +-
.../handlers/JobExceptionsHandler.java | 4 +-
.../handlers/JobManagerConfigHandler.java | 4 +-
.../webmonitor/handlers/JobPlanHandler.java | 2 +-
.../handlers/JobVertexAccumulatorsHandler.java | 4 +-
.../handlers/JobVertexCheckpointsHandler.java | 3 +-
.../handlers/JobVertexDetailsHandler.java | 4 +-
.../webmonitor/handlers/RequestHandler.java | 27 ++-
...taskExecutionAttemptAccumulatorsHandler.java | 4 +-
.../SubtaskExecutionAttemptDetailsHandler.java | 4 +-
.../SubtasksAllAccumulatorsHandler.java | 4 +-
.../handlers/SubtasksTimesHandler.java | 4 +-
.../handlers/TaskManagersHandler.java | 13 +-
.../handlers/JobCheckpointsHandlerTest.java | 35 ++--
.../JobVertexCheckpointsHandlerTest.java | 28 ++-
.../scripts/modules/submit/submit.ctrl.coffee | 5 +-
flink-runtime-web/web-dashboard/package.json | 1 -
flink-runtime-web/web-dashboard/web/js/index.js | 208 +++++++------------
.../apache/flink/runtime/instance/Hardware.java | 25 +--
.../flink/runtime/jobmanager/JobManager.scala | 2 +-
.../flink/test/web/WebFrontendITCase.java | 4 +-
.../src/test/resources/log4j-test.properties | 2 +-
44 files changed, 479 insertions(+), 530 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-clients/src/main/java/org/apache/flink/client/program/OptimizerPlanEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/OptimizerPlanEnvironment.java b/flink-clients/src/main/java/org/apache/flink/client/program/OptimizerPlanEnvironment.java
index a5a9362..64076de 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/OptimizerPlanEnvironment.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/OptimizerPlanEnvironment.java
@@ -94,14 +94,15 @@ public class OptimizerPlanEnvironment extends ExecutionEnvironment {
unsetAsContext();
System.setOut(originalOut);
System.setErr(originalErr);
- System.err.println(baes);
- System.out.println(baos);
}
+ String stdout = baos.toString();
+ String stderr = baes.toString();
+
throw new ProgramInvocationException(
- "The program plan could not be fetched - the program aborted pre-maturely.\n"
- + "System.err: " + baes.toString() + '\n'
- + "System.out: " + baos.toString() + '\n');
+ "The program plan could not be fetched - the program aborted pre-maturely."
+ + "\n\nSystem.err: " + (stdout.length() == 0 ? "(none)" : stdout)
+ + "\n\nSystem.out: " + (stderr.length() == 0 ? "(none)" : stderr));
}
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index 1626e79..728fef6 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -347,7 +347,7 @@ public final class ConfigConstants {
public static final String JOB_MANAGER_WEB_LOG_PATH_KEY = "jobmanager.web.log.path";
/** Config parameter indicating whether jobs can be uploaded and run from the web-frontend. */
- public static final String JOB_MANAGER_WEB_SUBMISSION_KEY = "jobmanager.web.submit.enable";
+ public static final String JOB_MANAGER_WEB_SUBMIT_ENABLED_KEY = "jobmanager.web.submit.enable";
/** Flag to disable checkpoint stats. */
public static final String JOB_MANAGER_WEB_CHECKPOINTS_DISABLE = "jobmanager.web.checkpoints.disable";
@@ -701,7 +701,7 @@ public final class ConfigConstants {
public static final int DEFAULT_JOB_MANAGER_WEB_ARCHIVE_COUNT = 5;
/** By default, submitting jobs from the web-frontend is allowed. */
- public static final boolean DEFAULT_JOB_MANAGER_WEB_SUBMISSION = true;
+ public static final boolean DEFAULT_JOB_MANAGER_WEB_SUBMIT_ENABLED = true;
/** Default flag to disable checkpoint stats. */
public static final boolean DEFAULT_JOB_MANAGER_WEB_CHECKPOINTS_DISABLE = false;
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/HttpRequestHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/HttpRequestHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/HttpRequestHandler.java
index f0f3c92..c9190c9 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/HttpRequestHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/HttpRequestHandler.java
@@ -26,13 +26,18 @@ package org.apache.flink.runtime.webmonitor;
* https://github.com/netty/netty/blob/netty-4.0.31.Final/example/src/main/java/io/netty/example/http/upload/HttpUploadServerHandler.java
*****************************************************************************/
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.HttpContent;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpObject;
import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.handler.codec.http.QueryStringEncoder;
@@ -44,7 +49,11 @@ import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder.EndOfDataDec
import io.netty.handler.codec.http.multipart.InterfaceHttpData;
import io.netty.handler.codec.http.multipart.InterfaceHttpData.HttpDataType;
+import org.apache.flink.util.ExceptionUtils;
+
import java.io.File;
+import java.io.IOException;
+import java.nio.charset.Charset;
import java.util.UUID;
/**
@@ -53,78 +62,124 @@ import java.util.UUID;
* If a file is required to be uploaded, it handles the upload, and in the http request to the
* next handler, passes the name of the file to the next handler.
*/
+@ChannelHandler.Sharable
public class HttpRequestHandler extends SimpleChannelInboundHandler<HttpObject> {
- private HttpRequest request;
-
- private boolean readingChunks;
-
- private static final HttpDataFactory factory = new DefaultHttpDataFactory(true); // use disk
-
- private String requestPath;
+ private static final Charset ENCODING = Charset.forName("UTF-8");
+
+ /** A decoder factory that always stores POST chunks on disk */
+ private static final HttpDataFactory DATA_FACTORY = new DefaultHttpDataFactory(true);
+
+ private static final File TMP_DIR = new File(System.getProperty("java.io.tmpdir"));
+
+
+ private HttpRequest currentRequest;
- private HttpPostRequestDecoder decoder;
+ private HttpPostRequestDecoder currentDecoder;
+
+ private String currentRequestPath;
- private final File uploadDir;
-
- /**
- * The directory where files should be uploaded.
- */
- public HttpRequestHandler(File uploadDir) {
- this.uploadDir = uploadDir;
- }
@Override
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
- if (decoder != null) {
- decoder.cleanFiles();
+ if (currentDecoder != null) {
+ currentDecoder.cleanFiles();
}
}
@Override
- public void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
- if (msg instanceof HttpRequest) {
- request = (HttpRequest) msg;
- requestPath = new QueryStringDecoder(request.getUri()).path();
- if (request.getMethod() != HttpMethod.POST) {
- ctx.fireChannelRead(request);
- } else {
- // try to decode the posted data now.
- decoder = new HttpPostRequestDecoder(factory, request);
- readingChunks = HttpHeaders.isTransferEncodingChunked(request);
- if (readingChunks) {
- readingChunks = true;
+ public void channelRead0(ChannelHandlerContext ctx, HttpObject msg) {
+ try {
+ if (msg instanceof HttpRequest) {
+ currentRequest = (HttpRequest) msg;
+ currentRequestPath = null;
+
+ if (currentDecoder != null) {
+ currentDecoder.destroy();
+ currentDecoder = null;
+ }
+
+ if (currentRequest.getMethod() == HttpMethod.GET || currentRequest.getMethod() == HttpMethod.DELETE) {
+ // directly delegate to the router
+ ctx.fireChannelRead(currentRequest);
+ }
+ else if (currentRequest.getMethod() == HttpMethod.POST) {
+ // POST comes in multiple objects. First the request, then the contents
+ // keep the request and path for the remaining objects of the POST request
+ currentRequestPath = new QueryStringDecoder(currentRequest.getUri()).path();
+ currentDecoder = new HttpPostRequestDecoder(DATA_FACTORY, currentRequest);
+ }
+ else {
+ throw new IOException("Unsupported HTTP method: " + currentRequest.getMethod().name());
}
}
- } else if (decoder != null && msg instanceof HttpContent) {
- // New chunk is received
- HttpContent chunk = (HttpContent) msg;
- decoder.offer(chunk);
- try {
- while (decoder.hasNext()) {
- InterfaceHttpData data = decoder.next();
- // IF SOMETHING EVER NEEDS POST PARAMETERS, THIS WILL BE THE PLACE TO HANDLE IT
- // all fields values will be passed with type Attribute.
- if (data.getHttpDataType() == HttpDataType.FileUpload) {
- DiskFileUpload file = (DiskFileUpload) data;
- if (file.isCompleted()) {
- String newName = UUID.randomUUID() + "_" + file.getFilename();
- file.renameTo(new File(uploadDir, newName));
- QueryStringEncoder encoder = new QueryStringEncoder(requestPath);
- encoder.addParam("file", newName);
- request.setUri(encoder.toString());
+ else if (currentDecoder != null && msg instanceof HttpContent) {
+ // received new chunk, give it to the current decoder
+ HttpContent chunk = (HttpContent) msg;
+ currentDecoder.offer(chunk);
+
+ try {
+ while (currentDecoder.hasNext()) {
+ InterfaceHttpData data = currentDecoder.next();
+
+ // IF SOMETHING EVER NEEDS POST PARAMETERS, THIS WILL BE THE PLACE TO HANDLE IT
+ // all fields values will be passed with type Attribute.
+
+ if (data.getHttpDataType() == HttpDataType.FileUpload) {
+ DiskFileUpload file = (DiskFileUpload) data;
+ if (file.isCompleted()) {
+ String name = file.getFilename();
+
+ File target = new File(TMP_DIR, UUID.randomUUID() + "_" + name);
+ file.renameTo(target);
+
+ QueryStringEncoder encoder = new QueryStringEncoder(currentRequestPath);
+ encoder.addParam("filepath", target.getAbsolutePath());
+ encoder.addParam("filename", name);
+
+ currentRequest.setUri(encoder.toString());
+ }
}
+
+ data.release();
}
- data.release();
}
- } catch (EndOfDataDecoderException e) {
- //
+ catch (EndOfDataDecoderException ignored) {}
+
+ if (chunk instanceof LastHttpContent) {
+ HttpRequest request = currentRequest;
+ currentRequest = null;
+ currentRequestPath = null;
+
+ currentDecoder.destroy();
+ currentDecoder = null;
+
+ // fire next channel handler
+ ctx.fireChannelRead(request);
+ }
}
- if (chunk instanceof LastHttpContent) {
- readingChunks = false;
- decoder.destroy();
- decoder = null;
- ctx.fireChannelRead(request);
+ }
+ catch (Throwable t) {
+ currentRequest = null;
+ currentRequestPath = null;
+
+ if (currentDecoder != null) {
+ currentDecoder.destroy();
+ currentDecoder = null;
+ }
+
+ if (ctx.channel().isActive()) {
+ byte[] bytes = ExceptionUtils.stringifyException(t).getBytes(ENCODING);
+
+ DefaultFullHttpResponse response = new DefaultFullHttpResponse(
+ HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR,
+ Unpooled.wrappedBuffer(bytes));
+
+ response.headers().set(HttpHeaders.Names.CONTENT_ENCODING, "utf-8");
+ response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "text/plain");
+ response.headers().set(HttpHeaders.Names.CONTENT_LENGTH, response.content().readableBytes());
+
+ ctx.writeAndFlush(response);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/PipelineErrorHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/PipelineErrorHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/PipelineErrorHandler.java
index 2618b21..23a0ba6 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/PipelineErrorHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/PipelineErrorHandler.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.webmonitor;
-import com.fasterxml.jackson.core.JsonGenerator;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
@@ -27,56 +26,47 @@ import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
-import org.apache.flink.runtime.webmonitor.handlers.JsonFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.io.IOException;
-import java.io.StringWriter;
+import org.apache.flink.util.ExceptionUtils;
+
+import org.slf4j.Logger;
/**
- * This is the last handler in the pipeline and logs all error messages.
+ * This is the last handler in the pipeline. It logs all error messages and sends exception
+ * responses.
*/
@ChannelHandler.Sharable
public class PipelineErrorHandler extends SimpleChannelInboundHandler<Object> {
- private static final Logger LOG = LoggerFactory.getLogger(PipelineErrorHandler.class);
+ /** The logger to which the handler writes the log statements */
+ private final Logger logger;
+
+ public PipelineErrorHandler(Logger logger) {
+ this.logger = logger;
+ }
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object message) {
// we can't deal with this message. No one in the pipeline handled it. Log it.
- LOG.debug("Unknown message received: {}", message);
+ logger.debug("Unknown message received: {}", message);
sendError(ctx, "Unknown message received.");
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
- LOG.debug("Unhandled exception: {}", cause);
- sendError(ctx, cause.getMessage());
+ logger.debug("Unhandled exception: {}", cause);
+ sendError(ctx, ExceptionUtils.stringifyException(cause));
}
private void sendError(ChannelHandlerContext ctx, String error) {
- DefaultFullHttpResponse response;
- StringWriter writer = new StringWriter();
- try {
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
- gen.writeStartObject();
- gen.writeStringField("error", error);
- gen.writeEndObject();
- gen.close();
- // send a bad request status code.
- response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
- HttpResponseStatus.BAD_REQUEST, Unpooled.wrappedBuffer(writer.toString().getBytes()));
- response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "application/json");
- } catch (IOException e) {
- // seriously? Let's just send some plain text.
- response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
- HttpResponseStatus.BAD_REQUEST, Unpooled.wrappedBuffer(error.getBytes()));
- response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "text/plain");
- }
- response.headers().set(HttpHeaders.Names.CONTENT_ENCODING, "utf-8");
- response.headers().set(HttpHeaders.Names.CONTENT_LENGTH, response.content().readableBytes());
if (ctx.channel().isActive()) {
+ DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
+ HttpResponseStatus.INTERNAL_SERVER_ERROR, Unpooled.wrappedBuffer(error.getBytes()));
+
+ response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "text/plain");
+ response.headers().set(HttpHeaders.Names.CONTENT_ENCODING, "utf-8");
+ response.headers().set(HttpHeaders.Names.CONTENT_LENGTH, response.content().readableBytes());
+
ctx.writeAndFlush(response);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/RuntimeMonitorHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/RuntimeMonitorHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/RuntimeMonitorHandler.java
index c7d1e51..4143434 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/RuntimeMonitorHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/RuntimeMonitorHandler.java
@@ -30,10 +30,12 @@ import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.router.KeepAliveWrite;
import io.netty.handler.codec.http.router.Routed;
+
import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.webmonitor.handlers.HandlerRedirectUtils;
import org.apache.flink.runtime.webmonitor.handlers.RequestHandler;
import org.apache.flink.util.ExceptionUtils;
+
import scala.Option;
import scala.Tuple2;
import scala.concurrent.Await;
@@ -58,6 +60,9 @@ public class RuntimeMonitorHandler extends SimpleChannelInboundHandler<Routed> {
private static final Charset ENCODING = Charset.forName("UTF-8");
+ public static final String WEB_MONITOR_ADDRESS_KEY = "web.monitor.address";
+
+
private final RequestHandler handler;
private final JobManagerRetriever retriever;
@@ -66,11 +71,8 @@ public class RuntimeMonitorHandler extends SimpleChannelInboundHandler<Routed> {
private final FiniteDuration timeout;
- private final String contentType;
-
private String localJobManagerAddress;
-
- public static final String WEB_MONITOR_ADDRESS_KEY = "web.monitor.address";
+
public RuntimeMonitorHandler(
RequestHandler handler,
@@ -82,7 +84,6 @@ public class RuntimeMonitorHandler extends SimpleChannelInboundHandler<Routed> {
this.retriever = checkNotNull(retriever);
this.localJobManagerAddressFuture = checkNotNull(localJobManagerAddressFuture);
this.timeout = checkNotNull(timeout);
- this.contentType = (handler instanceof RequestHandler.JsonResponse) ? "application/json" : "text/plain";
}
@Override
@@ -134,7 +135,7 @@ public class RuntimeMonitorHandler extends SimpleChannelInboundHandler<Routed> {
HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer(bytes));
response.headers().set(HttpHeaders.Names.ACCESS_CONTROL_ALLOW_ORIGIN, "*");
- response.headers().set(HttpHeaders.Names.CONTENT_TYPE, contentType);
+ response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "application/json");
}
catch (NotFoundException e) {
// this should result in a 404 error code (not found)
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorConfig.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorConfig.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorConfig.java
index 5b537b7..dde6d0a 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorConfig.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorConfig.java
@@ -69,4 +69,10 @@ public class WebMonitorConfig {
public long getRefreshInterval() {
return config.getLong(JOB_MANAGER_WEB_REFRESH_INTERVAL_KEY, DEFAULT_JOB_MANAGER_WEB_REFRESH_INTERVAL);
}
+
+ public boolean isProgramSubmitEnabled() {
+ return config.getBoolean(
+ ConfigConstants.JOB_MANAGER_WEB_SUBMIT_ENABLED_KEY,
+ ConfigConstants.DEFAULT_JOB_MANAGER_WEB_SUBMIT_ENABLED);
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
index 4117800..3523091 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
@@ -19,23 +19,30 @@
package org.apache.flink.runtime.webmonitor;
import akka.actor.ActorSystem;
+
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.router.Handler;
import io.netty.handler.codec.http.router.Router;
-import io.netty.handler.stream.ChunkedWriteHandler;
+
import org.apache.commons.io.FileUtils;
+
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.webmonitor.files.StaticFileServerHandler;
import org.apache.flink.runtime.webmonitor.handlers.ConstantTextHandler;
+import org.apache.flink.runtime.webmonitor.handlers.JarAccessDeniedHandler;
+import org.apache.flink.runtime.webmonitor.handlers.JarDeleteHandler;
+import org.apache.flink.runtime.webmonitor.handlers.JarListHandler;
+import org.apache.flink.runtime.webmonitor.handlers.JarPlanHandler;
+import org.apache.flink.runtime.webmonitor.handlers.JarRunHandler;
+import org.apache.flink.runtime.webmonitor.handlers.JarUploadHandler;
import org.apache.flink.runtime.webmonitor.handlers.JobAccumulatorsHandler;
import org.apache.flink.runtime.webmonitor.handlers.JobCancellationHandler;
import org.apache.flink.runtime.webmonitor.handlers.JobCheckpointsHandler;
@@ -114,7 +121,7 @@ public class WebRuntimeMonitor implements WebMonitor {
private final File uploadDir;
- private AtomicBoolean isShutdown = new AtomicBoolean();
+ private AtomicBoolean cleanedUp = new AtomicBoolean();
public WebRuntimeMonitor(
@@ -123,40 +130,36 @@ public class WebRuntimeMonitor implements WebMonitor {
ActorSystem actorSystem) throws IOException, InterruptedException {
this.leaderRetrievalService = checkNotNull(leaderRetrievalService);
-
+ this.timeout = AkkaUtils.getTimeout(config);
+ this.retriever = new JobManagerRetriever(this, actorSystem, AkkaUtils.getTimeout(config), timeout);
+
final WebMonitorConfig cfg = new WebMonitorConfig(config);
-
- // create an empty directory in temp for the web server
- String fileName = String.format("flink-web-%s", UUID.randomUUID().toString());
- webRootDir = new File(System.getProperty("java.io.tmpdir"), fileName);
-
- // create storage for uploads
- fileName = String.format("flink-web-upload-%s", UUID.randomUUID().toString());
- uploadDir = new File(System.getProperty("java.io.tmpdir"), fileName);
- if (!uploadDir.mkdir() || !uploadDir.canWrite()) {
- throw new IOException("Unable to create temporary directory to support jar uploads.");
- }
-
- LOG.info("Using directory {} for the web interface files", webRootDir);
- LOG.info("Using directory {} for web frontend JAR file uploads", uploadDir);
-
- final WebMonitorUtils.LogFileLocation logFiles = WebMonitorUtils.LogFileLocation.find(config);
-
- final boolean webSubmitAllow = config.getBoolean(
- ConfigConstants.JOB_MANAGER_WEB_SUBMISSION_KEY,
- ConfigConstants.DEFAULT_JOB_MANAGER_WEB_SUBMISSION
- );
-
- // port configuration
- int configuredPort = cfg.getWebFrontendPort();
+
+ final int configuredPort = cfg.getWebFrontendPort();
if (configuredPort < 0) {
throw new IllegalArgumentException("Web frontend port is invalid: " + configuredPort);
}
+
+ final WebMonitorUtils.LogFileLocation logFiles = WebMonitorUtils.LogFileLocation.find(config);
+
+ // create an empty directory in temp for the web server
+ String rootDirFileName = "flink-web-" + UUID.randomUUID();
+ webRootDir = new File(System.getProperty("java.io.tmpdir"), rootDirFileName);
+ LOG.info("Using directory {} for the web interface files", webRootDir);
- timeout = AkkaUtils.getTimeout(config);
- FiniteDuration lookupTimeout = AkkaUtils.getTimeout(config);
-
- retriever = new JobManagerRetriever(this, actorSystem, lookupTimeout, timeout);
+ final boolean webSubmitAllow = cfg.isProgramSubmitEnabled();
+ if (webSubmitAllow) {
+ // create storage for uploads
+ String uploadDirName = "flink-web-upload-" + UUID.randomUUID();
+ this.uploadDir = new File(System.getProperty("java.io.tmpdir"), uploadDirName);
+ if (!uploadDir.mkdir() || !uploadDir.canWrite()) {
+ throw new IOException("Unable to create temporary directory to support jar uploads.");
+ }
+ LOG.info("Using directory {} for web frontend JAR file uploads", uploadDir);
+ }
+ else {
+ this.uploadDir = null;
+ }
ExecutionGraphHolder currentGraphs = new ExecutionGraphHolder();
@@ -175,7 +178,7 @@ public class WebRuntimeMonitor implements WebMonitor {
.GET("/joboverview/running", handler(new CurrentJobsOverviewHandler(DEFAULT_REQUEST_TIMEOUT, true, false)))
.GET("/joboverview/completed", handler(new CurrentJobsOverviewHandler(DEFAULT_REQUEST_TIMEOUT, false, true)))
- .GET("/jobs", handler(new CurrentJobIdsHandler(retriever, DEFAULT_REQUEST_TIMEOUT)))
+ .GET("/jobs", handler(new CurrentJobIdsHandler(DEFAULT_REQUEST_TIMEOUT)))
.GET("/jobs/:jobid", handler(new JobDetailsHandler(currentGraphs)))
.GET("/jobs/:jobid/vertices", handler(new JobDetailsHandler(currentGraphs)))
@@ -208,6 +211,7 @@ public class WebRuntimeMonitor implements WebMonitor {
// Cancel a job via GET (for proper integration with YARN this has to be performed via GET)
.GET("/jobs/:jobid/yarn-cancel", handler(new JobCancellationHandler()))
+
// DELETE is the preferred way of cancelling a job (Rest-conform)
.DELETE("/jobs/:jobid", handler(new JobCancellationHandler()));
@@ -238,61 +242,59 @@ public class WebRuntimeMonitor implements WebMonitor {
// this handler serves all the static contents
router.GET("/:*", new StaticFileServerHandler(retriever, jobManagerAddressPromise.future(), timeout, webRootDir));
- synchronized (startupShutdownLock) {
+ // add shutdown hook for deleting the directories and remaining temp files on shutdown
+ try {
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ cleanup();
+ }
+ });
+ } catch (IllegalStateException e) {
+ // race, JVM is in shutdown already, we can safely ignore this
+ LOG.debug("Unable to add shutdown hook, shutdown already in progress", e);
+ } catch (Throwable t) {
+ // these errors usually happen when the shutdown is already in progress
+ LOG.warn("Error while adding shutdown hook", t);
+ }
- // add shutdown hook for deleting the directory
- try {
- Runtime.getRuntime().addShutdownHook(new Thread() {
- @Override
- public void run() {
- shutdown();
- }
- });
- } catch (IllegalStateException e) {
- // race, JVM is in shutdown already, we can safely ignore this
- LOG.debug("Unable to add shutdown hook, shutdown already in progress", e);
- } catch(Throwable t) {
- // these errors usually happen when the shutdown is already in progress
- LOG.warn("Error while adding shutdown hook", t);
- }
+ ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {
- ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel ch) {
+ Handler handler = new Handler(router);
- @Override
- protected void initChannel(SocketChannel ch) {
- Handler handler = new Handler(router);
-
- ch.pipeline()
- .addLast(new HttpServerCodec())
- .addLast(new HttpRequestHandler(uploadDir))
- .addLast(handler.name(), handler)
- .addLast(new PipelineErrorHandler());
- }
- };
+ ch.pipeline()
+ .addLast(new HttpServerCodec())
+ .addLast(new HttpRequestHandler())
+ .addLast(handler.name(), handler)
+ .addLast(new PipelineErrorHandler(LOG));
+ }
+ };
- NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
- NioEventLoopGroup workerGroup = new NioEventLoopGroup();
+ NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
+ NioEventLoopGroup workerGroup = new NioEventLoopGroup();
- this.bootstrap = new ServerBootstrap();
- this.bootstrap
- .group(bossGroup, workerGroup)
- .channel(NioServerSocketChannel.class)
- .childHandler(initializer);
+ this.bootstrap = new ServerBootstrap();
+ this.bootstrap
+ .group(bossGroup, workerGroup)
+ .channel(NioServerSocketChannel.class)
+ .childHandler(initializer);
- Channel ch = this.bootstrap.bind(configuredPort).sync().channel();
- this.serverChannel = ch;
+ Channel ch = this.bootstrap.bind(configuredPort).sync().channel();
+ this.serverChannel = ch;
- InetSocketAddress bindAddress = (InetSocketAddress) ch.localAddress();
- String address = bindAddress.getAddress().getHostAddress();
- int port = bindAddress.getPort();
+ InetSocketAddress bindAddress = (InetSocketAddress) ch.localAddress();
+ String address = bindAddress.getAddress().getHostAddress();
+ int port = bindAddress.getPort();
- LOG.info("Web frontend listening at " + address + ':' + port);
- }
+ LOG.info("Web frontend listening at " + address + ':' + port);
}
@Override
public void start(String jobManagerAkkaUrl) throws Exception {
LOG.info("Starting with JobManager {} on port {}", jobManagerAkkaUrl, getServerPort());
+
synchronized (startupShutdownLock) {
jobManagerAddressPromise.success(jobManagerAkkaUrl);
leaderRetrievalService.start(retriever);
@@ -314,7 +316,7 @@ public class WebRuntimeMonitor implements WebMonitor {
}
}
- shutdown();
+ cleanup();
}
}
@@ -333,8 +335,8 @@ public class WebRuntimeMonitor implements WebMonitor {
return -1;
}
- private void shutdown() {
- if (!isShutdown.compareAndSet(false, true)) {
+ private void cleanup() {
+ if (!cleanedUp.compareAndSet(false, true)) {
return;
}
try {
@@ -344,11 +346,13 @@ public class WebRuntimeMonitor implements WebMonitor {
LOG.warn("Error while deleting web root directory {}", webRootDir, t);
}
- try {
- LOG.info("Removing web dashboard jar upload directory {}", uploadDir);
- FileUtils.deleteDirectory(uploadDir);
- } catch (Throwable t) {
- LOG.warn("Error while deleting web storage dir {}", uploadDir, t);
+ if (uploadDir != null) {
+ try {
+ LOG.info("Removing web dashboard jar upload directory {}", uploadDir);
+ FileUtils.deleteDirectory(uploadDir);
+ } catch (Throwable t) {
+ LOG.warn("Error while deleting web storage dir {}", uploadDir, t);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/ClusterOverviewHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/ClusterOverviewHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/ClusterOverviewHandler.java
index ae43a00..2782ee4 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/ClusterOverviewHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/ClusterOverviewHandler.java
@@ -36,14 +36,14 @@ import static com.google.common.base.Preconditions.checkNotNull;
* Responder that returns the status of the Flink cluster, such as how many
* TaskManagers are currently connected, and how many jobs are running.
*/
-public class ClusterOverviewHandler implements RequestHandler, RequestHandler.JsonResponse {
-
- private final FiniteDuration timeout;
+public class ClusterOverviewHandler implements RequestHandler {
private static final String version = EnvironmentInformation.getVersion();
private static final String commitID = EnvironmentInformation.getRevisionInformation().commitId;
+ private final FiniteDuration timeout;
+
public ClusterOverviewHandler(FiniteDuration timeout) {
this.timeout = checkNotNull(timeout);
}
@@ -57,7 +57,7 @@ public class ClusterOverviewHandler implements RequestHandler, RequestHandler.Js
StatusOverview overview = (StatusOverview) Await.result(future, timeout);
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
gen.writeNumberField("taskmanagers", overview.getNumTaskManagersConnected());
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/ConstantTextHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/ConstantTextHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/ConstantTextHandler.java
index 30d825f..aedf0c2 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/ConstantTextHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/ConstantTextHandler.java
@@ -30,8 +30,6 @@ import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.router.KeepAliveWrite;
import io.netty.handler.codec.http.router.Routed;
-import org.apache.flink.runtime.webmonitor.files.MimeTypes;
-
import java.io.UnsupportedEncodingException;
/**
@@ -40,8 +38,6 @@ import java.io.UnsupportedEncodingException;
@ChannelHandler.Sharable
public class ConstantTextHandler extends SimpleChannelInboundHandler<Routed> {
- private static final String MIME_TYPE = MimeTypes.getMimeTypeForExtension("txt");
-
private final byte[] encodedText;
public ConstantTextHandler(String text) {
@@ -58,10 +54,10 @@ public class ConstantTextHandler extends SimpleChannelInboundHandler<Routed> {
HttpResponse response = new DefaultFullHttpResponse(
HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer(encodedText));
- response.headers().set(HttpHeaders.Names.CONTENT_ENCODING, "utf-8");
response.headers().set(HttpHeaders.Names.CONTENT_LENGTH, encodedText.length);
- response.headers().set(HttpHeaders.Names.CONTENT_TYPE, MIME_TYPE);
-
+ response.headers().set(HttpHeaders.Names.CONTENT_ENCODING, "utf-8");
+ response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "text/plain");
+
KeepAliveWrite.flush(ctx, routed.request(), response);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/CurrentJobIdsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/CurrentJobIdsHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/CurrentJobIdsHandler.java
index 41b118c..11f2a3b 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/CurrentJobIdsHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/CurrentJobIdsHandler.java
@@ -24,7 +24,6 @@ import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.messages.webmonitor.JobsWithIDsOverview;
import org.apache.flink.runtime.messages.webmonitor.RequestJobsWithIDsOverview;
-import org.apache.flink.runtime.webmonitor.JobManagerRetriever;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
@@ -32,24 +31,19 @@ import scala.concurrent.duration.FiniteDuration;
import java.io.StringWriter;
import java.util.Map;
+import static java.util.Objects.requireNonNull;
+
/**
* Responder that returns with a list of all JobIDs of jobs found at the target actor.
* May serve the IDs of current jobs, or past jobs, depending on whether this handler is
* given the JobManager or Archive Actor Reference.
*/
-public class CurrentJobIdsHandler implements RequestHandler, RequestHandler.JsonResponse {
-
- private final JobManagerRetriever retriever;
+public class CurrentJobIdsHandler implements RequestHandler {
private final FiniteDuration timeout;
-
-
- public CurrentJobIdsHandler(JobManagerRetriever retriever, FiniteDuration timeout) {
- if (retriever == null || timeout == null) {
- throw new NullPointerException();
- }
- this.retriever = retriever;
- this.timeout = timeout;
+
+ public CurrentJobIdsHandler(FiniteDuration timeout) {
+ this.timeout = requireNonNull(timeout);
}
@Override
@@ -61,7 +55,7 @@ public class CurrentJobIdsHandler implements RequestHandler, RequestHandler.Json
JobsWithIDsOverview overview = (JobsWithIDsOverview) Await.result(future, timeout);
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/CurrentJobsOverviewHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/CurrentJobsOverviewHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/CurrentJobsOverviewHandler.java
index d9e0096..e139c5c 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/CurrentJobsOverviewHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/CurrentJobsOverviewHandler.java
@@ -36,7 +36,7 @@ import static com.google.common.base.Preconditions.checkNotNull;
/**
* Request handler that returns a summary of the job status.
*/
-public class CurrentJobsOverviewHandler implements RequestHandler, RequestHandler.JsonResponse {
+public class CurrentJobsOverviewHandler implements RequestHandler {
private final FiniteDuration timeout;
@@ -66,7 +66,7 @@ public class CurrentJobsOverviewHandler implements RequestHandler, RequestHandle
final long now = System.currentTimeMillis();
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/DashboardConfigHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/DashboardConfigHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/DashboardConfigHandler.java
index 71816df..debb24c 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/DashboardConfigHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/DashboardConfigHandler.java
@@ -31,7 +31,7 @@ import java.util.TimeZone;
* against this web server should behave. It defines for example the refresh interval,
* and time zone of the server timestamps.
*/
-public class DashboardConfigHandler implements RequestHandler, RequestHandler.JsonResponse {
+public class DashboardConfigHandler implements RequestHandler {
private final String configString;
@@ -42,7 +42,7 @@ public class DashboardConfigHandler implements RequestHandler, RequestHandler.Js
try {
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
gen.writeNumberField("refresh-interval", refreshInterval);
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarAccessDeniedHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarAccessDeniedHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarAccessDeniedHandler.java
index 024433d..67673e2 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarAccessDeniedHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarAccessDeniedHandler.java
@@ -22,10 +22,10 @@ import org.apache.flink.runtime.instance.ActorGateway;
import java.util.Map;
-public class JarAccessDeniedHandler implements RequestHandler, RequestHandler.JsonResponse {
+public class JarAccessDeniedHandler implements RequestHandler {
private static final String ERROR_MESSAGE = "{\"error\": \"Web submission interface is not " +
- "available for this cluster. Please check the config key jobmanager.web.submit.enable.\"}";
+ "available for this cluster. To enable it, set the configuration key ' jobmanager.web.submit.enable.'\"}";
@Override
public String handleRequest(Map<String, String> pathParams, Map<String, String> queryParams, ActorGateway jobManager) throws Exception {
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarActionHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarActionHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarActionHandler.java
index e185003..03bae4e 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarActionHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarActionHandler.java
@@ -34,11 +34,9 @@ import org.apache.flink.optimizer.plan.OptimizedPlan;
import org.apache.flink.optimizer.plan.StreamingPlan;
import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.util.ExceptionUtils;
import java.io.File;
-import java.io.OutputStream;
-import java.io.PrintStream;
-import java.io.PrintWriter;
import java.io.StringWriter;
import java.net.URISyntaxException;
import java.net.URL;
@@ -49,12 +47,11 @@ import java.util.Map;
/**
* Abstract handler for fetching plan for a jar or running a jar.
*/
-public abstract class JarActionHandler implements RequestHandler, RequestHandler.JsonResponse {
-
+public abstract class JarActionHandler implements RequestHandler {
+
private final File jarDir;
- private static final PrintStream nullStream = new PrintStream(new NullPrintStream());
-
+
public JarActionHandler(File jarDirectory) {
jarDir = jarDirectory;
}
@@ -93,14 +90,10 @@ public abstract class JarActionHandler implements RequestHandler, RequestHandler
PackagedProgram program = new PackagedProgram(new File(jarDir, file), entryClass,
programArgs.toArray(new String[programArgs.size()]));
ClassLoader classLoader = program.getUserCodeClassLoader();
- Optimizer compiler = new Optimizer(new DataStatistics(), new DefaultCostEstimator(), new Configuration());
- PrintStream out = System.out;
- PrintStream err = System.err;
- System.setOut(nullStream);
- System.setErr(nullStream);
- FlinkPlan plan = Client.getOptimizedPlan(compiler, program, parallelism);
- System.setOut(out);
- System.setErr(err);
+
+ Optimizer optimizer = new Optimizer(new DataStatistics(), new DefaultCostEstimator(), new Configuration());
+ FlinkPlan plan = Client.getOptimizedPlan(optimizer, program, parallelism);
+
if (plan instanceof StreamingPlan) {
graph = ((StreamingPlan) plan).getJobGraph();
} else if (plan instanceof OptimizedPlan) {
@@ -112,7 +105,8 @@ public abstract class JarActionHandler implements RequestHandler, RequestHandler
for (URL jar : program.getAllLibraries()) {
try {
graph.addJar(new Path(jar.toURI()));
- } catch (URISyntaxException e) {
+ }
+ catch (URISyntaxException e) {
throw new ProgramInvocationException("Invalid jar path. Unexpected error. :(");
}
}
@@ -163,39 +157,14 @@ public abstract class JarActionHandler implements RequestHandler, RequestHandler
}
protected String sendError(Exception e) throws Exception {
- StringWriter sw = new StringWriter();
- PrintWriter p = new PrintWriter(sw);
- if (e instanceof ProgramInvocationException || e instanceof CompilerException || e instanceof IllegalArgumentException) {
- p.println(e.getClass().getSimpleName() + ((e.getMessage() != null) ? ": " + e.getMessage() : ""));
- Throwable cause = e.getCause();
- if (cause != null) {
- p.println(cause.toString());
- } else {
- cause = e;
- }
-
- for (StackTraceElement traceElement: cause.getStackTrace()) {
- p.println("\tat " + traceElement);
- if (traceElement.getMethodName().equals("handleRequest")) {
- break;
- }
- }
- } else {
- // if not something we expected, dump the entire stack trace.
- e.printStackTrace(p);
- }
- p.close();
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
+
gen.writeStartObject();
- gen.writeStringField("error", sw.toString());
+ gen.writeStringField("error", ExceptionUtils.stringifyException(e));
gen.writeEndObject();
gen.close();
+
return writer.toString();
}
-
- private static final class NullPrintStream extends OutputStream {
- public void write(int x) {
- }
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarDeleteHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarDeleteHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarDeleteHandler.java
index 770e7e9..6599fbd 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarDeleteHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarDeleteHandler.java
@@ -29,7 +29,7 @@ import java.util.Map;
/**
* Handles requests for deletion of jars.
*/
-public class JarDeleteHandler implements RequestHandler, RequestHandler.JsonResponse {
+public class JarDeleteHandler implements RequestHandler {
private final File jarDir;
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarListHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarListHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarListHandler.java
index 980ed13..ede80a3 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarListHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarListHandler.java
@@ -29,11 +29,10 @@ import java.io.FilenameFilter;
import java.io.IOException;
import java.io.StringWriter;
import java.util.Map;
-import java.util.UUID;
import java.util.jar.JarFile;
import java.util.jar.Manifest;
-public class JarListHandler implements RequestHandler, RequestHandler.JsonResponse {
+public class JarListHandler implements RequestHandler {
private final File jarDir;
@@ -45,7 +44,8 @@ public class JarListHandler implements RequestHandler, RequestHandler.JsonRespon
public String handleRequest(Map<String, String> pathParams, Map<String, String> queryParams, ActorGateway jobManager) throws Exception {
try {
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
+
gen.writeStartObject();
gen.writeStringField("address", queryParams.get(RuntimeMonitorHandler.WEB_MONITOR_ADDRESS_KEY));
gen.writeArrayFieldStart("files");
@@ -56,24 +56,26 @@ public class JarListHandler implements RequestHandler, RequestHandler.JsonRespon
return name.endsWith(".jar");
}
});
+
for (File f : list) {
// separate the uuid and the name parts.
String id = f.getName();
+
int startIndex = id.indexOf("_");
- try {
- UUID.fromString(id.substring(0, startIndex));
- if (id.substring(startIndex + 1).equals(".jar")) {
- throw new Exception();
- }
- } catch (Exception e) {
+ if (startIndex < 0) {
continue;
}
String name = id.substring(startIndex + 1);
+ if (name.length() < 5 || !name.endsWith(".jar")) {
+ continue;
+ }
+
gen.writeStartObject();
gen.writeStringField("id", id);
gen.writeStringField("name", name);
gen.writeNumberField("uploaded", f.lastModified());
gen.writeArrayFieldStart("entry");
+
String[] classes = new String[0];
try {
JarFile jar = new JarFile(f);
@@ -89,9 +91,10 @@ public class JarListHandler implements RequestHandler, RequestHandler.JsonRespon
if (assemblerClass != null) {
classes = assemblerClass.split(",");
}
- } catch (IOException e) {
- //
+ } catch (IOException ignored) {
+ // we simply show no entries here
}
+
// show every entry class that can be loaded later on.
PackagedProgram program;
for (String clazz : classes) {
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarPlanHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarPlanHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarPlanHandler.java
index 0d7aa0e..7e0a810 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarPlanHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarPlanHandler.java
@@ -41,14 +41,15 @@ public class JarPlanHandler extends JarActionHandler {
try {
JobGraph graph = getJobGraphAndClassLoader(pathParams, queryParams).f0;
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
gen.writeFieldName("plan");
gen.writeRawValue(JsonPlanGenerator.generatePlan(graph));
gen.writeEndObject();
gen.close();
return writer.toString();
- } catch (Exception e) {
+ }
+ catch (Exception e) {
return sendError(e);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarUploadHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarUploadHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarUploadHandler.java
index 2889c2a..011e8f9 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarUploadHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarUploadHandler.java
@@ -22,11 +22,12 @@ import org.apache.flink.runtime.instance.ActorGateway;
import java.io.File;
import java.util.Map;
+import java.util.UUID;
/**
* Handles requests for uploading of jars.
*/
-public class JarUploadHandler implements RequestHandler, RequestHandler.JsonResponse {
+public class JarUploadHandler implements RequestHandler {
private final File jarDir;
@@ -35,19 +36,33 @@ public class JarUploadHandler implements RequestHandler, RequestHandler.JsonResp
}
@Override
- public String handleRequest(Map<String, String> pathParams, Map<String, String> queryParams, ActorGateway jobManager) throws Exception {
- String filename = queryParams.get("file");
- if(filename != null) {
- File f = new File(jarDir, filename);
- if (f.exists()) {
- if (f.getName().endsWith(".jar")) {
- return "{}";
- } else {
- f.delete();
- return "{\"error\": \"Only Jar files are allowed.\"}";
- }
+ public String handleRequest(
+ Map<String, String> pathParams,
+ Map<String, String> queryParams,
+ ActorGateway jobManager) throws Exception {
+
+ String tempFilePath = queryParams.get("filepath");
+ String filename = queryParams.get("filename");
+
+ File tempFile;
+ if (tempFilePath != null && (tempFile = new File(tempFilePath)).exists()) {
+ if (!tempFile.getName().endsWith(".jar")) {
+ //noinspection ResultOfMethodCallIgnored
+ tempFile.delete();
+ return "{\"error\": \"Only Jar files are allowed.\"}";
+ }
+
+ File newFile = new File(jarDir, UUID.randomUUID() + "_" + filename);
+ if (tempFile.renameTo(newFile)) {
+ // all went well
+ return "{}";
+ }
+ else {
+ //noinspection ResultOfMethodCallIgnored
+ tempFile.delete();
}
}
+
return "{\"error\": \"Failed to upload the file.\"}";
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobAccumulatorsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobAccumulatorsHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobAccumulatorsHandler.java
index 2732497..c5418b3 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobAccumulatorsHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobAccumulatorsHandler.java
@@ -29,7 +29,7 @@ import java.util.Map;
/**
* Request handler that returns the aggregated user accumulators of a job.
*/
-public class JobAccumulatorsHandler extends AbstractExecutionGraphRequestHandler implements RequestHandler.JsonResponse {
+public class JobAccumulatorsHandler extends AbstractExecutionGraphRequestHandler {
public JobAccumulatorsHandler(ExecutionGraphHolder executionGraphHolder) {
super(executionGraphHolder);
@@ -40,7 +40,7 @@ public class JobAccumulatorsHandler extends AbstractExecutionGraphRequestHandler
StringifiedAccumulatorResult[] allAccumulators = graph.getAccumulatorResultsStringified();
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobCancellationHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobCancellationHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobCancellationHandler.java
index 396e96d..aae8b34 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobCancellationHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobCancellationHandler.java
@@ -25,7 +25,7 @@ import org.apache.flink.util.StringUtils;
import java.util.Map;
-public class JobCancellationHandler implements RequestHandler, RequestHandler.JsonResponse {
+public class JobCancellationHandler implements RequestHandler {
@Override
public String handleRequest(Map<String, String> pathParams, Map<String, String> queryParams, ActorGateway jobManager) throws Exception {
@@ -33,14 +33,14 @@ public class JobCancellationHandler implements RequestHandler, RequestHandler.Js
JobID jobid = new JobID(StringUtils.hexStringToByte(pathParams.get("jobid")));
if (jobManager != null) {
jobManager.tell(new JobManagerMessages.CancelJob(jobid));
- return "";
+ return "{}";
}
else {
throw new Exception("No connection to the leading JobManager.");
}
}
catch (Exception e) {
- throw new RuntimeException("Failed to cancel the job with id: " + pathParams.get("jobid") + e.getMessage(), e);
+ throw new Exception("Failed to cancel the job with id: " + pathParams.get("jobid") + e.getMessage(), e);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobCheckpointsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobCheckpointsHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobCheckpointsHandler.java
index c8ee862..41735c5 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobCheckpointsHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobCheckpointsHandler.java
@@ -32,8 +32,7 @@ import java.util.Map;
/**
* Request handler that returns checkpoint stats for a job.
*/
-public class JobCheckpointsHandler extends AbstractExecutionGraphRequestHandler
- implements RequestHandler.JsonResponse {
+public class JobCheckpointsHandler extends AbstractExecutionGraphRequestHandler {
public JobCheckpointsHandler(ExecutionGraphHolder executionGraphHolder) {
super(executionGraphHolder);
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobConfigHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobConfigHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobConfigHandler.java
index a6c16ff..f39bb1a 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobConfigHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobConfigHandler.java
@@ -29,7 +29,7 @@ import org.apache.flink.runtime.webmonitor.ExecutionGraphHolder;
/**
* Request handler that returns the execution config of a job.
*/
-public class JobConfigHandler extends AbstractExecutionGraphRequestHandler implements RequestHandler.JsonResponse {
+public class JobConfigHandler extends AbstractExecutionGraphRequestHandler {
public JobConfigHandler(ExecutionGraphHolder executionGraphHolder) {
super(executionGraphHolder);
@@ -39,7 +39,7 @@ public class JobConfigHandler extends AbstractExecutionGraphRequestHandler imple
public String handleRequest(ExecutionGraph graph, Map<String, String> params) throws Exception {
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
gen.writeStringField("jid", graph.getJobID().toString());
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobDetailsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobDetailsHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobDetailsHandler.java
index 25b0f19..4511c17 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobDetailsHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobDetailsHandler.java
@@ -43,7 +43,7 @@ import java.util.Map;
* <li>info about job vertices, including runtime, status, I/O bytes and records, subtasks in each status</li>
* </ul>
*/
-public class JobDetailsHandler extends AbstractExecutionGraphRequestHandler implements RequestHandler.JsonResponse {
+public class JobDetailsHandler extends AbstractExecutionGraphRequestHandler {
public JobDetailsHandler(ExecutionGraphHolder executionGraphHolder) {
super(executionGraphHolder);
@@ -52,7 +52,7 @@ public class JobDetailsHandler extends AbstractExecutionGraphRequestHandler impl
@Override
public String handleRequest(ExecutionGraph graph, Map<String, String> params) throws Exception {
final StringWriter writer = new StringWriter();
- final JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ final JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
final long now = System.currentTimeMillis();
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobExceptionsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobExceptionsHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobExceptionsHandler.java
index 8e01465..0d71dcb 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobExceptionsHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobExceptionsHandler.java
@@ -31,7 +31,7 @@ import java.util.Map;
/**
* Request handler that returns the configuration of a job.
*/
-public class JobExceptionsHandler extends AbstractExecutionGraphRequestHandler implements RequestHandler.JsonResponse {
+public class JobExceptionsHandler extends AbstractExecutionGraphRequestHandler {
private static final int MAX_NUMBER_EXCEPTION_TO_REPORT = 20;
@@ -42,7 +42,7 @@ public class JobExceptionsHandler extends AbstractExecutionGraphRequestHandler i
@Override
public String handleRequest(ExecutionGraph graph, Map<String, String> params) throws Exception {
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobManagerConfigHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobManagerConfigHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobManagerConfigHandler.java
index a692243..fce5b92 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobManagerConfigHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobManagerConfigHandler.java
@@ -28,7 +28,7 @@ import java.util.Map;
/**
* Returns the Job Manager's configuration.
*/
-public class JobManagerConfigHandler implements RequestHandler, RequestHandler.JsonResponse {
+public class JobManagerConfigHandler implements RequestHandler {
private final Configuration config;
@@ -39,7 +39,7 @@ public class JobManagerConfigHandler implements RequestHandler, RequestHandler.J
@Override
public String handleRequest(Map<String, String> pathParams, Map<String, String> queryParams, ActorGateway jobManager) throws Exception {
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartArray();
for (String key : config.keySet()) {
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobPlanHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobPlanHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobPlanHandler.java
index 99838b3..0389f5a 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobPlanHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobPlanHandler.java
@@ -26,7 +26,7 @@ import java.util.Map;
/**
* Request handler that returns the JSON program plan of a job graph.
*/
-public class JobPlanHandler extends AbstractExecutionGraphRequestHandler implements RequestHandler.JsonResponse {
+public class JobPlanHandler extends AbstractExecutionGraphRequestHandler {
public JobPlanHandler(ExecutionGraphHolder executionGraphHolder) {
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexAccumulatorsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexAccumulatorsHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexAccumulatorsHandler.java
index ed2c541..5df565a 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexAccumulatorsHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexAccumulatorsHandler.java
@@ -28,7 +28,7 @@ import java.io.StringWriter;
import java.util.Map;
-public class JobVertexAccumulatorsHandler extends AbstractJobVertexRequestHandler implements RequestHandler.JsonResponse {
+public class JobVertexAccumulatorsHandler extends AbstractJobVertexRequestHandler {
public JobVertexAccumulatorsHandler(ExecutionGraphHolder executionGraphHolder) {
super(executionGraphHolder);
@@ -39,7 +39,7 @@ public class JobVertexAccumulatorsHandler extends AbstractJobVertexRequestHandle
StringifiedAccumulatorResult[] accs = jobVertex.getAggregatedUserAccumulatorsStringified();
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
gen.writeStringField("id", jobVertex.getJobVertexId().toString());
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexCheckpointsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexCheckpointsHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexCheckpointsHandler.java
index 9feeeb7..6522de5 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexCheckpointsHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexCheckpointsHandler.java
@@ -31,8 +31,7 @@ import java.util.Map;
/**
* Request handler that returns checkpoint stats for a single job vertex.
*/
-public class JobVertexCheckpointsHandler extends AbstractJobVertexRequestHandler
- implements RequestHandler.JsonResponse {
+public class JobVertexCheckpointsHandler extends AbstractJobVertexRequestHandler {
public JobVertexCheckpointsHandler(ExecutionGraphHolder executionGraphHolder) {
super(executionGraphHolder);
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexDetailsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexDetailsHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexDetailsHandler.java
index 21e2868..4190cff 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexDetailsHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JobVertexDetailsHandler.java
@@ -36,7 +36,7 @@ import java.util.Map;
* A request handler that provides the details of a job vertex, including id, name, parallelism,
* and the runtime and metrics of all its subtasks.
*/
-public class JobVertexDetailsHandler extends AbstractJobVertexRequestHandler implements RequestHandler.JsonResponse {
+public class JobVertexDetailsHandler extends AbstractJobVertexRequestHandler {
public JobVertexDetailsHandler(ExecutionGraphHolder executionGraphHolder) {
super(executionGraphHolder);
@@ -47,7 +47,7 @@ public class JobVertexDetailsHandler extends AbstractJobVertexRequestHandler imp
final long now = System.currentTimeMillis();
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/RequestHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/RequestHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/RequestHandler.java
index 9c47544..0927b7e 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/RequestHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/RequestHandler.java
@@ -22,20 +22,25 @@ import org.apache.flink.runtime.instance.ActorGateway;
import java.util.Map;
+/**
+ * Base interface for all request handlers. The handlers must produce a JSOn response.
+ */
public interface RequestHandler {
/**
- * This interface marks handlers that return JSON data.
+ * Core method that handles the request and generates the response. The method needs to
+ * respond with a valid JSON string. Exceptions may be throws and will be handled.
+ *
+ * @param pathParams The map of REST path parameters, decoded by the router.
+ * @param queryParams The map of query parameters.
+ * @param jobManager The JobManager actor.
+ *
+ * @return The JSON string that is the HTTP response.
+ *
+ * @throws Exception Handlers may forward exceptions. Exceptions of type
+ * {@link org.apache.flink.runtime.webmonitor.NotFoundException} will cause a HTTP 404
+ * response with the exception message, other exceptions will cause a HTTP 500 response
+ * with the exception stack trace.
*/
- interface JsonResponse {}
-
- /**
- * This interface marks handlers that return plain text data.
- */
- interface TextResponse {}
-
-
- // --------------------------------------------------------------------------------------------
-
String handleRequest(Map<String, String> pathParams, Map<String, String> queryParams, ActorGateway jobManager) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtaskExecutionAttemptAccumulatorsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtaskExecutionAttemptAccumulatorsHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtaskExecutionAttemptAccumulatorsHandler.java
index 85392b8..0111e8c 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtaskExecutionAttemptAccumulatorsHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtaskExecutionAttemptAccumulatorsHandler.java
@@ -31,7 +31,7 @@ import java.util.Map;
* Base class for request handlers whose response depends on a specific job vertex (defined
* via the "vertexid" parameter) in a specific job, defined via (defined voa the "jobid" parameter).
*/
-public class SubtaskExecutionAttemptAccumulatorsHandler extends AbstractSubtaskAttemptRequestHandler implements RequestHandler.JsonResponse {
+public class SubtaskExecutionAttemptAccumulatorsHandler extends AbstractSubtaskAttemptRequestHandler {
public SubtaskExecutionAttemptAccumulatorsHandler(ExecutionGraphHolder executionGraphHolder) {
super(executionGraphHolder);
@@ -42,7 +42,7 @@ public class SubtaskExecutionAttemptAccumulatorsHandler extends AbstractSubtaskA
final StringifiedAccumulatorResult[] accs = execAttempt.getUserAccumulatorsStringified();
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtaskExecutionAttemptDetailsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtaskExecutionAttemptDetailsHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtaskExecutionAttemptDetailsHandler.java
index bde9ce5..3d80b23 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtaskExecutionAttemptDetailsHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtaskExecutionAttemptDetailsHandler.java
@@ -34,7 +34,7 @@ import java.util.Map;
/**
* Request handler providing details about a single task execution attempt.
*/
-public class SubtaskExecutionAttemptDetailsHandler extends AbstractSubtaskAttemptRequestHandler implements RequestHandler.JsonResponse {
+public class SubtaskExecutionAttemptDetailsHandler extends AbstractSubtaskAttemptRequestHandler {
public SubtaskExecutionAttemptDetailsHandler(ExecutionGraphHolder executionGraphHolder) {
super(executionGraphHolder);
@@ -75,7 +75,7 @@ public class SubtaskExecutionAttemptDetailsHandler extends AbstractSubtaskAttemp
}
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
gen.writeNumberField("subtask", execAttempt.getVertex().getParallelSubtaskIndex());
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtasksAllAccumulatorsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtasksAllAccumulatorsHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtasksAllAccumulatorsHandler.java
index 4f97292..6d9ce3a 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtasksAllAccumulatorsHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtasksAllAccumulatorsHandler.java
@@ -32,7 +32,7 @@ import java.util.Map;
/**
* Request handler that returns the accumulators for all subtasks of job vertex.
*/
-public class SubtasksAllAccumulatorsHandler extends AbstractJobVertexRequestHandler implements RequestHandler.JsonResponse {
+public class SubtasksAllAccumulatorsHandler extends AbstractJobVertexRequestHandler {
public SubtasksAllAccumulatorsHandler(ExecutionGraphHolder executionGraphHolder) {
super(executionGraphHolder);
@@ -41,7 +41,7 @@ public class SubtasksAllAccumulatorsHandler extends AbstractJobVertexRequestHand
@Override
public String handleRequest(ExecutionJobVertex jobVertex, Map<String, String> params) throws Exception {
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
gen.writeStringField("id", jobVertex.getJobVertexId().toString());
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtasksTimesHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtasksTimesHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtasksTimesHandler.java
index b6fa136..03d40dc 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtasksTimesHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/SubtasksTimesHandler.java
@@ -33,7 +33,7 @@ import java.util.Map;
* Request handler that returns the state transition timestamps for all subtasks, plus their
* location and duration.
*/
-public class SubtasksTimesHandler extends AbstractJobVertexRequestHandler implements RequestHandler.JsonResponse {
+public class SubtasksTimesHandler extends AbstractJobVertexRequestHandler {
public SubtasksTimesHandler(ExecutionGraphHolder executionGraphHolder) {
@@ -45,7 +45,7 @@ public class SubtasksTimesHandler extends AbstractJobVertexRequestHandler implem
final long now = System.currentTimeMillis();
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/TaskManagersHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/TaskManagersHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/TaskManagersHandler.java
index c44b42b..b60cd10 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/TaskManagersHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/TaskManagersHandler.java
@@ -26,6 +26,7 @@ import org.apache.flink.runtime.messages.JobManagerMessages;
import org.apache.flink.runtime.messages.JobManagerMessages.RegisteredTaskManagers;
import org.apache.flink.runtime.messages.JobManagerMessages.TaskManagerInstance;
import org.apache.flink.util.StringUtils;
+
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
@@ -35,16 +36,16 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-import static com.google.common.base.Preconditions.checkNotNull;
-
-public class TaskManagersHandler implements RequestHandler, RequestHandler.JsonResponse {
+import static java.util.Objects.requireNonNull;
- private final FiniteDuration timeout;
+public class TaskManagersHandler implements RequestHandler {
public static final String TASK_MANAGER_ID_KEY = "taskmanagerid";
+ private final FiniteDuration timeout;
+
public TaskManagersHandler(FiniteDuration timeout) {
- this.timeout = checkNotNull(timeout);
+ this.timeout = requireNonNull(timeout);
}
@Override
@@ -75,7 +76,7 @@ public class TaskManagersHandler implements RequestHandler, RequestHandler.JsonR
}
StringWriter writer = new StringWriter();
- JsonGenerator gen = JsonFactory.jacksonFactory.createJsonGenerator(writer);
+ JsonGenerator gen = JsonFactory.jacksonFactory.createGenerator(writer);
gen.writeStartObject();
gen.writeArrayFieldStart("taskmanagers");
http://git-wip-us.apache.org/repos/asf/flink/blob/6f2d9d79/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JobCheckpointsHandlerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JobCheckpointsHandlerTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JobCheckpointsHandlerTest.java
index 1f0f641..b6d50ea 100644
--- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JobCheckpointsHandlerTest.java
+++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JobCheckpointsHandlerTest.java
@@ -18,14 +18,14 @@
package org.apache.flink.runtime.webmonitor.handlers;
-import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.runtime.checkpoint.stats.CheckpointStats;
import org.apache.flink.runtime.checkpoint.stats.CheckpointStatsTracker;
import org.apache.flink.runtime.checkpoint.stats.JobCheckpointStats;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
import org.apache.flink.runtime.webmonitor.ExecutionGraphHolder;
-import org.codehaus.jackson.JsonNode;
-import org.codehaus.jackson.map.ObjectMapper;
+
import org.junit.Test;
import scala.Option;
@@ -44,11 +44,6 @@ import static org.mockito.Mockito.when;
public class JobCheckpointsHandlerTest {
@Test
- public void testIsJsonResponse() throws Exception {
- assertTrue(RequestHandler.JsonResponse.class.isAssignableFrom(JobCheckpointsHandler.class));
- }
-
- @Test
public void testNoCoordinator() throws Exception {
JobCheckpointsHandler handler = new JobCheckpointsHandler(
mock(ExecutionGraphHolder.class));
@@ -150,16 +145,16 @@ public class JobCheckpointsHandlerTest {
JsonNode rootNode = mapper.readTree(response);
// Count
- int count = rootNode.get("count").getIntValue();
+ int count = rootNode.get("count").asInt();
assertEquals(stats.getCount(), count);
// Duration
JsonNode durationNode = rootNode.get("duration");
assertNotNull(durationNode);
- long minDuration = durationNode.get("min").getLongValue();
- long maxDuration = durationNode.get("max").getLongValue();
- long avgDuration = durationNode.get("avg").getLongValue();
+ long minDuration = durationNode.get("min").asLong();
+ long maxDuration = durationNode.get("max").asLong();
+ long avgDuration = durationNode.get("avg").asLong();
assertEquals(stats.getMinDuration(), minDuration);
assertEquals(stats.getMaxDuration(), maxDuration);
@@ -169,9 +164,9 @@ public class JobCheckpointsHandlerTest {
JsonNode sizeNode = rootNode.get("size");
assertNotNull(sizeNode);
- long minSize = sizeNode.get("min").getLongValue();
- long maxSize = sizeNode.get("max").getLongValue();
- long avgSize = sizeNode.get("avg").getLongValue();
+ long minSize = sizeNode.get("min").asLong();
+ long maxSize = sizeNode.get("max").asLong();
+ long avgSize = sizeNode.get("avg").asLong();
assertEquals(stats.getMinStateSize(), minSize);
assertEquals(stats.getMaxStateSize(), maxSize);
@@ -181,17 +176,17 @@ public class JobCheckpointsHandlerTest {
assertNotNull(historyNode);
assertTrue(historyNode.isArray());
- Iterator<JsonNode> it = historyNode.getElements();
+ Iterator<JsonNode> it = historyNode.elements();
for (int i = 0; i < history.size(); i++) {
CheckpointStats s = history.get(i);
JsonNode node = it.next();
- long checkpointId = node.get("id").getLongValue();
- long timestamp = node.get("timestamp").getLongValue();
- long duration = node.get("duration").getLongValue();
- long size = node.get("size").getLongValue();
+ long checkpointId = node.get("id").asLong();
+ long timestamp = node.get("timestamp").asLong();
+ long duration = node.get("duration").asLong();
+ long size = node.get("size").asLong();
assertEquals(s.getCheckpointId(), checkpointId);
assertEquals(s.getTriggerTimestamp(), timestamp);