You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2019/02/19 16:38:07 UTC
[flink] 05/05: [FLINK-11652] Remove legacy JobClient
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d30957a419ab276f2c5937d8ab2d8887b176704c
Author: tison <wa...@gmail.com>
AuthorDate: Mon Feb 18 21:28:58 2019 +0800
[FLINK-11652] Remove legacy JobClient
This closes #7743.
---
.../flink-ml/src/test/resources/logback-test.xml | 1 -
.../src/test/resources/logback-test.xml | 1 -
.../org/apache/flink/runtime/client/JobClient.java | 46 -------------
.../runtime/client/JobExecutionException.java | 3 +-
.../flink/runtime/messages/JobClientMessages.scala | 78 ----------------------
.../yarn/YARNSessionCapacitySchedulerITCase.java | 8 +--
6 files changed, 5 insertions(+), 132 deletions(-)
diff --git a/flink-libraries/flink-ml/src/test/resources/logback-test.xml b/flink-libraries/flink-ml/src/test/resources/logback-test.xml
index 76854c2..482da3f 100644
--- a/flink-libraries/flink-ml/src/test/resources/logback-test.xml
+++ b/flink-libraries/flink-ml/src/test/resources/logback-test.xml
@@ -32,7 +32,6 @@
<!---->
<logger name="org.apache.flink.runtime.operators.DataSinkTask" level="OFF"/>
<logger name="org.apache.flink.runtime.operators.BatchTask" level="OFF"/>
- <logger name="org.apache.flink.runtime.client.JobClient" level="OFF"/>
<logger name="org.apache.flink.runtime.taskmanager.Task" level="OFF"/>
<logger name="org.apache.flink.runtime.testingUtils" level="OFF"/>
<logger name="org.apache.flink.runtime.executiongraph.ExecutionGraph" level="OFF"/>
diff --git a/flink-libraries/flink-python/src/test/resources/logback-test.xml b/flink-libraries/flink-python/src/test/resources/logback-test.xml
index 76854c2..482da3f 100644
--- a/flink-libraries/flink-python/src/test/resources/logback-test.xml
+++ b/flink-libraries/flink-python/src/test/resources/logback-test.xml
@@ -32,7 +32,6 @@
<!---->
<logger name="org.apache.flink.runtime.operators.DataSinkTask" level="OFF"/>
<logger name="org.apache.flink.runtime.operators.BatchTask" level="OFF"/>
- <logger name="org.apache.flink.runtime.client.JobClient" level="OFF"/>
<logger name="org.apache.flink.runtime.taskmanager.Task" level="OFF"/>
<logger name="org.apache.flink.runtime.testingUtils" level="OFF"/>
<logger name="org.apache.flink.runtime.executiongraph.ExecutionGraph" level="OFF"/>
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
deleted file mode 100644
index 2ab012e..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.client;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.clusterframework.BootstrapTools;
-
-import akka.actor.ActorSystem;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * The JobClient bridges between the JobManager's asynchronous actor messages and
- * the synchronous method calls to trigger.
- */
-public class JobClient {
-
- private static final Logger LOG = LoggerFactory.getLogger(JobClient.class);
-
- public static ActorSystem startJobClientActorSystem(Configuration config, String hostname)
- throws Exception {
- LOG.info("Starting JobClient actor system");
-
- // start a remote actor system to listen on an arbitrary port
- ActorSystem system = BootstrapTools.startActorSystem(config, hostname, 0, LOG);
-
- return system;
- }
-
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobExecutionException.java b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobExecutionException.java
index 47eaaf5..030de1d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobExecutionException.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobExecutionException.java
@@ -23,8 +23,7 @@ import org.apache.flink.util.FlinkException;
/**
* This exception is the base exception for all exceptions that denote any failure during
- * the execution of a job. The JobExecutionException and its subclasses are thrown by
- * the {@link JobClient}.
+ * the execution of a job.
*/
public class JobExecutionException extends FlinkException {
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobClientMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobClientMessages.scala
deleted file mode 100644
index 7268b0f..0000000
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobClientMessages.scala
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.messages
-
-import java.util.UUID
-
-import akka.actor.ActorRef
-import org.apache.flink.api.common.JobID
-import org.apache.flink.runtime.jobgraph.JobGraph
-
-/**
- * This object contains the [[org.apache.flink.runtime.client.JobClient]] specific messages
- */
-object JobClientMessages {
-
- /**
- * This message is sent to the JobClientActor (via ask) to submit a job and
- * get a response when the job execution has finished.
- *
- * The response to this message is a
- * [[org.apache.flink.runtime.client.SerializedJobExecutionResult]]
- *
- * @param jobGraph The job to be executed.
- */
- case class SubmitJobAndWait(jobGraph: JobGraph)
-
- /**
- * This message is sent to the JobClientActor to ask it to register at the JobManager
- * and then return once the job execution is complete.
- * @param jobID The job id
- */
- case class AttachToJobAndWait(jobID: JobID)
-
- /** Notifies the JobClientActor about a new leader address and a leader session ID.
- *
- * @param address New leader address
- * @param leaderSessionID New leader session ID
- */
- case class JobManagerLeaderAddress(address: String, leaderSessionID: UUID)
-
- /** Notifies the JobClientActor about the ActorRef of the new leader.
- *
- * @param jobManager ActorRef of the new leader
- */
- case class JobManagerActorRef(jobManager: ActorRef) extends RequiresLeaderSessionID
-
- /** Message which is triggered when the submission timeout has been reached. */
- case object SubmissionTimeout extends RequiresLeaderSessionID
-
- /** Message which is triggered when the JobClient registration at the JobManager times out */
- case object RegistrationTimeout extends RequiresLeaderSessionID
-
- /**
- * Message which is triggered when the connection timeout has been reached.
- *
- * @param id Timeout id which identifies the concurrent timeouts
- */
- case class ConnectionTimeout(id: UUID)
-
- def getSubmissionTimeout(): AnyRef = SubmissionTimeout
- def getRegistrationTimeout(): AnyRef = RegistrationTimeout
-}
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
index 2e68988..d6abc5a 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
@@ -20,11 +20,11 @@ package org.apache.flink.yarn;
import org.apache.flink.api.common.time.Deadline;
import org.apache.flink.api.common.time.Time;
+import org.apache.flink.client.cli.CliFrontend;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.ResourceManagerOptions;
-import org.apache.flink.runtime.client.JobClient;
import org.apache.flink.runtime.rest.RestClient;
import org.apache.flink.runtime.rest.RestClientConfiguration;
import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterOverviewWithVersion;
@@ -167,7 +167,7 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
@Test
public void perJobYarnCluster() throws IOException {
LOG.info("Starting perJobYarnCluster()");
- addTestAppender(JobClient.class, Level.INFO);
+ addTestAppender(CliFrontend.class, Level.INFO);
File exampleJarLocation = getTestJarPath("BatchWordCount.jar");
runWithArgs(new String[]{"run", "-m", "yarn-cluster",
"-yj", flinkUberjar.getAbsolutePath(),
@@ -197,7 +197,7 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
@Test
public void perJobYarnClusterOffHeap() throws IOException {
LOG.info("Starting perJobYarnCluster()");
- addTestAppender(JobClient.class, Level.INFO);
+ addTestAppender(CliFrontend.class, Level.INFO);
File exampleJarLocation = getTestJarPath("BatchWordCount.jar");
// set memory constraints (otherwise this is the same test as perJobYarnCluster() above)
@@ -424,7 +424,7 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
LOG.info("Starting perJobYarnClusterWithParallelism()");
// write log messages to stdout as well, so that the runWithArgs() method
// is catching the log output
- addTestAppender(JobClient.class, Level.INFO);
+ addTestAppender(CliFrontend.class, Level.INFO);
File exampleJarLocation = getTestJarPath("BatchWordCount.jar");
runWithArgs(new String[]{"run",
"-p", "2", //test that the job is executed with a DOP of 2