[FLINK-7647] [flip6] Rename JobManagerConfigHandler to ClusterConfigHandler

The original naming wouldn't make sense for the FLIP-6 redesign, since
we would have multiple per-job JobManagers for each cluster, which
shares the same configuration.

The REST path is still left untouched and not part of this commit, as
that would involve more changes in flink-runtime-web.

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

@@ -28,6 +28,7 @@ import org.apache.flink.runtime.jobmaster.JobManagerGateway;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
 import org.apache.flink.runtime.net.SSLUtils;
 import org.apache.flink.runtime.rest.handler.WebHandler;
+import org.apache.flink.runtime.rest.handler.legacy.ClusterConfigHandler;
 import org.apache.flink.runtime.rest.handler.legacy.ClusterOverviewHandler;
 import org.apache.flink.runtime.rest.handler.legacy.ConstantTextHandler;
 import org.apache.flink.runtime.rest.handler.legacy.CurrentJobIdsHandler;
@@ -40,7 +41,6 @@ import org.apache.flink.runtime.rest.handler.legacy.JobCancellationWithSavepoint
 import org.apache.flink.runtime.rest.handler.legacy.JobConfigHandler;
 import org.apache.flink.runtime.rest.handler.legacy.JobDetailsHandler;
 import org.apache.flink.runtime.rest.handler.legacy.JobExceptionsHandler;
-import org.apache.flink.runtime.rest.handler.legacy.JobManagerConfigHandler;
 import org.apache.flink.runtime.rest.handler.legacy.JobPlanHandler;
 import org.apache.flink.runtime.rest.handler.legacy.JobStoppingHandler;
 import org.apache.flink.runtime.rest.handler.legacy.JobVertexAccumulatorsHandler;
@@ -244,7 +244,7 @@ public class WebRuntimeMonitor implements WebMonitor {
 		get(router, new ClusterOverviewHandler(executor, DEFAULT_REQUEST_TIMEOUT));
 
 		// job manager configuration
-		get(router, new JobManagerConfigHandler(executor, config));
+		get(router, new ClusterConfigHandler(executor, config));
 
 		// overview over jobs
 		get(router, new CurrentJobsOverviewHandler(executor, DEFAULT_REQUEST_TIMEOUT, true, true));

flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java

@@ -27,10 +27,10 @@ import org.apache.flink.runtime.rest.RestServerEndpointConfiguration;
 import org.apache.flink.runtime.rest.handler.LegacyRestHandlerAdapter;
 import org.apache.flink.runtime.rest.handler.RestHandlerConfiguration;
 import org.apache.flink.runtime.rest.handler.RestHandlerSpecification;
+import org.apache.flink.runtime.rest.handler.legacy.ClusterConfigHandler;
 import org.apache.flink.runtime.rest.handler.legacy.ClusterOverviewHandler;
 import org.apache.flink.runtime.rest.handler.legacy.CurrentJobsOverviewHandler;
 import org.apache.flink.runtime.rest.handler.legacy.DashboardConfigHandler;
-import org.apache.flink.runtime.rest.handler.legacy.JobManagerConfigHandler;
 import org.apache.flink.runtime.rest.handler.legacy.files.StaticFileServerHandler;
 import org.apache.flink.runtime.rest.handler.legacy.files.WebContentHandlerSpecification;
 import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfo;
@@ -119,7 +119,7 @@ public class DispatcherRestEndpoint extends RestServerEndpoint {
 			leaderRetriever,
 			timeout,
 			ClusterConfigurationInfoHeaders.getInstance(),
-			new JobManagerConfigHandler(
+			new ClusterConfigHandler(
 				executor,
 				clusterConfiguration));
 

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandler.java → flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandler.java

@@ -43,13 +43,13 @@ import java.util.concurrent.Executor;
 /**
  * Returns the Job Manager's configuration.
  */
-public class JobManagerConfigHandler extends AbstractJsonRequestHandler
+public class ClusterConfigHandler extends AbstractJsonRequestHandler
 		implements LegacyRestHandler<DispatcherGateway, ClusterConfigurationInfo, EmptyMessageParameters> {
 
 	private final ClusterConfigurationInfo clusterConfig;
 	private final String clusterConfigJson;
 
-	public JobManagerConfigHandler(Executor executor, Configuration config) {
+	public ClusterConfigHandler(Executor executor, Configuration config) {
 		super(executor);
 
 		Preconditions.checkNotNull(config);

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java

@@ -19,13 +19,13 @@
 package org.apache.flink.runtime.rest.handler.legacy.messages;
 
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.rest.handler.legacy.JobManagerConfigHandler;
+import org.apache.flink.runtime.rest.handler.legacy.ClusterConfigHandler;
 import org.apache.flink.runtime.rest.messages.ResponseBody;
 
 import java.util.ArrayList;
 
 /**
- * Response of the {@link JobManagerConfigHandler}, respresented as a list
+ * Response of the {@link ClusterConfigHandler}, respresented as a list
  * of key-value pairs of the cluster {@link Configuration}.
  */
 public class ClusterConfigurationInfo extends ArrayList<ClusterConfigurationInfoEntry> implements ResponseBody {

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java

@@ -19,18 +19,20 @@
 package org.apache.flink.runtime.rest.messages;
 
 import org.apache.flink.runtime.rest.HttpMethodWrapper;
-import org.apache.flink.runtime.rest.handler.legacy.JobManagerConfigHandler;
+import org.apache.flink.runtime.rest.handler.legacy.ClusterConfigHandler;
 import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfo;
 
 import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
 
 /**
- * Message headers for the {@link JobManagerConfigHandler}.
+ * Message headers for the {@link ClusterConfigHandler}.
  */
 public final class ClusterConfigurationInfoHeaders implements MessageHeaders<EmptyRequestBody, ClusterConfigurationInfo, EmptyMessageParameters> {
 
 	private static final ClusterConfigurationInfoHeaders INSTANCE = new ClusterConfigurationInfoHeaders();
 
+	// TODO this REST path is inappropriately set due to legacy design reasons, and ideally should be '/config';
+	// TODO changing it would require corresponding path changes in flink-runtime-web
 	public static final String CLUSTER_CONFIG_REST_PATH = "/jobmanager/config";
 
 	private ClusterConfigurationInfoHeaders() {}

flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/JobManagerConfigHandlerTest.java → flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandlerTest.java

@@ -25,12 +25,12 @@ import org.junit.Assert;
 import org.junit.Test;
 
 /**
- * Tests for the JobManagerConfigHandler.
+ * Tests for the ClusterConfigHandler.
  */
-public class JobManagerConfigHandlerTest {
+public class ClusterConfigHandlerTest {
 	@Test
 	public void testGetPaths() {
-		JobManagerConfigHandler handler = new JobManagerConfigHandler(Executors.directExecutor(), new Configuration());
+		ClusterConfigHandler handler = new ClusterConfigHandler(Executors.directExecutor(), new Configuration());
 		String[] paths = handler.getPaths();
 		Assert.assertEquals(1, paths.length);
 		Assert.assertEquals("/jobmanager/config", paths[0]);