[FLINK-11853] [rest] Minimal implementation of POST for /jars/:jarid/plan

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebSubmissionExtension.java

@@ -27,8 +27,9 @@
 import org.apache.flink.runtime.webmonitor.handlers.JarDeleteHeaders;
 import org.apache.flink.runtime.webmonitor.handlers.JarListHandler;
 import org.apache.flink.runtime.webmonitor.handlers.JarListHeaders;
+import org.apache.flink.runtime.webmonitor.handlers.JarPlanGetHeaders;
 import org.apache.flink.runtime.webmonitor.handlers.JarPlanHandler;
-import org.apache.flink.runtime.webmonitor.handlers.JarPlanHeaders;
+import org.apache.flink.runtime.webmonitor.handlers.JarPlanPostHeaders;
 import org.apache.flink.runtime.webmonitor.handlers.JarRunHandler;
 import org.apache.flink.runtime.webmonitor.handlers.JarRunHeaders;
 import org.apache.flink.runtime.webmonitor.handlers.JarUploadHandler;
@@ -100,7 +101,17 @@ public WebSubmissionExtension(
 			leaderRetriever,
 			timeout,
 			responseHeaders,
-			JarPlanHeaders.getInstance(),
+			JarPlanGetHeaders.getInstance(),
+			jarDir,
+			configuration,
+			executor
+		);
+
+		final JarPlanHandler postJarPlanHandler = new JarPlanHandler(
+			leaderRetriever,
+			timeout,
+			responseHeaders,
+			JarPlanPostHeaders.getInstance(),
 			jarDir,
 			configuration,
 			executor
@@ -110,7 +121,8 @@ public WebSubmissionExtension(
 		webSubmissionHandlers.add(Tuple2.of(JarListHeaders.getInstance(), jarListHandler));
 		webSubmissionHandlers.add(Tuple2.of(JarRunHeaders.getInstance(), jarRunHandler));
 		webSubmissionHandlers.add(Tuple2.of(JarDeleteHeaders.getInstance(), jarDeleteHandler));
-		webSubmissionHandlers.add(Tuple2.of(JarPlanHeaders.getInstance(), jarPlanHandler));
+		webSubmissionHandlers.add(Tuple2.of(JarPlanGetHeaders.getInstance(), jarPlanHandler));
+		webSubmissionHandlers.add(Tuple2.of(JarPlanGetHeaders.getInstance(), postJarPlanHandler));
 	}
 
 	@Override

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarPlanHeaders.java → flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/AbstractJarPlanHeaders.java

@@ -18,7 +18,6 @@
 
 package org.apache.flink.runtime.webmonitor.handlers;
 
-import org.apache.flink.runtime.rest.HttpMethodWrapper;
 import org.apache.flink.runtime.rest.messages.JobPlanInfo;
 import org.apache.flink.runtime.rest.messages.MessageHeaders;
 
@@ -27,9 +26,7 @@
 /**
  * Message headers for {@link JarPlanHandler}.
  */
-public class JarPlanHeaders implements MessageHeaders<JarPlanRequestBody, JobPlanInfo, JarPlanMessageParameters> {
-
-	private static final JarPlanHeaders INSTANCE = new JarPlanHeaders();
+public abstract class AbstractJarPlanHeaders implements MessageHeaders<JarPlanRequestBody, JobPlanInfo, JarPlanMessageParameters> {
 
 	@Override
 	public Class<JobPlanInfo> getResponseClass() {
@@ -51,20 +48,11 @@ public JarPlanMessageParameters getUnresolvedMessageParameters() {
 		return new JarPlanMessageParameters();
 	}
 
-	@Override
-	public HttpMethodWrapper getHttpMethod() {
-		return HttpMethodWrapper.GET;
-	}
-
 	@Override
 	public String getTargetRestEndpointURL() {
 		return "/jars/:" + JarIdPathParameter.KEY + "/plan";
 	}
 
-	public static JarPlanHeaders getInstance() {
-		return INSTANCE;
-	}
-
 	@Override
 	public String getDescription() {
 		return "Returns the dataflow plan of a job contained in a jar previously uploaded via '" + JarUploadHeaders.URL + "'. " +

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarPlanGetHeaders.java

@@ -0,0 +1,38 @@
+/*
+ * 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.webmonitor.handlers;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+
+/**
+ * Message headers for {@link JarPlanHandler}.
+ */
+public class JarPlanGetHeaders extends AbstractJarPlanHeaders {
+
+	private static final JarPlanGetHeaders INSTANCE = new JarPlanGetHeaders();
+
+	@Override
+	public HttpMethodWrapper getHttpMethod() {
+		return HttpMethodWrapper.GET;
+	}
+
+	public static JarPlanGetHeaders getInstance() {
+		return INSTANCE;
+	}
+}

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarPlanPostHeaders.java

@@ -0,0 +1,38 @@
+/*
+ * 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.webmonitor.handlers;
+
+import org.apache.flink.runtime.rest.HttpMethodWrapper;
+
+/**
+ * Message headers for {@link JarPlanHandler}.
+ */
+public class JarPlanPostHeaders extends AbstractJarPlanHeaders {
+
+	private static final JarPlanPostHeaders INSTANCE = new JarPlanPostHeaders();
+
+	@Override
+	public HttpMethodWrapper getHttpMethod() {
+		return HttpMethodWrapper.POST;
+	}
+
+	public static JarPlanPostHeaders getInstance() {
+		return INSTANCE;
+	}
+}

flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JarPlanHandlerParameterTest.java

@@ -45,7 +45,7 @@ public static void setup() throws Exception {
 			gatewayRetriever,
 			timeout,
 			responseHeaders,
-			JarPlanHeaders.getInstance(),
+			JarPlanGetHeaders.getInstance(),
 			jarDir,
 			new Configuration(),
 			executor,

flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JarSubmissionITCase.java

@@ -126,7 +126,7 @@ private static JarListInfo listJars(JarListHandler handler, RestfulGateway restf
 	}
 
 	private static JobPlanInfo showPlan(JarPlanHandler handler, String jarName, RestfulGateway restfulGateway) throws Exception {
-		JarPlanMessageParameters planParameters = JarPlanHeaders.getInstance().getUnresolvedMessageParameters();
+		JarPlanMessageParameters planParameters = JarPlanGetHeaders.getInstance().getUnresolvedMessageParameters();
 		HandlerRequest<JarPlanRequestBody, JarPlanMessageParameters> planRequest = new HandlerRequest<>(
 			new JarPlanRequestBody(),
 			planParameters,
@@ -195,7 +195,7 @@ private static class JarHandlers {
 				gatewayRetriever,
 				timeout,
 				responseHeaders,
-				JarPlanHeaders.getInstance(),
+				JarPlanGetHeaders.getInstance(),
 				jarDir,
 				new Configuration(),
 				executor);