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 2018/01/12 22:25:27 UTC
[06/14] flink git commit: [FLINK-8347] [flip6] Make cluster id used
by ClusterDescriptor typesafe
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
index b17b030..938b979 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.client.program.rest;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.client.deployment.StandaloneClusterId;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
@@ -115,7 +116,7 @@ public class RestClusterClientTest extends TestLogger {
private RestServerEndpointConfiguration restServerEndpointConfiguration;
- private RestClusterClient restClusterClient;
+ private RestClusterClient<StandaloneClusterId> restClusterClient;
@Before
public void setUp() throws Exception {
@@ -126,7 +127,7 @@ public class RestClusterClientTest extends TestLogger {
config.setString(JobManagerOptions.ADDRESS, "localhost");
restServerEndpointConfiguration = RestServerEndpointConfiguration.fromConfiguration(config);
mockGatewayRetriever = () -> CompletableFuture.completedFuture(mockRestfulGateway);
- restClusterClient = new RestClusterClient(config, (attempt) -> 0);
+ restClusterClient = new RestClusterClient(config, StandaloneClusterId.getInstance(), (attempt) -> 0);
}
@After
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkClient.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkClient.java b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkClient.java
index 24cb0fc..f4ea659 100644
--- a/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkClient.java
+++ b/flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkClient.java
@@ -20,7 +20,6 @@ package org.apache.flink.storm.api;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.JobWithJars;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.client.program.StandaloneClusterClient;
@@ -205,7 +204,7 @@ public class FlinkClient {
configuration.setString(JobManagerOptions.ADDRESS, jobManagerHost);
configuration.setInteger(JobManagerOptions.PORT, jobManagerPort);
- final ClusterClient client;
+ final StandaloneClusterClient client;
try {
client = new StandaloneClusterClient(configuration);
} catch (final Exception e) {
@@ -245,7 +244,7 @@ public class FlinkClient {
configuration.setString(JobManagerOptions.ADDRESS, this.jobManagerHost);
configuration.setInteger(JobManagerOptions.PORT, this.jobManagerPort);
- final ClusterClient client;
+ final StandaloneClusterClient client;
try {
client = new StandaloneClusterClient(configuration);
} catch (final Exception e) {
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcServer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcServer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcServer.java
index 25ce10a..ac2f7eb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcServer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcServer.java
@@ -21,7 +21,7 @@ package org.apache.flink.runtime.rpc;
import java.util.concurrent.CompletableFuture;
/**
- * Interface for self gateways
+ * Interface for self gateways.
*/
public interface RpcServer extends StartStoppable, MainThreadExecutable, RpcGateway {
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala
----------------------------------------------------------------------
diff --git a/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala b/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala
index 349d940..6fa6aa5 100644
--- a/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala
+++ b/flink-scala-shell/src/main/scala/org/apache/flink/api/scala/FlinkShell.scala
@@ -21,6 +21,7 @@ package org.apache.flink.api.scala
import java.io._
import org.apache.flink.client.cli.{CliFrontend, CliFrontendParser, RunOptions}
+import org.apache.flink.client.deployment.{ClusterDescriptor, StandaloneClusterId}
import org.apache.flink.client.program.ClusterClient
import org.apache.flink.configuration.{Configuration, GlobalConfiguration, JobManagerOptions}
import org.apache.flink.runtime.minicluster.StandaloneMiniCluster
@@ -138,7 +139,7 @@ object FlinkShell {
def fetchConnectionInfo(
configuration: Configuration,
config: Config
- ): (String, Int, Option[Either[StandaloneMiniCluster, ClusterClient]]) = {
+ ): (String, Int, Option[Either[StandaloneMiniCluster, ClusterClient[_]]]) = {
config.executionMode match {
case ExecutionMode.LOCAL => // Local mode
val config = GlobalConfiguration.loadConfiguration()
@@ -294,7 +295,9 @@ object FlinkShell {
CliFrontend.loadCustomCommandLines(configuration, configurationDirectory))
val customCLI = frontend.getActiveCustomCommandLine(commandLine)
- val clusterDescriptor = customCLI.createClusterDescriptor(commandLine)
+ val clusterDescriptor = customCLI
+ .createClusterDescriptor(commandLine)
+ .asInstanceOf[ClusterDescriptor[Any]]
val clusterId = customCLI.getClusterId(commandLine)
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
index 74b1c68..a0b4a40 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
@@ -21,7 +21,6 @@ import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.JobWithJars;
import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.client.program.StandaloneClusterClient;
@@ -199,7 +198,7 @@ public class RemoteStreamEnvironment extends StreamExecutionEnvironment {
configuration.setString(JobManagerOptions.ADDRESS, host);
configuration.setInteger(JobManagerOptions.PORT, port);
- ClusterClient client;
+ StandaloneClusterClient client;
try {
client = new StandaloneClusterClient(configuration);
client.setPrintStatusDuringExecution(getConfig().isSysoutLoggingEnabled());
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-tests/src/test/java/org/apache/flink/test/example/client/JobRetrievalITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/example/client/JobRetrievalITCase.java b/flink-tests/src/test/java/org/apache/flink/test/example/client/JobRetrievalITCase.java
index 7f8572b..221f3fa 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/example/client/JobRetrievalITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/example/client/JobRetrievalITCase.java
@@ -20,6 +20,7 @@
package org.apache.flink.test.example.client;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.client.deployment.StandaloneClusterId;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.StandaloneClusterClient;
import org.apache.flink.configuration.Configuration;
@@ -78,7 +79,7 @@ public class JobRetrievalITCase extends TestLogger {
final JobGraph jobGraph = new JobGraph(jobID, "testjob", imalock);
- final ClusterClient client = new StandaloneClusterClient(cluster.configuration(), cluster.highAvailabilityServices());
+ final ClusterClient<StandaloneClusterId> client = new StandaloneClusterClient(cluster.configuration(), cluster.highAvailabilityServices());
// acquire the lock to make sure that the job cannot complete until the job client
// has been attached in resumingThread
@@ -123,7 +124,7 @@ public class JobRetrievalITCase extends TestLogger {
@Test
public void testNonExistingJobRetrieval() throws Exception {
final JobID jobID = new JobID();
- ClusterClient client = new StandaloneClusterClient(cluster.configuration());
+ ClusterClient<StandaloneClusterId> client = new StandaloneClusterClient(cluster.configuration());
try {
client.retrieveJob(jobID);
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
index 897dc8c..55669f1 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
@@ -41,6 +41,7 @@ import akka.actor.PoisonPill;
import akka.testkit.JavaTestKit;
import org.apache.curator.test.TestingServer;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.junit.AfterClass;
import org.junit.Assert;
@@ -125,7 +126,7 @@ public class YARNHighAvailabilityITCase extends YarnTestBase {
"@@" + FsStateBackendFactory.CHECKPOINT_DIRECTORY_URI_CONF_KEY + "=" + fsStateHandlePath + "/checkpoints" +
"@@" + HighAvailabilityOptions.HA_STORAGE_PATH.key() + "=" + fsStateHandlePath + "/recovery");
- ClusterClient yarnCluster = null;
+ ClusterClient<ApplicationId> yarnCluster = null;
final FiniteDuration timeout = new FiniteDuration(2, TimeUnit.MINUTES);
@@ -141,8 +142,6 @@ public class YARNHighAvailabilityITCase extends YarnTestBase {
try {
yarnCluster = flinkYarnClient.deploySessionCluster(clusterSpecification);
- final ClusterClient finalYarnCluster = yarnCluster;
-
highAvailabilityServices = HighAvailabilityServicesUtils.createHighAvailabilityServices(
yarnCluster.getFlinkConfiguration(),
Executors.directExecutor(),
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
index 15fe355..ec6c105 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
@@ -246,7 +246,7 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
.setSlotsPerTaskManager(1)
.createClusterSpecification();
// deploy
- ClusterClient yarnCluster = null;
+ ClusterClient<ApplicationId> yarnCluster = null;
try {
yarnCluster = clusterDescriptor.deploySessionCluster(clusterSpecification);
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
index 4affb78..a7e8c36 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java
@@ -19,9 +19,10 @@
package org.apache.flink.yarn;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.client.deployment.ClusterDeploymentException;
import org.apache.flink.client.deployment.ClusterDescriptor;
+import org.apache.flink.client.deployment.ClusterRetrieveException;
import org.apache.flink.client.deployment.ClusterSpecification;
-import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
@@ -61,7 +62,6 @@ import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.client.api.YarnClientApplication;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
-import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -101,7 +101,7 @@ import static org.apache.flink.yarn.cli.FlinkYarnSessionCli.getDynamicProperties
/**
* The descriptor with deployment information for spawning or resuming a {@link YarnClusterClient}.
*/
-public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor {
+public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
private static final Logger LOG = LoggerFactory.getLogger(AbstractYarnClusterDescriptor.class);
/**
@@ -332,7 +332,7 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
// -------------------------------------------------------------
@Override
- public ClusterClient retrieve(String applicationID) {
+ public YarnClusterClient retrieve(ApplicationId applicationId) throws ClusterRetrieveException {
try {
// check if required Hadoop environment variables are set. If not, warn user
@@ -343,18 +343,17 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
"configuration for accessing YARN.");
}
- final ApplicationId yarnAppId = ConverterUtils.toApplicationId(applicationID);
- final ApplicationReport appReport = yarnClient.getApplicationReport(yarnAppId);
+ final ApplicationReport appReport = yarnClient.getApplicationReport(applicationId);
if (appReport.getFinalApplicationStatus() != FinalApplicationStatus.UNDEFINED) {
// Flink cluster is not running anymore
LOG.error("The application {} doesn't run anymore. It has previously completed with final status: {}",
- applicationID, appReport.getFinalApplicationStatus());
- throw new RuntimeException("The Yarn application " + applicationID + " doesn't run anymore.");
+ applicationId, appReport.getFinalApplicationStatus());
+ throw new RuntimeException("The Yarn application " + applicationId + " doesn't run anymore.");
}
LOG.info("Found application JobManager host name '{}' and port '{}' from supplied application id '{}'",
- appReport.getHost(), appReport.getRpcPort(), applicationID);
+ appReport.getHost(), appReport.getRpcPort(), applicationId);
flinkConfiguration.setString(JobManagerOptions.ADDRESS, appReport.getHost());
flinkConfiguration.setInteger(JobManagerOptions.PORT, appReport.getRpcPort());
@@ -367,43 +366,35 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
flinkConfiguration,
false);
} catch (Exception e) {
- throw new RuntimeException("Couldn't retrieve Yarn cluster", e);
+ throw new ClusterRetrieveException("Couldn't retrieve Yarn cluster", e);
}
}
@Override
- public YarnClusterClient deploySessionCluster(ClusterSpecification clusterSpecification) {
+ public YarnClusterClient deploySessionCluster(ClusterSpecification clusterSpecification) throws ClusterDeploymentException {
try {
return deployInternal(
clusterSpecification,
getYarnSessionClusterEntrypoint(),
null);
} catch (Exception e) {
- throw new RuntimeException("Couldn't deploy Yarn session cluster", e);
+ throw new ClusterDeploymentException("Couldn't deploy Yarn session cluster", e);
}
}
@Override
- public YarnClusterClient deployJobCluster(ClusterSpecification clusterSpecification, JobGraph jobGraph) {
+ public YarnClusterClient deployJobCluster(ClusterSpecification clusterSpecification, JobGraph jobGraph) throws ClusterDeploymentException {
try {
return deployInternal(
clusterSpecification,
getYarnJobClusterEntrypoint(),
jobGraph);
} catch (Exception e) {
- throw new RuntimeException("Could not deploy Yarn job cluster.", e);
+ throw new ClusterDeploymentException("Could not deploy Yarn job cluster.", e);
}
}
@Override
- public void terminateCluster(String clusterId) throws FlinkException {
- try {
- yarnClient.killApplication(ConverterUtils.toApplicationId(clusterId));
- } catch (IOException | YarnException e) {
- throw new FlinkException("Could not terminate cluster with id " + clusterId + '.', e);
- }
- }
-
public void terminateCluster(ApplicationId applicationId) throws FlinkException {
try {
yarnClient.killApplication(applicationId);
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
index 991b3b9..f2be264 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
@@ -41,7 +41,6 @@ import akka.util.Timeout;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.exceptions.YarnException;
-import org.apache.hadoop.yarn.util.ConverterUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,7 +57,7 @@ import scala.concurrent.duration.FiniteDuration;
/**
* Java representation of a running Flink cluster within YARN.
*/
-public class YarnClusterClient extends ClusterClient {
+public class YarnClusterClient extends ClusterClient<ApplicationId> {
private static final Logger LOG = LoggerFactory.getLogger(YarnClusterClient.class);
@@ -170,11 +169,6 @@ public class YarnClusterClient extends ClusterClient {
}
}
- @Override
- public String getClusterIdentifier() {
- return ConverterUtils.toString(appReport.getApplicationId());
- }
-
/**
* This method is only available if the cluster hasn't been started in detached mode.
*/
@@ -234,6 +228,11 @@ public class YarnClusterClient extends ClusterClient {
}
@Override
+ public ApplicationId getClusterId() {
+ return appId;
+ }
+
+ @Override
public boolean isDetached() {
return super.isDetached() || clusterDescriptor.isDetachedMode();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
index 24a422e..fea8a5c 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
@@ -86,7 +86,7 @@ import static org.apache.flink.configuration.HighAvailabilityOptions.HA_CLUSTER_
/**
* Class handling the command line interface to the YARN session.
*/
-public class FlinkYarnSessionCli extends AbstractCustomCommandLine {
+public class FlinkYarnSessionCli extends AbstractCustomCommandLine<ApplicationId> {
private static final Logger LOG = LoggerFactory.getLogger(FlinkYarnSessionCli.class);
//------------------------------------ Constants -------------------------
@@ -152,7 +152,7 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine {
private final Properties yarnPropertiesFile;
- private final String yarnApplicationIdFromYarnProperties;
+ private final ApplicationId yarnApplicationIdFromYarnProperties;
private final String yarnPropertiesFileLocation;
@@ -235,20 +235,20 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine {
". Please delete the file at " + yarnPropertiesLocation.getAbsolutePath() + '.', ioe);
}
- yarnApplicationIdFromYarnProperties = yarnPropertiesFile.getProperty(YARN_APPLICATION_ID_KEY);
+ final String yarnApplicationIdString = yarnPropertiesFile.getProperty(YARN_APPLICATION_ID_KEY);
- if (yarnApplicationIdFromYarnProperties == null) {
+ if (yarnApplicationIdString == null) {
throw new FlinkException("Yarn properties file found but doesn't contain a " +
"Yarn application id. Please delete the file at " + yarnPropertiesLocation.getAbsolutePath());
}
try {
// try converting id to ApplicationId
- ConverterUtils.toApplicationId(yarnApplicationIdFromYarnProperties);
+ yarnApplicationIdFromYarnProperties = ConverterUtils.toApplicationId(yarnApplicationIdString);
}
catch (Exception e) {
throw new FlinkException("YARN properties contains an invalid entry for " +
- "application id: " + yarnApplicationIdFromYarnProperties + ". Please delete the file at " +
+ "application id: " + yarnApplicationIdString + ". Please delete the file at " +
yarnPropertiesLocation.getAbsolutePath(), e);
}
} else {
@@ -458,9 +458,9 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine {
@Override
@Nullable
- public String getClusterId(CommandLine commandLine) {
+ public ApplicationId getClusterId(CommandLine commandLine) {
if (commandLine.hasOption(applicationId.getOpt())) {
- return commandLine.getOptionValue(applicationId.getOpt());
+ return ConverterUtils.toApplicationId(commandLine.getOptionValue(applicationId.getOpt()));
} else if (isYarnPropertiesFileMode(commandLine)) {
return yarnApplicationIdFromYarnProperties;
} else {
@@ -485,14 +485,14 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine {
effectiveConfiguration.setString(HA_CLUSTER_ID, zkNamespace);
}
- final String applicationId = getClusterId(commandLine);
+ final ApplicationId applicationId = getClusterId(commandLine);
if (applicationId != null) {
final String zooKeeperNamespace;
if (commandLine.hasOption(zookeeperNamespace.getOpt())){
zooKeeperNamespace = commandLine.getOptionValue(zookeeperNamespace.getOpt());
} else {
- zooKeeperNamespace = effectiveConfiguration.getString(HA_CLUSTER_ID, applicationId);
+ zooKeeperNamespace = effectiveConfiguration.getString(HA_CLUSTER_ID, applicationId.toString());
}
effectiveConfiguration.setString(HA_CLUSTER_ID, zooKeeperNamespace);
@@ -581,20 +581,20 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine {
System.out.println(description);
return 0;
} else {
- final ClusterClient clusterClient;
+ final ClusterClient<ApplicationId> clusterClient;
final ApplicationId yarnApplicationId;
if (cmd.hasOption(applicationId.getOpt())) {
yarnApplicationId = ConverterUtils.toApplicationId(cmd.getOptionValue(applicationId.getOpt()));
- clusterClient = yarnClusterDescriptor.retrieve(cmd.getOptionValue(applicationId.getOpt()));
+ clusterClient = yarnClusterDescriptor.retrieve(yarnApplicationId);
} else {
final ClusterSpecification clusterSpecification = getClusterSpecification(cmd);
clusterClient = yarnClusterDescriptor.deploySessionCluster(clusterSpecification);
//------------------ ClusterClient deployed, handle connection details
- yarnApplicationId = ConverterUtils.toApplicationId(clusterClient.getClusterIdentifier());
+ yarnApplicationId = clusterClient.getClusterId();
String jobManagerAddress =
clusterClient.getJobManagerAddress().getAddress().getHostName() +
@@ -790,7 +790,7 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine {
}
private static void runInteractiveCli(
- ClusterClient clusterClient,
+ ClusterClient<?> clusterClient,
YarnApplicationStatusMonitor yarnApplicationStatusMonitor,
boolean readConsoleInput) {
try (BufferedReader in = new BufferedReader(new InputStreamReader(System.in))) {
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-yarn/src/test/java/org/apache/flink/yarn/AbstractYarnClusterTest.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/AbstractYarnClusterTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/AbstractYarnClusterTest.java
index 2581e68..4e68612 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/AbstractYarnClusterTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/AbstractYarnClusterTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.yarn;
+import org.apache.flink.client.deployment.ClusterRetrieveException;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.TestLogger;
@@ -49,10 +50,9 @@ public class AbstractYarnClusterTest extends TestLogger {
/**
* Tests that the cluster retrieval of a finished YARN application fails.
*/
- @Test(expected = RuntimeException.class)
- public void testClusterClientRetrievalOfFinishedYarnApplication() throws IOException {
+ @Test(expected = ClusterRetrieveException.class)
+ public void testClusterClientRetrievalOfFinishedYarnApplication() throws Exception {
final ApplicationId applicationId = ApplicationId.newInstance(System.currentTimeMillis(), 42);
- final String clusterId = applicationId.toString();
final ApplicationReport applicationReport = createApplicationReport(
applicationId,
YarnApplicationState.FINISHED,
@@ -65,30 +65,7 @@ public class AbstractYarnClusterTest extends TestLogger {
temporaryFolder.newFolder().getAbsolutePath(),
yarnClient);
- clusterDescriptor.retrieve(clusterId);
- }
-
- /**
- * Tests that the cluster retrieval fails if an invalid application id is provided.
- */
- @Test(expected = RuntimeException.class)
- public void testClusterClientRetrievalFromInvalidApplicationId() throws IOException {
- final ApplicationId applicationId = ApplicationId.newInstance(System.currentTimeMillis(), 42);
- final String clusterId = "foobar";
-
- final ApplicationReport applicationReport = createApplicationReport(
- applicationId,
- YarnApplicationState.RUNNING,
- FinalApplicationStatus.UNDEFINED);
-
- final YarnClient yarnClient = new TestingYarnClient(Collections.singletonMap(applicationId, applicationReport));
-
- final TestingAbstractYarnClusterDescriptor clusterDescriptor = new TestingAbstractYarnClusterDescriptor(
- new Configuration(),
- temporaryFolder.newFolder().getAbsolutePath(),
- yarnClient);
-
- clusterDescriptor.retrieve(clusterId);
+ clusterDescriptor.retrieve(applicationId);
}
private ApplicationReport createApplicationReport(
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
index 2f6c157..9be9cb5 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
@@ -179,9 +179,9 @@ public class FlinkYarnSessionCliTest extends TestLogger {
final CommandLine commandLine = flinkYarnSessionCli.parseCommandLineOptions(new String[] {}, true);
- final String clusterId = flinkYarnSessionCli.getClusterId(commandLine);
+ final ApplicationId clusterId = flinkYarnSessionCli.getClusterId(commandLine);
- assertEquals(TEST_YARN_APPLICATION_ID.toString(), clusterId);
+ assertEquals(TEST_YARN_APPLICATION_ID, clusterId);
}
/**
@@ -214,9 +214,9 @@ public class FlinkYarnSessionCliTest extends TestLogger {
final CommandLine commandLine = flinkYarnSessionCli.parseCommandLineOptions(new String[] {"-yid", TEST_YARN_APPLICATION_ID.toString()}, true);
- final String clusterId = flinkYarnSessionCli.getClusterId(commandLine);
+ final ApplicationId clusterId = flinkYarnSessionCli.getClusterId(commandLine);
- assertEquals(TEST_YARN_APPLICATION_ID.toString(), clusterId);
+ assertEquals(TEST_YARN_APPLICATION_ID, clusterId);
}
@Test
@@ -272,8 +272,8 @@ public class FlinkYarnSessionCliTest extends TestLogger {
"y",
"yarn");
final CommandLine commandLine = flinkYarnSessionCli.parseCommandLineOptions(new String[] {"-yid", TEST_YARN_APPLICATION_ID_2.toString() }, true);
- final String clusterId = flinkYarnSessionCli.getClusterId(commandLine);
- assertEquals(TEST_YARN_APPLICATION_ID_2.toString(), clusterId);
+ final ApplicationId clusterId = flinkYarnSessionCli.getClusterId(commandLine);
+ assertEquals(TEST_YARN_APPLICATION_ID_2, clusterId);
}
///////////
http://git-wip-us.apache.org/repos/asf/flink/blob/38d37208/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
index 5be54b0..e0f69cb 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.yarn;
+import org.apache.flink.client.deployment.ClusterDeploymentException;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
@@ -66,7 +67,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
}
@Test
- public void testFailIfTaskSlotsHigherThanMaxVcores() {
+ public void testFailIfTaskSlotsHigherThanMaxVcores() throws ClusterDeploymentException {
final YarnClient yarnClient = YarnClient.createYarnClient();
@@ -88,7 +89,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
clusterDescriptor.deploySessionCluster(clusterSpecification);
fail("The deploy call should have failed.");
- } catch (RuntimeException e) {
+ } catch (ClusterDeploymentException e) {
// we expect the cause to be an IllegalConfigurationException
if (!(e.getCause() instanceof IllegalConfigurationException)) {
throw e;
@@ -99,7 +100,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
}
@Test
- public void testConfigOverwrite() {
+ public void testConfigOverwrite() throws ClusterDeploymentException {
Configuration configuration = new Configuration();
// overwrite vcores in config
configuration.setInteger(YarnConfigOptions.VCORES, Integer.MAX_VALUE);
@@ -125,7 +126,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
clusterDescriptor.deploySessionCluster(clusterSpecification);
fail("The deploy call should have failed.");
- } catch (RuntimeException e) {
+ } catch (ClusterDeploymentException e) {
// we expect the cause to be an IllegalConfigurationException
if (!(e.getCause() instanceof IllegalConfigurationException)) {
throw e;