[FLINK-8840] [yarn] Pull YarnClient and YarnConfiguration instantiati…

…on out of AbstractYarnClusterClient

For better testability, this commit moves the YarnClient and YarnConfiguration out of
the AbstractYarnClusterDescriptor.

flink-yarn-tests/src/test/java/org/apache/flink/yarn/TestingYarnClusterDescriptor.java

@@ -24,6 +24,7 @@
 import org.apache.flink.util.Preconditions;
 
 import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
 
 import java.io.File;
 import java.io.FilenameFilter;
@@ -37,11 +38,18 @@
  */
 public class TestingYarnClusterDescriptor extends YarnClusterDescriptor {
 
-	public TestingYarnClusterDescriptor(Configuration configuration, String configurationDirectory) {
+	public TestingYarnClusterDescriptor(
+			Configuration configuration,
+			YarnConfiguration yarnConfiguration,
+			String configurationDirectory,
+			YarnClient yarnClient,
+			boolean sharedYarnClient) {
 		super(
 			configuration,
+			yarnConfiguration,
 			configurationDirectory,
-			YarnClient.createYarnClient());
+			yarnClient,
+			sharedYarnClient);
 		List<File> filesToShip = new ArrayList<>();
 
 		File testingJar = YarnTestBase.findFile("..", new TestJarFinder("flink-yarn-tests"));

flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java

@@ -110,7 +110,12 @@ public void testMultipleAMKill() throws Exception {
 		final int numberKillingAttempts = numberApplicationAttempts - 1;
 		String confDirPath = System.getenv(ConfigConstants.ENV_FLINK_CONF_DIR);
 		final Configuration configuration = GlobalConfiguration.loadConfiguration();
-		TestingYarnClusterDescriptor flinkYarnClient = new TestingYarnClusterDescriptor(configuration, confDirPath);
+		TestingYarnClusterDescriptor flinkYarnClient = new TestingYarnClusterDescriptor(
+			configuration,
+			getYarnConfiguration(),
+			confDirPath,
+			getYarnClient(),
+			true);
 
 		Assert.assertNotNull("unable to get yarn client", flinkYarnClient);
 		flinkYarnClient.setLocalJarPath(new Path(flinkUberjar.getAbsolutePath()));

flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java

@@ -55,12 +55,14 @@ public static void setup() {
 	public void testPerJobMode() throws Exception {
 		Configuration configuration = new Configuration();
 		configuration.setString(AkkaOptions.ASK_TIMEOUT, "30 s");
-		final YarnClient yarnClient = YarnClient.createYarnClient();
+		final YarnClient yarnClient = getYarnClient();
 
 		try (final Flip6YarnClusterDescriptor flip6YarnClusterDescriptor = new Flip6YarnClusterDescriptor(
 			configuration,
+			getYarnConfiguration(),
 			System.getenv(ConfigConstants.ENV_FLINK_CONF_DIR),
-			yarnClient)) {
+			yarnClient,
+			true)) {
 
 			flip6YarnClusterDescriptor.setLocalJarPath(new Path(flinkUberjar.getAbsolutePath()));
 			flip6YarnClusterDescriptor.addShipFiles(Arrays.asList(flinkLibFolder.listFiles()));

flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java

@@ -231,12 +231,13 @@ public void testJavaAPI() throws Exception {
 
 		String confDirPath = System.getenv(ConfigConstants.ENV_FLINK_CONF_DIR);
 		Configuration configuration = GlobalConfiguration.loadConfiguration();
-		final YarnClient yarnClient = YarnClient.createYarnClient();
 
 		try (final AbstractYarnClusterDescriptor clusterDescriptor = new YarnClusterDescriptor(
 			configuration,
+			getYarnConfiguration(),
 			confDirPath,
-			yarnClient)) {
+			getYarnClient(),
+			true)) {
 			Assert.assertNotNull("unable to get yarn client", clusterDescriptor);
 			clusterDescriptor.setLocalJarPath(new Path(flinkUberjar.getAbsolutePath()));
 			clusterDescriptor.addShipFiles(Arrays.asList(flinkLibFolder.listFiles()));

flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java

@@ -128,7 +128,7 @@ public abstract class YarnTestBase extends TestLogger {
 	 */
 	protected static File flinkUberjar;
 
-	protected static final Configuration YARN_CONFIGURATION;
+	protected static final YarnConfiguration YARN_CONFIGURATION;
 
 	/**
 	 * lib/ folder of the flink distribution.
@@ -213,6 +213,14 @@ public void checkClusterEmpty() throws IOException, YarnException {
 		flip6 = CoreOptions.FLIP6_MODE.equalsIgnoreCase(flinkConfiguration.getString(CoreOptions.MODE));
 	}
 
+	protected YarnClient getYarnClient() {
+		return yarnClient;
+	}
+
+	protected static YarnConfiguration getYarnConfiguration() {
+		return YARN_CONFIGURATION;
+	}
+
 	/**
 	 * Locate a file or directory.
 	 */

flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java

@@ -118,6 +118,9 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
 
 	private final YarnClient yarnClient;
 
+	/** True if the descriptor must not shut down the YarnClient. */
+	private final boolean sharedYarnClient;
+
 	private String yarnQueue;
 
 	private String configurationDirectory;
@@ -145,10 +148,12 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor
 
 	public AbstractYarnClusterDescriptor(
 			Configuration flinkConfiguration,
+			YarnConfiguration yarnConfiguration,
 			String configurationDirectory,
-			YarnClient yarnClient) {
+			YarnClient yarnClient,
+			boolean sharedYarnClient) {
 
-		yarnConfiguration = new YarnConfiguration();
+		this.yarnConfiguration = Preconditions.checkNotNull(yarnConfiguration);
 
 		// for unit tests only
 		if (System.getenv("IN_TESTS") != null) {
@@ -160,8 +165,7 @@ public AbstractYarnClusterDescriptor(
 		}
 
 		this.yarnClient = Preconditions.checkNotNull(yarnClient);
-		yarnClient.init(yarnConfiguration);
-		yarnClient.start();
+		this.sharedYarnClient = sharedYarnClient;
 
 		this.flinkConfiguration = Preconditions.checkNotNull(flinkConfiguration);
 		userJarInclusion = getUserJarInclusionMode(flinkConfiguration);
@@ -328,7 +332,9 @@ public void setZookeeperNamespace(String zookeeperNamespace) {
 
 	@Override
 	public void close() {
-		yarnClient.stop();
+		if (!sharedYarnClient) {
+			yarnClient.stop();
+		}
 	}
 
 	// -------------------------------------------------------------

flink-yarn/src/main/java/org/apache/flink/yarn/Flip6YarnClusterDescriptor.java

@@ -30,6 +30,7 @@
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
 
 /**
  * Implementation of {@link org.apache.flink.yarn.AbstractYarnClusterDescriptor} which is used to start the
@@ -39,9 +40,16 @@ public class Flip6YarnClusterDescriptor extends AbstractYarnClusterDescriptor {
 
 	public Flip6YarnClusterDescriptor(
 			Configuration flinkConfiguration,
+			YarnConfiguration yarnConfiguration,
 			String configurationDirectory,
-			YarnClient yarnCLient) {
-		super(flinkConfiguration, configurationDirectory, yarnCLient);
+			YarnClient yarnClient,
+			boolean sharedYarnClient) {
+		super(
+			flinkConfiguration,
+			yarnConfiguration,
+			configurationDirectory,
+			yarnClient,
+			sharedYarnClient);
 	}
 
 	@Override

flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java

@@ -26,6 +26,7 @@
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
 
 /**
  * Default implementation of {@link AbstractYarnClusterDescriptor} which starts an {@link YarnApplicationMasterRunner}.
@@ -34,9 +35,16 @@ public class YarnClusterDescriptor extends AbstractYarnClusterDescriptor {
 
 	public YarnClusterDescriptor(
 			Configuration flinkConfiguration,
+			YarnConfiguration yarnConfiguration,
 			String configurationDirectory,
-			YarnClient yarnClient) {
-		super(flinkConfiguration, configurationDirectory, yarnClient);
+			YarnClient yarnClient,
+			boolean sharedYarnClient) {
+		super(
+			flinkConfiguration,
+			yarnConfiguration,
+			configurationDirectory,
+			yarnClient,
+			sharedYarnClient);
 	}
 
 	@Override

flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java

@@ -54,6 +54,7 @@
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.api.records.YarnApplicationState;
 import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.util.ConverterUtils;
 import org.slf4j.Logger;
@@ -159,6 +160,8 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine<ApplicationId
 
 	private final boolean flip6;
 
+	private final YarnConfiguration yarnConfiguration;
+
 	//------------------------------------ Internal fields -------------------------
 	private boolean detachedMode = false;
 
@@ -257,16 +260,20 @@ public FlinkYarnSessionCli(
 		} else {
 			yarnApplicationIdFromYarnProperties = null;
 		}
+
+		this.yarnConfiguration = new YarnConfiguration();
 	}
 
 	private AbstractYarnClusterDescriptor createDescriptor(
 		Configuration configuration,
+		YarnConfiguration yarnConfiguration,
 		String configurationDirectory,
 		String defaultApplicationName,
 		CommandLine cmd) {
 
 		AbstractYarnClusterDescriptor yarnClusterDescriptor = getClusterDescriptor(
 			configuration,
+			yarnConfiguration,
 			configurationDirectory);
 
 		// Jar Path
@@ -440,6 +447,7 @@ public AbstractYarnClusterDescriptor createClusterDescriptor(CommandLine command
 
 		return createDescriptor(
 			effectiveConfiguration,
+			yarnConfiguration,
 			configurationDirectory,
 			null,
 			commandLine);
@@ -955,12 +963,28 @@ public static File getYarnPropertiesLocation(@Nullable String yarnPropertiesFile
 		return new File(propertiesFileLocation, YARN_PROPERTIES_FILE + currentUser);
 	}
 
-	private AbstractYarnClusterDescriptor getClusterDescriptor(Configuration configuration, String configurationDirectory) {
+	private AbstractYarnClusterDescriptor getClusterDescriptor(
+			Configuration configuration,
+			YarnConfiguration yarnConfiguration,
+			String configurationDirectory) {
 		final YarnClient yarnClient = YarnClient.createYarnClient();
+		yarnClient.init(yarnConfiguration);
+		yarnClient.start();
+
 		if (flip6) {
-			return new Flip6YarnClusterDescriptor(configuration, configurationDirectory, yarnClient);
+			return new Flip6YarnClusterDescriptor(
+				configuration,
+				yarnConfiguration,
+				configurationDirectory,
+				yarnClient,
+				false);
 		} else {
-			return new YarnClusterDescriptor(configuration, configurationDirectory, yarnClient);
+			return new YarnClusterDescriptor(
+				configuration,
+				yarnConfiguration,
+				configurationDirectory,
+				yarnClient,
+				false);
 		}
 	}
 }

flink-yarn/src/test/java/org/apache/flink/yarn/AbstractYarnClusterTest.java

@@ -34,6 +34,7 @@
 import org.apache.hadoop.yarn.api.records.YarnApplicationState;
 import org.apache.hadoop.yarn.client.api.YarnClient;
 import org.apache.hadoop.yarn.client.api.impl.YarnClientImpl;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.junit.Rule;
 import org.junit.Test;
@@ -63,13 +64,22 @@ public void testClusterClientRetrievalOfFinishedYarnApplication() throws Excepti
 			FinalApplicationStatus.SUCCEEDED);
 
 		final YarnClient yarnClient = new TestingYarnClient(Collections.singletonMap(applicationId, applicationReport));
+		final YarnConfiguration yarnConfiguration = new YarnConfiguration();
+		yarnClient.init(yarnConfiguration);
+		yarnClient.start();
 
 		final TestingAbstractYarnClusterDescriptor clusterDescriptor = new TestingAbstractYarnClusterDescriptor(
 			new Configuration(),
+			yarnConfiguration,
 			temporaryFolder.newFolder().getAbsolutePath(),
-			yarnClient);
+			yarnClient,
+			false);
 
-		clusterDescriptor.retrieve(applicationId);
+		try {
+			clusterDescriptor.retrieve(applicationId);
+		} finally {
+			clusterDescriptor.close();
+		}
 	}
 
 	private ApplicationReport createApplicationReport(
@@ -121,9 +131,11 @@ private static final class TestingAbstractYarnClusterDescriptor extends Abstract
 
 		private TestingAbstractYarnClusterDescriptor(
 				Configuration flinkConfiguration,
+				YarnConfiguration yarnConfiguration,
 				String configurationDirectory,
-				YarnClient yarnClient) {
-			super(flinkConfiguration, configurationDirectory, yarnClient);
+				YarnClient yarnClient,
+				boolean sharedYarnClient) {
+			super(flinkConfiguration, yarnConfiguration, configurationDirectory, yarnClient, sharedYarnClient);
 		}
 
 		@Override