You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by az...@apache.org on 2020/04/15 16:53:04 UTC
[flink] 07/12: [FLINK-16745][yarn] Start Yarn JM with FLIP-116 JVM
memory args
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 42e913adc772980de0e0d0c26e72bf90a83dcb5a
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Wed Apr 8 18:06:58 2020 +0300
[FLINK-16745][yarn] Start Yarn JM with FLIP-116 JVM memory args
---
.../flink/yarn/YARNFileReplicationITCase.java | 5 +++-
.../flink/yarn/YARNHighAvailabilityITCase.java | 4 ++-
.../java/org/apache/flink/yarn/YARNITCase.java | 5 +++-
.../apache/flink/yarn/YarnConfigurationITCase.java | 2 +-
.../apache/flink/yarn/YarnClusterDescriptor.java | 10 ++++---
.../apache/flink/yarn/cli/FlinkYarnSessionCli.java | 2 +-
.../apache/flink/yarn/FlinkYarnSessionCliTest.java | 33 ++++++++++----------
.../flink/yarn/YarnClusterDescriptorTest.java | 35 ++++++++++++----------
8 files changed, 55 insertions(+), 41 deletions(-)
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNFileReplicationITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNFileReplicationITCase.java
index 055380b..5cb9e5f 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNFileReplicationITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNFileReplicationITCase.java
@@ -22,6 +22,7 @@ import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.AkkaOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -84,8 +85,9 @@ public class YARNFileReplicationITCase extends YarnTestBase {
yarnClusterDescriptor.addShipFiles(Arrays.asList(flinkLibFolder.listFiles()));
yarnClusterDescriptor.addShipFiles(Arrays.asList(flinkShadedHadoopDir.listFiles()));
+ final int masterMemory = yarnClusterDescriptor.getFlinkConfiguration().get(JobManagerOptions.TOTAL_PROCESS_MEMORY).getMebiBytes();
final ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
- .setMasterMemoryMB(768)
+ .setMasterMemoryMB(masterMemory)
.setTaskManagerMemoryMB(1024)
.setSlotsPerTaskManager(1)
.createClusterSpecification();
@@ -130,6 +132,7 @@ public class YARNFileReplicationITCase extends YarnTestBase {
private Configuration getDefaultConfiguration() {
final Configuration configuration = new Configuration();
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1g"));
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1g"));
configuration.setString(AkkaOptions.ASK_TIMEOUT, "30 s");
configuration.setString(CLASSPATH_INCLUDE_USER_JAR, YarnConfigOptions.UserJarInclusion.DISABLED.toString());
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 6420078..087d9a9 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
@@ -26,6 +26,7 @@ import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.configuration.RestartStrategyOptions;
@@ -273,6 +274,7 @@ public class YARNHighAvailabilityITCase extends YarnTestBase {
@Nonnull
private YarnClusterDescriptor setupYarnClusterDescriptor() {
final Configuration flinkConfiguration = new Configuration();
+ flinkConfiguration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1g"));
flinkConfiguration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1g"));
flinkConfiguration.setString(YarnConfigOptions.APPLICATION_ATTEMPTS, "10");
flinkConfiguration.setString(HighAvailabilityOptions.HA_MODE, "zookeeper");
@@ -290,7 +292,7 @@ public class YARNHighAvailabilityITCase extends YarnTestBase {
}
private RestClusterClient<ApplicationId> deploySessionCluster(YarnClusterDescriptor yarnClusterDescriptor) throws ClusterDeploymentException {
- final int masterMemory = 256;
+ final int masterMemory = yarnClusterDescriptor.getFlinkConfiguration().get(JobManagerOptions.TOTAL_PROCESS_MEMORY).getMebiBytes();
final int taskManagerMemory = 1024;
final ClusterClient<ApplicationId> yarnClusterClient = yarnClusterDescriptor
.deploySessionCluster(new ClusterSpecification.ClusterSpecificationBuilder()
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java
index bb15e28..da51e3c 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java
@@ -22,6 +22,7 @@ import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.AkkaOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -94,8 +95,9 @@ public class YARNITCase extends YarnTestBase {
yarnClusterDescriptor.addShipFiles(Arrays.asList(flinkLibFolder.listFiles()));
yarnClusterDescriptor.addShipFiles(Arrays.asList(flinkShadedHadoopDir.listFiles()));
+ final int masterMemory = yarnClusterDescriptor.getFlinkConfiguration().get(JobManagerOptions.TOTAL_PROCESS_MEMORY).getMebiBytes();
final ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
- .setMasterMemoryMB(768)
+ .setMasterMemoryMB(masterMemory)
.setTaskManagerMemoryMB(1024)
.setSlotsPerTaskManager(1)
.createClusterSpecification();
@@ -138,6 +140,7 @@ public class YARNITCase extends YarnTestBase {
private Configuration createDefaultConfiguration(YarnConfigOptions.UserJarInclusion userJarInclusion) {
Configuration configuration = new Configuration();
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1g"));
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1g"));
configuration.setString(AkkaOptions.ASK_TIMEOUT, "30 s");
configuration.setString(CLASSPATH_INCLUDE_USER_JAR, userJarInclusion.toString());
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
index d44d612..4d2fff2 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
@@ -90,7 +90,7 @@ public class YarnConfigurationITCase extends YarnTestBase {
configuration.set(TaskManagerOptions.NUM_TASK_SLOTS, slotsPerTaskManager);
final TaskExecutorProcessSpec tmResourceSpec = TaskExecutorProcessUtils.processSpecFromConfig(configuration);
- final int masterMemory = 64;
+ final int masterMemory = 768;
final int taskManagerMemory = tmResourceSpec.getTotalProcessMemorySize().getMebiBytes();
final YarnConfiguration yarnConfiguration = getYarnConfiguration();
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
index cde732b..0d94178 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
@@ -45,6 +45,8 @@ import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.runtime.util.config.memory.ProcessMemorySpec;
+import org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.ShutdownHookUtil;
@@ -556,6 +558,7 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
}
return new ClusterSpecification.ClusterSpecificationBuilder()
+ .setMasterProcessSpec(clusterSpecification.getMasterProcessSpec())
.setMasterMemoryMB(jobManagerMemoryMb)
.setTaskManagerMemoryMB(taskManagerMemoryMb)
.setSlotsPerTaskManager(clusterSpecification.getSlotsPerTaskManager())
@@ -932,7 +935,7 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
hasLogback,
hasLog4j,
hasKrb5,
- clusterSpecification.getMasterMemoryMB());
+ clusterSpecification.getMasterProcessSpec());
// setup security tokens
if (UserGroupInformation.isSecurityEnabled()) {
@@ -1570,7 +1573,7 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
boolean hasLogback,
boolean hasLog4j,
boolean hasKrb5,
- int jobManagerMemoryMb) {
+ ProcessMemorySpec processSpec) {
// ------------------ Prepare Application Master Container ------------------------------
// respect custom JVM options in the YAML file
@@ -1590,8 +1593,7 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
final Map<String, String> startCommandValues = new HashMap<>();
startCommandValues.put("java", "$JAVA_HOME/bin/java");
- int heapSize = BootstrapTools.calculateHeapSize(jobManagerMemoryMb, flinkConfiguration);
- String jvmHeapMem = String.format("-Xms%sm -Xmx%sm", heapSize, heapSize);
+ String jvmHeapMem = ProcessMemoryUtils.generateJvmParametersStr(processSpec);
startCommandValues.put("jvmmem", jvmHeapMem);
startCommandValues.put("jvmopts", javaOpts);
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 352ca6e..4ca49c6 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
@@ -355,7 +355,7 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine {
if (!MemorySize.MemoryUnit.hasUnit(jmMemoryVal)) {
jmMemoryVal += "m";
}
- effectiveConfiguration.set(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY, MemorySize.parse(jmMemoryVal));
+ effectiveConfiguration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse(jmMemoryVal));
}
if (commandLine.hasOption(tmMemory.getOpt())) {
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 bf097fb..6657190 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
@@ -111,7 +111,7 @@ public class FlinkYarnSessionCliTest extends TestLogger {
String[] params =
new String[] {"-ys", "3"};
- FlinkYarnSessionCli yarnCLI = createFlinkYarnSessionCliWithTmTotalMemory(2048);
+ FlinkYarnSessionCli yarnCLI = createFlinkYarnSessionCliWithJmAndTmTotalMemory(2048);
final CommandLine commandLine = yarnCLI.parseCommandLineOptions(params, true);
@@ -314,7 +314,7 @@ public class FlinkYarnSessionCliTest extends TestLogger {
final int taskManagerMemory = 7331;
final int slotsPerTaskManager = 30;
- configuration.set(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY, MemorySize.ofMebiBytes(jobManagerMemory));
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(jobManagerMemory));
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(taskManagerMemory));
configuration.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, slotsPerTaskManager);
@@ -340,7 +340,7 @@ public class FlinkYarnSessionCliTest extends TestLogger {
public void testConfigurationClusterSpecification() throws Exception {
final Configuration configuration = new Configuration();
final int jobManagerMemory = 1337;
- configuration.set(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY, MemorySize.ofMebiBytes(jobManagerMemory));
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(jobManagerMemory));
final int taskManagerMemory = 7331;
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(taskManagerMemory));
final int slotsPerTaskManager = 42;
@@ -361,10 +361,10 @@ public class FlinkYarnSessionCliTest extends TestLogger {
}
/**
- * Tests the specifying heap memory without unit for job manager and task manager.
+ * Tests the specifying total process memory without unit for job manager and task manager.
*/
@Test
- public void testHeapMemoryPropertyWithoutUnit() throws Exception {
+ public void testMemoryPropertyWithoutUnit() throws Exception {
final String[] args = new String[] { "-yjm", "1024", "-ytm", "2048" };
final FlinkYarnSessionCli flinkYarnSessionCli = createFlinkYarnSessionCli();
@@ -379,10 +379,10 @@ public class FlinkYarnSessionCliTest extends TestLogger {
}
/**
- * Tests the specifying heap memory with unit (MB) for job manager and task manager.
+ * Tests the specifying total process memory with unit (MB) for job manager and task manager.
*/
@Test
- public void testHeapMemoryPropertyWithUnitMB() throws Exception {
+ public void testMemoryPropertyWithUnitMB() throws Exception {
final String[] args = new String[] { "-yjm", "1024m", "-ytm", "2048m" };
final FlinkYarnSessionCli flinkYarnSessionCli = createFlinkYarnSessionCli();
final CommandLine commandLine = flinkYarnSessionCli.parseCommandLineOptions(args, false);
@@ -396,10 +396,10 @@ public class FlinkYarnSessionCliTest extends TestLogger {
}
/**
- * Tests the specifying heap memory with arbitrary unit for job manager and task manager.
+ * Tests the specifying total process memory with arbitrary unit for job manager and task manager.
*/
@Test
- public void testHeapMemoryPropertyWithArbitraryUnit() throws Exception {
+ public void testMemoryPropertyWithArbitraryUnit() throws Exception {
final String[] args = new String[] { "-yjm", "1g", "-ytm", "2g" };
final FlinkYarnSessionCli flinkYarnSessionCli = createFlinkYarnSessionCli();
final CommandLine commandLine = flinkYarnSessionCli.parseCommandLineOptions(args, false);
@@ -434,12 +434,12 @@ public class FlinkYarnSessionCliTest extends TestLogger {
}
/**
- * Tests the specifying job manager heap memory with config default value for job manager and task manager.
+ * Tests the specifying job manager total process memory with config default value for job manager and task manager.
*/
@Test
- public void testJobManagerHeapMemoryPropertyWithConfigDefaultValue() throws Exception {
- int tmMemory = 2048;
- final FlinkYarnSessionCli flinkYarnSessionCli = createFlinkYarnSessionCliWithTmTotalMemory(tmMemory);
+ public void testJobManagerMemoryPropertyWithConfigDefaultValue() throws Exception {
+ int procMemory = 2048;
+ final FlinkYarnSessionCli flinkYarnSessionCli = createFlinkYarnSessionCliWithJmAndTmTotalMemory(procMemory);
final CommandLine commandLine = flinkYarnSessionCli.parseCommandLineOptions(new String[0], false);
@@ -447,8 +447,8 @@ public class FlinkYarnSessionCliTest extends TestLogger {
final ClusterClientFactory<ApplicationId> clientFactory = getClusterClientFactory(executorConfig);
final ClusterSpecification clusterSpecification = clientFactory.getClusterSpecification(executorConfig);
- assertThat(clusterSpecification.getMasterMemoryMB(), is(1024));
- assertThat(clusterSpecification.getTaskManagerMemoryMB(), is(tmMemory));
+ assertThat(clusterSpecification.getMasterMemoryMB(), is(procMemory));
+ assertThat(clusterSpecification.getTaskManagerMemoryMB(), is(procMemory));
}
@Test
@@ -490,8 +490,9 @@ public class FlinkYarnSessionCliTest extends TestLogger {
return createFlinkYarnSessionCli(new Configuration());
}
- private FlinkYarnSessionCli createFlinkYarnSessionCliWithTmTotalMemory(int totalMemomory) throws FlinkException {
+ private FlinkYarnSessionCli createFlinkYarnSessionCliWithJmAndTmTotalMemory(int totalMemomory) throws FlinkException {
Configuration configuration = new Configuration();
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(totalMemomory));
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(totalMemomory));
return createFlinkYarnSessionCli(configuration);
}
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 7a855e9..09d2133 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
@@ -26,6 +26,8 @@ import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.runtime.jobmanager.JobManagerProcessSpec;
+import org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils;
import org.apache.flink.util.TestLogger;
import org.apache.flink.yarn.cli.FlinkYarnSessionCli;
import org.apache.flink.yarn.configuration.YarnConfigOptions;
@@ -55,6 +57,7 @@ import java.util.Map;
import java.util.Set;
import static junit.framework.TestCase.assertTrue;
+import static org.apache.flink.runtime.jobmanager.JobManagerProcessUtils.createDefaultJobManagerProcessSpec;
import static org.apache.flink.yarn.configuration.YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR;
import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.assertEquals;
@@ -156,8 +159,9 @@ public class YarnClusterDescriptorTest extends TestLogger {
Configuration cfg = new Configuration();
YarnClusterDescriptor clusterDescriptor = createYarnClusterDescriptor(cfg);
+ final JobManagerProcessSpec jobManagerProcessSpec = createDefaultJobManagerProcessSpec(1024);
final String java = "$JAVA_HOME/bin/java";
- final String jvmmem = "-Xms424m -Xmx424m";
+ final String jvmmem = ProcessMemoryUtils.generateJvmParametersStr(jobManagerProcessSpec);
final String jvmOpts = "-Djvm"; // if set
final String jmJvmOpts = "-DjmJvm"; // if set
final String krb5 = "-Djava.security.krb5.conf=krb5.conf";
@@ -173,7 +177,6 @@ public class YarnClusterDescriptorTest extends TestLogger {
final String redirects =
"1> " + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/jobmanager.out " +
"2> " + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/jobmanager.err";
- final int jobManagerMemory = 1024;
try {
// no logging, with/out krb5
@@ -188,7 +191,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
false,
false,
false,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
assertEquals(
@@ -202,7 +205,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
false,
false,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
// logback only, with/out krb5
@@ -217,7 +220,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
false,
false,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
assertEquals(
@@ -231,7 +234,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
false,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
// log4j, with/out krb5
@@ -246,7 +249,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
false,
true,
false,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
assertEquals(
@@ -260,7 +263,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
false,
true,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
// logback + log4j, with/out krb5
@@ -275,7 +278,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
false,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
assertEquals(
@@ -289,7 +292,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
// logback + log4j, with/out krb5, different JVM opts
@@ -307,7 +310,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
false,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
assertEquals(
@@ -321,7 +324,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
// logback + log4j, with/out krb5, different JVM opts
@@ -338,7 +341,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
false,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
assertEquals(
@@ -352,7 +355,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
// now try some configurations with different yarn.container-start-command-template
@@ -370,7 +373,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
cfg.setString(ConfigConstants.YARN_CONTAINER_START_COMMAND_TEMPLATE,
@@ -388,7 +391,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
} finally {
clusterDescriptor.close();