TestMRJobs.java
/**
* 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.hadoop.mapreduce.v2;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.StringReader;
import java.net.URI;
import java.security.PrivilegedExceptionAction;
import java.util.Arrays;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.Map;
import java.util.jar.JarOutputStream;
import java.util.zip.ZipEntry;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.FailingMapper;
import org.apache.hadoop.RandomTextWriterJob;
import org.apache.hadoop.RandomTextWriterJob.RandomInputFormat;
import org.apache.hadoop.fs.viewfs.ConfigUtil;
import org.apache.hadoop.mapreduce.SleepJob;
import org.apache.hadoop.mapreduce.SleepJob.SleepMapper;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TaskLog;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobCounter;
import org.apache.hadoop.mapreduce.JobPriority;
import org.apache.hadoop.mapreduce.JobStatus;
import org.apache.hadoop.mapreduce.MRConfig;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.TaskCompletionEvent;
import org.apache.hadoop.mapreduce.TaskID;
import org.apache.hadoop.mapreduce.TaskReport;
import org.apache.hadoop.mapreduce.TaskType;
import org.apache.hadoop.mapreduce.TypeConverter;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.v2.api.records.JobId;
import org.apache.hadoop.mapreduce.v2.app.AppContext;
import org.apache.hadoop.mapreduce.v2.app.MRAppMaster;
import org.apache.hadoop.mapreduce.v2.app.speculate.DefaultSpeculator;
import org.apache.hadoop.mapreduce.v2.app.speculate.Speculator;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.security.token.TokenIdentifier;
import org.apache.hadoop.util.ApplicationClassLoader;
import org.apache.hadoop.util.ClassUtil;
import org.apache.hadoop.util.JarFinder;
import org.apache.hadoop.util.Shell;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.WorkflowPriorityMappingsManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.WorkflowPriorityMappingsManager.WorkflowPriorityMapping;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
import org.apache.log4j.Level;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.assertj.core.api.Assertions.assertThat;
public class TestMRJobs {
private static final Logger LOG = LoggerFactory.getLogger(TestMRJobs.class);
private static final EnumSet<RMAppState> TERMINAL_RM_APP_STATES =
EnumSet.of(RMAppState.FINISHED, RMAppState.FAILED, RMAppState.KILLED);
private static final int NUM_NODE_MGRS = 3;
private static final String TEST_IO_SORT_MB = "11";
private static final int DEFAULT_REDUCES = 2;
protected int numSleepReducers = DEFAULT_REDUCES;
protected static MiniMRYarnCluster mrCluster;
protected static MiniDFSCluster dfsCluster;
private static Configuration conf = new Configuration();
private static FileSystem localFs;
private static FileSystem remoteFs;
static {
try {
localFs = FileSystem.getLocal(conf);
} catch (IOException io) {
throw new RuntimeException("problem getting local fs", io);
}
}
private static Path TEST_ROOT_DIR = localFs.makeQualified(
new Path("target", TestMRJobs.class.getName() + "-tmpDir"));
static Path APP_JAR = new Path(TEST_ROOT_DIR, "MRAppJar.jar");
private static final String OUTPUT_ROOT_DIR = "/tmp/" +
TestMRJobs.class.getSimpleName();
private static final Path TEST_RESOURCES_DIR = new Path(TEST_ROOT_DIR,
"localizedResources");
@BeforeClass
public static void setup() throws IOException {
try {
dfsCluster = new MiniDFSCluster.Builder(conf).numDataNodes(2)
.format(true).racks(null).build();
remoteFs = dfsCluster.getFileSystem();
} catch (IOException io) {
throw new RuntimeException("problem starting mini dfs cluster", io);
}
if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
LOG.info("MRAppJar " + MiniMRYarnCluster.APPJAR
+ " not found. Not running test.");
return;
}
if (mrCluster == null) {
mrCluster = new MiniMRYarnCluster(TestMRJobs.class.getName(),
NUM_NODE_MGRS);
Configuration conf = new Configuration();
conf.set("fs.defaultFS", remoteFs.getUri().toString()); // use HDFS
conf.set(MRJobConfig.MR_AM_STAGING_DIR, "/apps_staging_dir");
conf.setInt(YarnConfiguration.MAX_CLUSTER_LEVEL_APPLICATION_PRIORITY, 10);
mrCluster.init(conf);
mrCluster.start();
}
// Copy MRAppJar and make it private. TODO: FIXME. This is a hack to
// workaround the absent public discache.
localFs.copyFromLocalFile(new Path(MiniMRYarnCluster.APPJAR), APP_JAR);
localFs.setPermission(APP_JAR, new FsPermission("700"));
}
@AfterClass
public static void tearDown() throws IOException {
if (mrCluster != null) {
mrCluster.stop();
mrCluster = null;
}
if (dfsCluster != null) {
dfsCluster.shutdown();
dfsCluster = null;
}
if (localFs.exists(TEST_RESOURCES_DIR)) {
// clean up resource directory
localFs.delete(TEST_RESOURCES_DIR, true);
}
}
@After
public void resetInit() {
numSleepReducers = DEFAULT_REDUCES;
}
private static void setupJobResourceDirs() throws IOException {
if (localFs.exists(TEST_RESOURCES_DIR)) {
// clean up directory
localFs.delete(TEST_RESOURCES_DIR, true);
}
localFs.mkdirs(TEST_RESOURCES_DIR);
FSDataOutputStream outF1 = null;
try {
// 10KB file
outF1 = localFs.create(new Path(TEST_RESOURCES_DIR, "file1.txt"));
outF1.write(new byte[10 * 1024]);
} finally {
if (outF1 != null) {
outF1.close();
}
}
localFs.createNewFile(new Path(TEST_RESOURCES_DIR, "file2.txt"));
Path subDir = new Path(TEST_RESOURCES_DIR, "subDir");
localFs.mkdirs(subDir);
FSDataOutputStream outF3 = null;
try {
// 1MB (plus 10 Bytes) file
outF3 = localFs.create(new Path(subDir, "file3.txt"));
outF3.write(new byte[(1 * 1024 * 1024) + 10]);
} finally {
if (outF3 != null) {
outF3.close();
}
}
localFs.createNewFile(new Path(subDir, "file4.txt"));
}
@Test (timeout = 300000)
public void testSleepJob() throws Exception {
testSleepJobInternal(false);
}
@Test (timeout = 300000)
public void testSleepJobWithRemoteJar() throws Exception {
testSleepJobInternal(true);
}
@Test(timeout = 300000)
public void testSleepJobWithLocalResourceUnderLimit() throws Exception {
Configuration sleepConf = new Configuration(mrCluster.getConfig());
// set limits to well above what is expected
sleepConf.setInt(MRJobConfig.MAX_RESOURCES, 6);
sleepConf.setLong(MRJobConfig.MAX_RESOURCES_MB, 6);
setupJobResourceDirs();
sleepConf.set("tmpfiles", TEST_RESOURCES_DIR.toString());
testSleepJobInternal(sleepConf, false, true, null);
}
@Test(timeout = 300000)
public void testSleepJobWithLocalResourceSizeOverLimit() throws Exception {
Configuration sleepConf = new Configuration(mrCluster.getConfig());
// set limits to well below what is expected
sleepConf.setLong(MRJobConfig.MAX_RESOURCES_MB, 1);
setupJobResourceDirs();
sleepConf.set("tmpfiles", TEST_RESOURCES_DIR.toString());
testSleepJobInternal(sleepConf, false, false,
ResourceViolation.TOTAL_RESOURCE_SIZE);
}
@Test(timeout = 300000)
public void testSleepJobWithLocalResourceNumberOverLimit() throws Exception {
Configuration sleepConf = new Configuration(mrCluster.getConfig());
// set limits to well below what is expected
sleepConf.setInt(MRJobConfig.MAX_RESOURCES, 1);
setupJobResourceDirs();
sleepConf.set("tmpfiles", TEST_RESOURCES_DIR.toString());
testSleepJobInternal(sleepConf, false, false,
ResourceViolation.NUMBER_OF_RESOURCES);
}
@Test(timeout = 300000)
public void testSleepJobWithLocalResourceCheckAndRemoteJar()
throws Exception {
Configuration sleepConf = new Configuration(mrCluster.getConfig());
// set limits to well above what is expected
sleepConf.setInt(MRJobConfig.MAX_RESOURCES, 6);
sleepConf.setLong(MRJobConfig.MAX_RESOURCES_MB, 6);
setupJobResourceDirs();
sleepConf.set("tmpfiles", TEST_RESOURCES_DIR.toString());
testSleepJobInternal(sleepConf, true, true, null);
}
@Test(timeout = 300000)
public void testSleepJobWithLocalIndividualResourceOverLimit()
throws Exception {
Configuration sleepConf = new Configuration(mrCluster.getConfig());
// set limits to well below what is expected
sleepConf.setInt(MRJobConfig.MAX_SINGLE_RESOURCE_MB, 1);
setupJobResourceDirs();
sleepConf.set("tmpfiles", TEST_RESOURCES_DIR.toString());
testSleepJobInternal(sleepConf, false, false,
ResourceViolation.SINGLE_RESOURCE_SIZE);
}
@Test(timeout = 300000)
public void testSleepJobWithLocalIndividualResourceUnderLimit()
throws Exception {
Configuration sleepConf = new Configuration(mrCluster.getConfig());
// set limits to well below what is expected
sleepConf.setInt(MRJobConfig.MAX_SINGLE_RESOURCE_MB, 2);
setupJobResourceDirs();
sleepConf.set("tmpfiles", TEST_RESOURCES_DIR.toString());
testSleepJobInternal(sleepConf, false, true, null);
}
private void testSleepJobInternal(boolean useRemoteJar) throws Exception {
testSleepJobInternal(new Configuration(mrCluster.getConfig()),
useRemoteJar, true, null);
}
private enum ResourceViolation {
NUMBER_OF_RESOURCES, TOTAL_RESOURCE_SIZE, SINGLE_RESOURCE_SIZE;
}
private void testSleepJobInternal(Configuration sleepConf,
boolean useRemoteJar, boolean jobSubmissionShouldSucceed,
ResourceViolation violation) throws Exception {
LOG.info("\n\n\nStarting testSleepJob: useRemoteJar=" + useRemoteJar);
if (!jobSubmissionShouldSucceed && violation == null) {
Assert.fail("Test is misconfigured. jobSubmissionShouldSucceed is set"
+ " to false and a ResourceViolation is not specified.");
}
if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
LOG.info("MRAppJar " + MiniMRYarnCluster.APPJAR
+ " not found. Not running test.");
return;
}
// set master address to local to test that local mode applied iff framework == local
sleepConf.set(MRConfig.MASTER_ADDRESS, "local");
SleepJob sleepJob = new SleepJob();
sleepJob.setConf(sleepConf);
// job with 3 maps (10s) and numReduces reduces (5s), 1 "record" each:
Job job = sleepJob.createJob(3, numSleepReducers, 10000, 1, 5000, 1);
job.addFileToClassPath(APP_JAR); // The AppMaster jar itself.
if (useRemoteJar) {
final Path localJar = new Path(
ClassUtil.findContainingJar(SleepJob.class));
ConfigUtil.addLink(job.getConfiguration(), "/jobjars",
localFs.makeQualified(localJar.getParent()).toUri());
job.setJar("viewfs:///jobjars/" + localJar.getName());
} else {
job.setJarByClass(SleepJob.class);
}
job.setMaxMapAttempts(1); // speed up failures
try {
job.submit();
Assert.assertTrue("JobSubmission succeeded when it should have failed.",
jobSubmissionShouldSucceed);
} catch (IOException e) {
if (jobSubmissionShouldSucceed) {
Assert
.fail("Job submission failed when it should have succeeded: " + e);
}
switch (violation) {
case NUMBER_OF_RESOURCES:
if (!e.getMessage().contains(
"This job has exceeded the maximum number of"
+ " submitted resources")) {
Assert.fail("Test failed unexpectedly: " + e);
}
break;
case TOTAL_RESOURCE_SIZE:
if (!e.getMessage().contains(
"This job has exceeded the maximum size of submitted resources")) {
Assert.fail("Test failed unexpectedly: " + e);
}
break;
case SINGLE_RESOURCE_SIZE:
if (!e.getMessage().contains(
"This job has exceeded the maximum size of a single submitted")) {
Assert.fail("Test failed unexpectedly: " + e);
}
break;
default:
Assert.fail("Test failed unexpectedly: " + e);
break;
}
// we are done with the test (job submission failed)
return;
}
String trackingUrl = job.getTrackingURL();
String jobId = job.getJobID().toString();
boolean succeeded = job.waitForCompletion(true);
Assert.assertTrue(succeeded);
Assert.assertEquals(JobStatus.State.SUCCEEDED, job.getJobState());
Assert.assertTrue("Tracking URL was " + trackingUrl +
" but didn't Match Job ID " + jobId ,
trackingUrl.endsWith(jobId.substring(jobId.lastIndexOf("_")) + "/"));
verifySleepJobCounters(job);
verifyTaskProgress(job);
// TODO later: add explicit "isUber()" checks of some sort (extend
// JobStatus?)--compare against MRJobConfig.JOB_UBERTASK_ENABLE value
}
@Test(timeout = 3000000)
public void testJobWithChangePriority() throws Exception {
Configuration sleepConf = new Configuration(mrCluster.getConfig());
// Assumption can be removed when FS priority support is implemented
Assume.assumeFalse(sleepConf.get(YarnConfiguration.RM_SCHEDULER)
.equals(FairScheduler.class.getCanonicalName()));
if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
LOG.info("MRAppJar " + MiniMRYarnCluster.APPJAR
+ " not found. Not running test.");
return;
}
// set master address to local to test that local mode applied if framework
// equals local
sleepConf.set(MRConfig.MASTER_ADDRESS, "local");
sleepConf.setInt(MRJobConfig.MR_AM_TO_RM_HEARTBEAT_INTERVAL_MS, 5);
SleepJob sleepJob = new SleepJob();
sleepJob.setConf(sleepConf);
Job job = sleepJob.createJob(1, 1, 1000, 20, 50, 1);
job.addFileToClassPath(APP_JAR); // The AppMaster jar itself.
job.setJarByClass(SleepJob.class);
job.setMaxMapAttempts(1); // speed up failures
job.submit();
// Set the priority to HIGH
job.setPriority(JobPriority.HIGH);
waitForPriorityToUpdate(job, JobPriority.HIGH);
// Verify the priority from job itself
assertThat(job.getPriority()).isEqualTo(JobPriority.HIGH);
// Change priority to NORMAL (3) with new api
job.setPriorityAsInteger(3); // Verify the priority from job itself
waitForPriorityToUpdate(job, JobPriority.NORMAL);
assertThat(job.getPriority()).isEqualTo(JobPriority.NORMAL);
// Change priority to a high integer value with new api
job.setPriorityAsInteger(89); // Verify the priority from job itself
waitForPriorityToUpdate(job, JobPriority.UNDEFINED_PRIORITY);
assertThat(job.getPriority()).isEqualTo(JobPriority.UNDEFINED_PRIORITY);
boolean succeeded = job.waitForCompletion(true);
Assert.assertTrue(succeeded);
Assert.assertEquals(JobStatus.State.SUCCEEDED, job.getJobState());
}
@Test(timeout = 300000)
public void testJobWithWorkflowPriority() throws Exception {
Configuration sleepConf = new Configuration(mrCluster.getConfig());
if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
LOG.info("MRAppJar " + MiniMRYarnCluster.APPJAR
+ " not found. Not running test.");
return;
}
CapacityScheduler scheduler = (CapacityScheduler) mrCluster
.getResourceManager().getResourceScheduler();
CapacitySchedulerConfiguration csConf = scheduler.getConfiguration();
csConf.set(CapacitySchedulerConfiguration.WORKFLOW_PRIORITY_MAPPINGS,
WorkflowPriorityMappingsManager.getWorkflowPriorityMappingStr(
Arrays.asList(new WorkflowPriorityMapping(
"wf1", "root.default", Priority.newInstance(1)))));
csConf.setBoolean(CapacitySchedulerConfiguration.
ENABLE_WORKFLOW_PRIORITY_MAPPINGS_OVERRIDE, true);
scheduler.reinitialize(csConf, scheduler.getRMContext());
// set master address to local to test that local mode applied if framework
// equals local
sleepConf.set(MRConfig.MASTER_ADDRESS, "local");
sleepConf
.setInt("yarn.app.mapreduce.am.scheduler.heartbeat.interval-ms", 5);
sleepConf.set(MRJobConfig.JOB_TAGS,
YarnConfiguration.DEFAULT_YARN_WORKFLOW_ID_TAG_PREFIX + "wf1");
SleepJob sleepJob = new SleepJob();
sleepJob.setConf(sleepConf);
Job job = sleepJob.createJob(1, 1, 1000, 20, 50, 1);
job.addFileToClassPath(APP_JAR); // The AppMaster jar itself.
job.setJarByClass(SleepJob.class);
job.setMaxMapAttempts(1); // speed up failures
// VERY_HIGH priority should get overwritten by workflow priority mapping
job.setPriority(JobPriority.VERY_HIGH);
job.submit();
waitForPriorityToUpdate(job, JobPriority.VERY_LOW);
// Verify the priority from job itself
Assert.assertEquals(JobPriority.VERY_LOW, job.getPriority());
boolean succeeded = job.waitForCompletion(true);
Assert.assertTrue(succeeded);
Assert.assertEquals(JobStatus.State.SUCCEEDED, job.getJobState());
}
private void waitForPriorityToUpdate(Job job, JobPriority expectedStatus)
throws IOException, InterruptedException {
// Max wait time to get the priority update can be kept as 20sec (100 *
// 100ms)
int waitCnt = 200;
while (waitCnt-- > 0) {
if (job.getPriority().equals(expectedStatus)) {
// Stop waiting as priority is updated.
break;
} else {
Thread.sleep(100);
}
}
}
@Test(timeout = 300000)
public void testJobClassloader() throws IOException, InterruptedException,
ClassNotFoundException {
testJobClassloader(false);
}
@Test(timeout = 300000)
public void testJobClassloaderWithCustomClasses() throws IOException,
InterruptedException, ClassNotFoundException {
testJobClassloader(true);
}
private void testJobClassloader(boolean useCustomClasses) throws IOException,
InterruptedException, ClassNotFoundException {
LOG.info("\n\n\nStarting testJobClassloader()"
+ " useCustomClasses=" + useCustomClasses);
if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
LOG.info("MRAppJar " + MiniMRYarnCluster.APPJAR
+ " not found. Not running test.");
return;
}
final Configuration sleepConf = new Configuration(mrCluster.getConfig());
// set master address to local to test that local mode applied iff framework == local
sleepConf.set(MRConfig.MASTER_ADDRESS, "local");
sleepConf.setBoolean(MRJobConfig.MAPREDUCE_JOB_CLASSLOADER, true);
if (useCustomClasses) {
// to test AM loading user classes such as output format class, we want
// to blacklist them from the system classes (they need to be prepended
// as the first match wins)
String systemClasses = ApplicationClassLoader.SYSTEM_CLASSES_DEFAULT;
// exclude the custom classes from system classes
systemClasses = "-" + CustomOutputFormat.class.getName() + ",-" +
CustomSpeculator.class.getName() + "," +
systemClasses;
sleepConf.set(MRJobConfig.MAPREDUCE_JOB_CLASSLOADER_SYSTEM_CLASSES,
systemClasses);
}
sleepConf.set(MRJobConfig.IO_SORT_MB, TEST_IO_SORT_MB);
sleepConf.set(MRJobConfig.MR_AM_LOG_LEVEL, Level.ALL.toString());
sleepConf.set(MRJobConfig.MAP_LOG_LEVEL, Level.ALL.toString());
sleepConf.set(MRJobConfig.REDUCE_LOG_LEVEL, Level.ALL.toString());
sleepConf.set(MRJobConfig.MAP_JAVA_OPTS, "-verbose:class");
final SleepJob sleepJob = new SleepJob();
sleepJob.setConf(sleepConf);
final Job job = sleepJob.createJob(1, 1, 10, 1, 10, 1);
job.setMapperClass(ConfVerificationMapper.class);
job.addFileToClassPath(APP_JAR); // The AppMaster jar itself.
job.setJarByClass(SleepJob.class);
job.setMaxMapAttempts(1); // speed up failures
if (useCustomClasses) {
// set custom output format class and speculator class
job.setOutputFormatClass(CustomOutputFormat.class);
final Configuration jobConf = job.getConfiguration();
jobConf.setClass(MRJobConfig.MR_AM_JOB_SPECULATOR, CustomSpeculator.class,
Speculator.class);
// speculation needs to be enabled for the speculator to be loaded
jobConf.setBoolean(MRJobConfig.MAP_SPECULATIVE, true);
}
job.submit();
boolean succeeded = job.waitForCompletion(true);
Assert.assertTrue("Job status: " + job.getStatus().getFailureInfo(),
succeeded);
}
public static class CustomOutputFormat<K,V> extends NullOutputFormat<K,V> {
public CustomOutputFormat() {
verifyClassLoader(getClass());
}
/**
* Verifies that the class was loaded by the job classloader if it is in the
* context of the MRAppMaster, and if not throws an exception to fail the
* job.
*/
private void verifyClassLoader(Class<?> cls) {
// to detect that it is instantiated in the context of the MRAppMaster, we
// inspect the stack trace and determine a caller is MRAppMaster
for (StackTraceElement e: new Throwable().getStackTrace()) {
if (e.getClassName().equals(MRAppMaster.class.getName()) &&
!(cls.getClassLoader() instanceof ApplicationClassLoader)) {
throw new ExceptionInInitializerError("incorrect classloader used");
}
}
}
}
public static class CustomSpeculator extends DefaultSpeculator {
public CustomSpeculator(Configuration conf, AppContext context) {
super(conf, context);
verifyClassLoader(getClass());
}
/**
* Verifies that the class was loaded by the job classloader if it is in the
* context of the MRAppMaster, and if not throws an exception to fail the
* job.
*/
private void verifyClassLoader(Class<?> cls) {
// to detect that it is instantiated in the context of the MRAppMaster, we
// inspect the stack trace and determine a caller is MRAppMaster
for (StackTraceElement e: new Throwable().getStackTrace()) {
if (e.getClassName().equals(MRAppMaster.class.getName()) &&
!(cls.getClassLoader() instanceof ApplicationClassLoader)) {
throw new ExceptionInInitializerError("incorrect classloader used");
}
}
}
}
protected void verifySleepJobCounters(Job job) throws InterruptedException,
IOException {
Counters counters = job.getCounters();
Assert.assertEquals(3, counters.findCounter(JobCounter.OTHER_LOCAL_MAPS)
.getValue());
Assert.assertEquals(3, counters.findCounter(JobCounter.TOTAL_LAUNCHED_MAPS)
.getValue());
Assert.assertEquals(numSleepReducers,
counters.findCounter(JobCounter.TOTAL_LAUNCHED_REDUCES).getValue());
}
protected void verifyTaskProgress(Job job) throws InterruptedException,
IOException {
for (TaskReport taskReport : job.getTaskReports(TaskType.MAP)) {
Assert.assertTrue(0.9999f < taskReport.getProgress()
&& 1.0001f > taskReport.getProgress());
}
for (TaskReport taskReport : job.getTaskReports(TaskType.REDUCE)) {
Assert.assertTrue(0.9999f < taskReport.getProgress()
&& 1.0001f > taskReport.getProgress());
}
}
@Test (timeout = 60000)
public void testRandomWriter() throws IOException, InterruptedException,
ClassNotFoundException {
LOG.info("\n\n\nStarting testRandomWriter().");
if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
LOG.info("MRAppJar " + MiniMRYarnCluster.APPJAR
+ " not found. Not running test.");
return;
}
RandomTextWriterJob randomWriterJob = new RandomTextWriterJob();
mrCluster.getConfig().set(RandomTextWriterJob.TOTAL_BYTES, "3072");
mrCluster.getConfig().set(RandomTextWriterJob.BYTES_PER_MAP, "1024");
Job job = randomWriterJob.createJob(mrCluster.getConfig());
Path outputDir = new Path(OUTPUT_ROOT_DIR, "random-output");
FileOutputFormat.setOutputPath(job, outputDir);
job.setSpeculativeExecution(false);
job.addFileToClassPath(APP_JAR); // The AppMaster jar itself.
job.setJarByClass(RandomTextWriterJob.class);
job.setMaxMapAttempts(1); // speed up failures
job.submit();
String trackingUrl = job.getTrackingURL();
String jobId = job.getJobID().toString();
boolean succeeded = job.waitForCompletion(true);
Assert.assertTrue(succeeded);
Assert.assertEquals(JobStatus.State.SUCCEEDED, job.getJobState());
Assert.assertTrue("Tracking URL was " + trackingUrl +
" but didn't Match Job ID " + jobId ,
trackingUrl.endsWith(jobId.substring(jobId.lastIndexOf("_")) + "/"));
// Make sure there are three files in the output-dir
RemoteIterator<FileStatus> iterator =
FileContext.getFileContext(mrCluster.getConfig()).listStatus(
outputDir);
int count = 0;
while (iterator.hasNext()) {
FileStatus file = iterator.next();
if (!file.getPath().getName()
.equals(FileOutputCommitter.SUCCEEDED_FILE_NAME)) {
count++;
}
}
Assert.assertEquals("Number of part files is wrong!", 3, count);
verifyRandomWriterCounters(job);
// TODO later: add explicit "isUber()" checks of some sort
}
protected void verifyRandomWriterCounters(Job job)
throws InterruptedException, IOException {
Counters counters = job.getCounters();
Assert.assertEquals(3, counters.findCounter(JobCounter.OTHER_LOCAL_MAPS)
.getValue());
Assert.assertEquals(3, counters.findCounter(JobCounter.TOTAL_LAUNCHED_MAPS)
.getValue());
}
@Test (timeout = 60000)
public void testFailingMapper() throws IOException, InterruptedException,
ClassNotFoundException {
LOG.info("\n\n\nStarting testFailingMapper().");
if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
LOG.info("MRAppJar " + MiniMRYarnCluster.APPJAR
+ " not found. Not running test.");
return;
}
Job job = runFailingMapperJob();
TaskID taskID = new TaskID(job.getJobID(), TaskType.MAP, 0);
TaskAttemptID aId = new TaskAttemptID(taskID, 0);
System.out.println("Diagnostics for " + aId + " :");
for (String diag : job.getTaskDiagnostics(aId)) {
System.out.println(diag);
}
aId = new TaskAttemptID(taskID, 1);
System.out.println("Diagnostics for " + aId + " :");
for (String diag : job.getTaskDiagnostics(aId)) {
System.out.println(diag);
}
TaskCompletionEvent[] events = job.getTaskCompletionEvents(0, 2);
Assert.assertEquals(TaskCompletionEvent.Status.FAILED,
events[0].getStatus());
Assert.assertEquals(TaskCompletionEvent.Status.TIPFAILED,
events[1].getStatus());
Assert.assertEquals(JobStatus.State.FAILED, job.getJobState());
verifyFailingMapperCounters(job);
// TODO later: add explicit "isUber()" checks of some sort
}
protected void verifyFailingMapperCounters(Job job)
throws InterruptedException, IOException {
Counters counters = job.getCounters();
Assert.assertEquals(2, counters.findCounter(JobCounter.OTHER_LOCAL_MAPS)
.getValue());
Assert.assertEquals(2, counters.findCounter(JobCounter.TOTAL_LAUNCHED_MAPS)
.getValue());
Assert.assertEquals(2, counters.findCounter(JobCounter.NUM_FAILED_MAPS)
.getValue());
Assert
.assertTrue(counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS) != null
&& counters.findCounter(JobCounter.SLOTS_MILLIS_MAPS).getValue() != 0);
}
protected Job runFailingMapperJob()
throws IOException, InterruptedException, ClassNotFoundException {
Configuration myConf = new Configuration(mrCluster.getConfig());
myConf.setInt(MRJobConfig.NUM_MAPS, 1);
myConf.setInt(MRJobConfig.MAP_MAX_ATTEMPTS, 2); //reduce the number of attempts
Job job = Job.getInstance(myConf);
job.setJarByClass(FailingMapper.class);
job.setJobName("failmapper");
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.setInputFormatClass(RandomInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setMapperClass(FailingMapper.class);
job.setNumReduceTasks(0);
FileOutputFormat.setOutputPath(job, new Path(OUTPUT_ROOT_DIR,
"failmapper-output"));
job.addFileToClassPath(APP_JAR); // The AppMaster jar itself.
job.submit();
String trackingUrl = job.getTrackingURL();
String jobId = job.getJobID().toString();
boolean succeeded = job.waitForCompletion(true);
Assert.assertFalse(succeeded);
Assert.assertTrue("Tracking URL was " + trackingUrl +
" but didn't Match Job ID " + jobId ,
trackingUrl.endsWith(jobId.substring(jobId.lastIndexOf("_")) + "/"));
return job;
}
//@Test (timeout = 60000)
public void testSleepJobWithSecurityOn() throws IOException,
InterruptedException, ClassNotFoundException {
LOG.info("\n\n\nStarting testSleepJobWithSecurityOn().");
if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
return;
}
mrCluster.getConfig().set(
CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION,
"kerberos");
mrCluster.getConfig().set(YarnConfiguration.RM_KEYTAB, "/etc/krb5.keytab");
mrCluster.getConfig().set(YarnConfiguration.NM_KEYTAB, "/etc/krb5.keytab");
mrCluster.getConfig().set(YarnConfiguration.RM_PRINCIPAL,
"rm/sightbusy-lx@LOCALHOST");
mrCluster.getConfig().set(YarnConfiguration.NM_PRINCIPAL,
"nm/sightbusy-lx@LOCALHOST");
UserGroupInformation.setConfiguration(mrCluster.getConfig());
// Keep it in here instead of after RM/NM as multiple user logins happen in
// the same JVM.
UserGroupInformation user = UserGroupInformation.getCurrentUser();
LOG.info("User name is " + user.getUserName());
for (Token<? extends TokenIdentifier> str : user.getTokens()) {
LOG.info("Token is " + str.encodeToUrlString());
}
user.doAs(new PrivilegedExceptionAction<Void>() {
@Override
public Void run() throws Exception {
SleepJob sleepJob = new SleepJob();
sleepJob.setConf(mrCluster.getConfig());
Job job = sleepJob.createJob(3, 0, 10000, 1, 0, 0);
// //Job with reduces
// Job job = sleepJob.createJob(3, 2, 10000, 1, 10000, 1);
job.addFileToClassPath(APP_JAR); // The AppMaster jar itself.
job.submit();
String trackingUrl = job.getTrackingURL();
String jobId = job.getJobID().toString();
job.waitForCompletion(true);
Assert.assertEquals(JobStatus.State.SUCCEEDED, job.getJobState());
Assert.assertTrue("Tracking URL was " + trackingUrl +
" but didn't Match Job ID " + jobId ,
trackingUrl.endsWith(jobId.substring(jobId.lastIndexOf("_")) + "/"));
return null;
}
});
// TODO later: add explicit "isUber()" checks of some sort
}
@Test(timeout = 120000)
public void testContainerRollingLog() throws IOException,
InterruptedException, ClassNotFoundException {
if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
LOG.info("MRAppJar " + MiniMRYarnCluster.APPJAR
+ " not found. Not running test.");
return;
}
final SleepJob sleepJob = new SleepJob();
final JobConf sleepConf = new JobConf(mrCluster.getConfig());
sleepConf.set(MRJobConfig.MAP_LOG_LEVEL, Level.ALL.toString());
final long userLogKb = 4;
sleepConf.setLong(MRJobConfig.TASK_USERLOG_LIMIT, userLogKb);
sleepConf.setInt(MRJobConfig.TASK_LOG_BACKUPS, 3);
sleepConf.set(MRJobConfig.MR_AM_LOG_LEVEL, Level.ALL.toString());
final long amLogKb = 7;
sleepConf.setLong(MRJobConfig.MR_AM_LOG_KB, amLogKb);
sleepConf.setInt(MRJobConfig.MR_AM_LOG_BACKUPS, 7);
sleepJob.setConf(sleepConf);
final Job job = sleepJob.createJob(1, 0, 1L, 100, 0L, 0);
job.setJarByClass(SleepJob.class);
job.addFileToClassPath(APP_JAR); // The AppMaster jar itself.
job.waitForCompletion(true);
final JobId jobId = TypeConverter.toYarn(job.getJobID());
final ApplicationId appID = jobId.getAppId();
int pollElapsed = 0;
while (true) {
Thread.sleep(1000);
pollElapsed += 1000;
if (TERMINAL_RM_APP_STATES.contains(
mrCluster.getResourceManager().getRMContext().getRMApps().get(appID)
.getState())) {
break;
}
if (pollElapsed >= 60000) {
LOG.warn("application did not reach terminal state within 60 seconds");
break;
}
}
Assert.assertEquals(RMAppState.FINISHED, mrCluster.getResourceManager()
.getRMContext().getRMApps().get(appID).getState());
// Job finished, verify logs
//
final String appIdStr = appID.toString();
final String appIdSuffix = appIdStr.substring("application_".length(),
appIdStr.length());
final String containerGlob = "container_" + appIdSuffix + "_*_*";
final String syslogGlob = appIdStr
+ Path.SEPARATOR + containerGlob
+ Path.SEPARATOR + TaskLog.LogName.SYSLOG;
int numAppMasters = 0;
int numMapTasks = 0;
for (int i = 0; i < NUM_NODE_MGRS; i++) {
final Configuration nmConf = mrCluster.getNodeManager(i).getConfig();
for (String logDir :
nmConf.getTrimmedStrings(YarnConfiguration.NM_LOG_DIRS)) {
final Path absSyslogGlob =
new Path(logDir + Path.SEPARATOR + syslogGlob);
LOG.info("Checking for glob: " + absSyslogGlob);
final FileStatus[] syslogs = localFs.globStatus(absSyslogGlob);
for (FileStatus slog : syslogs) {
boolean foundAppMaster = job.isUber();
final Path containerPathComponent = slog.getPath().getParent();
if (!foundAppMaster) {
final ContainerId cid = ContainerId.fromString(
containerPathComponent.getName());
foundAppMaster =
((cid.getContainerId() & ContainerId.CONTAINER_ID_BITMASK)== 1);
}
final FileStatus[] sysSiblings = localFs.globStatus(new Path(
containerPathComponent, TaskLog.LogName.SYSLOG + "*"));
// sort to ensure for i > 0 sysSiblings[i] == "syslog.i"
Arrays.sort(sysSiblings);
if (foundAppMaster) {
numAppMasters++;
} else {
numMapTasks++;
}
if (foundAppMaster) {
Assert.assertSame("Unexpected number of AM sylog* files",
sleepConf.getInt(MRJobConfig.MR_AM_LOG_BACKUPS, 0) + 1,
sysSiblings.length);
Assert.assertTrue("AM syslog.1 length kb should be >= " + amLogKb,
sysSiblings[1].getLen() >= amLogKb * 1024);
} else {
Assert.assertSame("Unexpected number of MR task sylog* files",
sleepConf.getInt(MRJobConfig.TASK_LOG_BACKUPS, 0) + 1,
sysSiblings.length);
Assert.assertTrue("MR syslog.1 length kb should be >= " + userLogKb,
sysSiblings[1].getLen() >= userLogKb * 1024);
}
}
}
}
// Make sure we checked non-empty set
//
Assert.assertEquals("No AppMaster log found!", 1, numAppMasters);
if (sleepConf.getBoolean(MRJobConfig.JOB_UBERTASK_ENABLE, false)) {
Assert.assertEquals("MapTask log with uber found!", 0, numMapTasks);
} else {
Assert.assertEquals("No MapTask log found!", 1, numMapTasks);
}
}
public static class DistributedCacheChecker extends
Mapper<LongWritable, Text, NullWritable, NullWritable> {
@Override
public void setup(Context context) throws IOException {
Configuration conf = context.getConfiguration();
Path[] localFiles = context.getLocalCacheFiles();
URI[] files = context.getCacheFiles();
Path[] localArchives = context.getLocalCacheArchives();
URI[] archives = context.getCacheArchives();
// Check that 4 (2 + appjar + DistrubutedCacheChecker jar) files
// and 2 archives are present
Assert.assertEquals(4, localFiles.length);
Assert.assertEquals(4, files.length);
Assert.assertEquals(2, localArchives.length);
Assert.assertEquals(2, archives.length);
// Check lengths of the files
Map<String, Path> filesMap = pathsToMap(localFiles);
Assert.assertTrue(filesMap.containsKey("distributed.first.symlink"));
Assert.assertEquals(1, localFs.getFileStatus(
filesMap.get("distributed.first.symlink")).getLen());
Assert.assertTrue(filesMap.containsKey("distributed.second.jar"));
Assert.assertTrue(localFs.getFileStatus(
filesMap.get("distributed.second.jar")).getLen() > 1);
// Check extraction of the archive
Map<String, Path> archivesMap = pathsToMap(localArchives);
Assert.assertTrue(archivesMap.containsKey("distributed.third.jar"));
Assert.assertTrue(localFs.exists(new Path(
archivesMap.get("distributed.third.jar"), "distributed.jar.inside3")));
Assert.assertTrue(archivesMap.containsKey("distributed.fourth.jar"));
Assert.assertTrue(localFs.exists(new Path(
archivesMap.get("distributed.fourth.jar"), "distributed.jar.inside4")));
// Check the class loaders
LOG.info("Java Classpath: " + System.getProperty("java.class.path"));
ClassLoader cl = Thread.currentThread().getContextClassLoader();
// Both the file and the archive should have been added to classpath, so
// both should be reachable via the class loader.
Assert.assertNotNull(cl.getResource("distributed.jar.inside2"));
Assert.assertNotNull(cl.getResource("distributed.jar.inside3"));
Assert.assertNotNull(cl.getResource("distributed.jar.inside4"));
// The Job Jar should have been extracted to a folder named "job.jar" and
// added to the classpath; the two jar files in the lib folder in the Job
// Jar should have also been added to the classpath
Assert.assertNotNull(cl.getResource("job.jar/"));
Assert.assertNotNull(cl.getResource("job.jar/lib/lib1.jar"));
Assert.assertNotNull(cl.getResource("job.jar/lib/lib2.jar"));
// Check that the symlink for the renaming was created in the cwd;
File symlinkFile = new File("distributed.first.symlink");
Assert.assertTrue(symlinkFile.exists());
Assert.assertEquals(1, symlinkFile.length());
// Check that the symlink for the Job Jar was created in the cwd and
// points to the extracted directory
File jobJarDir = new File("job.jar");
if (Shell.WINDOWS) {
Assert.assertTrue(isWindowsSymlinkedDirectory(jobJarDir));
} else {
Assert.assertTrue(FileUtils.isSymlink(jobJarDir));
Assert.assertTrue(jobJarDir.isDirectory());
}
}
/**
* Used on Windows to determine if the specified file is a symlink that
* targets a directory. On most platforms, these checks can be done using
* commons-io. On Windows, the commons-io implementation is unreliable and
* always returns false. Instead, this method checks the output of the dir
* command. After migrating to Java 7, this method can be removed in favor
* of the new method java.nio.file.Files.isSymbolicLink, which is expected to
* work cross-platform.
*
* @param file File to check
* @return boolean true if the file is a symlink that targets a directory
* @throws IOException thrown for any I/O error
*/
private static boolean isWindowsSymlinkedDirectory(File file)
throws IOException {
String dirOut = Shell.execCommand("cmd", "/c", "dir",
file.getAbsoluteFile().getParent());
StringReader sr = new StringReader(dirOut);
BufferedReader br = new BufferedReader(sr);
try {
String line = br.readLine();
while (line != null) {
line = br.readLine();
if (line.contains(file.getName()) && line.contains("<SYMLINKD>")) {
return true;
}
}
return false;
} finally {
IOUtils.closeStream(br);
IOUtils.closeStream(sr);
}
}
/**
* Returns a mapping of the final component of each path to the corresponding
* Path instance. This assumes that every given Path has a unique string in
* the final path component, which is true for these tests.
*
* @param paths Path[] to map
* @return Map<String, Path> mapping the final component of each path to the
* corresponding Path instance
*/
private static Map<String, Path> pathsToMap(Path[] paths) {
Map<String, Path> map = new HashMap<String, Path>();
for (Path path: paths) {
map.put(path.getName(), path);
}
return map;
}
}
private void testDistributedCache(String jobJarPath, boolean withWildcard)
throws Exception {
if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
LOG.info("MRAppJar " + MiniMRYarnCluster.APPJAR
+ " not found. Not running test.");
return;
}
// Create a temporary file of length 1.
Path first = createTempFile("distributed.first", "x");
// Create three jars with a single file inside them.
Path second =
makeJar(new Path(TEST_ROOT_DIR, "distributed.second.jar"), 2);
Path third =
makeJar(new Path(TEST_ROOT_DIR, "distributed.third.jar"), 3);
Path fourth =
makeJar(new Path(TEST_ROOT_DIR, "distributed.fourth.jar"), 4);
Job job = Job.getInstance(mrCluster.getConfig());
// Set the job jar to a new "dummy" jar so we can check that its extracted
// properly
job.setJar(jobJarPath);
if (withWildcard) {
// If testing with wildcards, upload the DistributedCacheChecker into HDFS
// and add the directory as a wildcard.
Path libs = new Path("testLibs");
Path wildcard = remoteFs.makeQualified(new Path(libs, "*"));
remoteFs.mkdirs(libs);
remoteFs.copyFromLocalFile(third, libs);
job.addCacheFile(wildcard.toUri());
} else {
// Otherwise add the DistributedCacheChecker directly to the classpath.
// Because the job jar is a "dummy" jar, we need to include the jar with
// DistributedCacheChecker or it won't be able to find it
Path distributedCacheCheckerJar = new Path(
JarFinder.getJar(DistributedCacheChecker.class));
job.addFileToClassPath(localFs.makeQualified(distributedCacheCheckerJar));
}
job.setMapperClass(DistributedCacheChecker.class);
job.setOutputFormatClass(NullOutputFormat.class);
FileInputFormat.setInputPaths(job, first);
// Creates the Job Configuration
job.addCacheFile(
new URI(first.toUri().toString() + "#distributed.first.symlink"));
job.addFileToClassPath(second);
// The AppMaster jar itself
job.addFileToClassPath(
APP_JAR.makeQualified(localFs.getUri(), APP_JAR.getParent()));
job.addArchiveToClassPath(third);
job.addCacheArchive(fourth.toUri());
job.setMaxMapAttempts(1); // speed up failures
job.submit();
String trackingUrl = job.getTrackingURL();
String jobId = job.getJobID().toString();
Assert.assertTrue(job.waitForCompletion(false));
Assert.assertTrue("Tracking URL was " + trackingUrl +
" but didn't Match Job ID " + jobId ,
trackingUrl.endsWith(jobId.substring(jobId.lastIndexOf("_")) + "/"));
}
private void testDistributedCache(boolean withWildcard) throws Exception {
// Test with a local (file:///) Job Jar
Path localJobJarPath = makeJobJarWithLib(TEST_ROOT_DIR.toUri().toString());
testDistributedCache(localJobJarPath.toUri().toString(), withWildcard);
// Test with a remote (hdfs://) Job Jar
Path remoteJobJarPath = new Path(remoteFs.getUri().toString() + "/",
localJobJarPath.getName());
remoteFs.moveFromLocalFile(localJobJarPath, remoteJobJarPath);
File localJobJarFile = new File(localJobJarPath.toUri().toString());
if (localJobJarFile.exists()) { // just to make sure
localJobJarFile.delete();
}
testDistributedCache(remoteJobJarPath.toUri().toString(), withWildcard);
}
@Test (timeout = 300000)
public void testDistributedCache() throws Exception {
testDistributedCache(false);
}
@Test (timeout = 300000)
public void testDistributedCacheWithWildcards() throws Exception {
testDistributedCache(true);
}
@Test(timeout = 120000)
public void testThreadDumpOnTaskTimeout() throws IOException,
InterruptedException, ClassNotFoundException {
if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
LOG.info("MRAppJar " + MiniMRYarnCluster.APPJAR
+ " not found. Not running test.");
return;
}
final SleepJob sleepJob = new SleepJob();
final JobConf sleepConf = new JobConf(mrCluster.getConfig());
sleepConf.setLong(MRJobConfig.TASK_TIMEOUT, 3 * 1000L);
sleepConf.setInt(MRJobConfig.MAP_MAX_ATTEMPTS, 1);
sleepJob.setConf(sleepConf);
if (this instanceof TestUberAM) {
sleepConf.setInt(MRJobConfig.MR_AM_TO_RM_HEARTBEAT_INTERVAL_MS,
30 * 1000);
}
// sleep for 10 seconds to trigger a kill with thread dump
final Job job = sleepJob.createJob(1, 0, 10 * 60 * 1000L, 1, 0L, 0);
job.setJarByClass(SleepJob.class);
job.addFileToClassPath(APP_JAR); // The AppMaster jar itself.
job.waitForCompletion(true);
final JobId jobId = TypeConverter.toYarn(job.getJobID());
final ApplicationId appID = jobId.getAppId();
int pollElapsed = 0;
while (true) {
Thread.sleep(1000);
pollElapsed += 1000;
if (TERMINAL_RM_APP_STATES.contains(mrCluster.getResourceManager()
.getRMContext().getRMApps().get(appID).getState())) {
break;
}
if (pollElapsed >= 60000) {
LOG.warn("application did not reach terminal state within 60 seconds");
break;
}
}
// Job finished, verify logs
//
final String appIdStr = appID.toString();
final String appIdSuffix = appIdStr.substring("application_".length(),
appIdStr.length());
final String containerGlob = "container_" + appIdSuffix + "_*_*";
final String syslogGlob = appIdStr
+ Path.SEPARATOR + containerGlob
+ Path.SEPARATOR + TaskLog.LogName.SYSLOG;
int numAppMasters = 0;
int numMapTasks = 0;
for (int i = 0; i < NUM_NODE_MGRS; i++) {
final Configuration nmConf = mrCluster.getNodeManager(i).getConfig();
for (String logDir :
nmConf.getTrimmedStrings(YarnConfiguration.NM_LOG_DIRS)) {
final Path absSyslogGlob =
new Path(logDir + Path.SEPARATOR + syslogGlob);
LOG.info("Checking for glob: " + absSyslogGlob);
for (FileStatus syslog : localFs.globStatus(absSyslogGlob)) {
boolean foundAppMaster = false;
boolean foundThreadDump = false;
// Determine the container type
final BufferedReader syslogReader = new BufferedReader(
new InputStreamReader(localFs.open(syslog.getPath())));
try {
for (String line; (line = syslogReader.readLine()) != null; ) {
if (line.contains(MRAppMaster.class.getName())) {
foundAppMaster = true;
break;
}
}
} finally {
syslogReader.close();
}
// Check for thread dump in stdout
final Path stdoutPath = new Path(syslog.getPath().getParent(),
TaskLog.LogName.STDOUT.toString());
final BufferedReader stdoutReader = new BufferedReader(
new InputStreamReader(localFs.open(stdoutPath)));
try {
for (String line; (line = stdoutReader.readLine()) != null; ) {
if (line.contains("Full thread dump")) {
foundThreadDump = true;
break;
}
}
} finally {
stdoutReader.close();
}
if (foundAppMaster) {
numAppMasters++;
if (this instanceof TestUberAM) {
Assert.assertTrue("No thread dump", foundThreadDump);
} else {
Assert.assertFalse("Unexpected thread dump", foundThreadDump);
}
} else {
numMapTasks++;
Assert.assertTrue("No thread dump", foundThreadDump);
}
}
}
}
// Make sure we checked non-empty set
//
Assert.assertEquals("No AppMaster log found!", 1, numAppMasters);
if (sleepConf.getBoolean(MRJobConfig.JOB_UBERTASK_ENABLE, false)) {
Assert.assertSame("MapTask log with uber found!", 0, numMapTasks);
} else {
Assert.assertSame("No MapTask log found!", 1, numMapTasks);
}
}
private Path createTempFile(String filename, String contents)
throws IOException {
Path path = new Path(TEST_ROOT_DIR, filename);
FSDataOutputStream os = localFs.create(path);
os.writeBytes(contents);
os.close();
localFs.setPermission(path, new FsPermission("700"));
return path;
}
private Path makeJar(Path p, int index) throws FileNotFoundException,
IOException {
FileOutputStream fos =
new FileOutputStream(new File(p.toUri().getPath()));
JarOutputStream jos = new JarOutputStream(fos);
ZipEntry ze = new ZipEntry("distributed.jar.inside" + index);
jos.putNextEntry(ze);
jos.write(("inside the jar!" + index).getBytes());
jos.closeEntry();
jos.close();
localFs.setPermission(p, new FsPermission("700"));
return p;
}
private Path makeJobJarWithLib(String testDir) throws FileNotFoundException,
IOException{
Path jobJarPath = new Path(testDir, "thejob.jar");
FileOutputStream fos =
new FileOutputStream(new File(jobJarPath.toUri().getPath()));
JarOutputStream jos = new JarOutputStream(fos);
// Have to put in real jar files or it will complain
createAndAddJarToJar(jos, new File(
new Path(testDir, "lib1.jar").toUri().getPath()));
createAndAddJarToJar(jos, new File(
new Path(testDir, "lib2.jar").toUri().getPath()));
jos.close();
localFs.setPermission(jobJarPath, new FsPermission("700"));
return jobJarPath;
}
private void createAndAddJarToJar(JarOutputStream jos, File jarFile)
throws FileNotFoundException, IOException {
FileOutputStream fos2 = new FileOutputStream(jarFile);
JarOutputStream jos2 = new JarOutputStream(fos2);
// Have to have at least one entry or it will complain
ZipEntry ze = new ZipEntry("lib1.inside");
jos2.putNextEntry(ze);
jos2.closeEntry();
jos2.close();
ze = new ZipEntry("lib/" + jarFile.getName());
jos.putNextEntry(ze);
FileInputStream in = new FileInputStream(jarFile);
byte buf[] = new byte[1024];
int numRead;
do {
numRead = in.read(buf);
if (numRead >= 0) {
jos.write(buf, 0, numRead);
}
} while (numRead != -1);
in.close();
jos.closeEntry();
jarFile.delete();
}
@Test
public void testSharedCache() throws Exception {
Path localJobJarPath = makeJobJarWithLib(TEST_ROOT_DIR.toUri().toString());
if (!(new File(MiniMRYarnCluster.APPJAR)).exists()) {
LOG.info("MRAppJar " + MiniMRYarnCluster.APPJAR
+ " not found. Not running test.");
return;
}
Job job = Job.getInstance(mrCluster.getConfig());
Configuration jobConf = job.getConfiguration();
jobConf.set(MRJobConfig.SHARED_CACHE_MODE, "enabled");
Path inputFile = createTempFile("input-file", "x");
// Create jars with a single file inside them.
Path second = makeJar(new Path(TEST_ROOT_DIR, "distributed.second.jar"), 2);
Path third = makeJar(new Path(TEST_ROOT_DIR, "distributed.third.jar"), 3);
Path fourth = makeJar(new Path(TEST_ROOT_DIR, "distributed.fourth.jar"), 4);
// Add libjars to job conf
jobConf.set("tmpjars", second.toString() + "," + third.toString() + ","
+ fourth.toString());
// Because the job jar is a "dummy" jar, we need to include the jar with
// DistributedCacheChecker or it won't be able to find it
Path distributedCacheCheckerJar =
new Path(JarFinder.getJar(SharedCacheChecker.class));
job.addFileToClassPath(distributedCacheCheckerJar.makeQualified(
localFs.getUri(), distributedCacheCheckerJar.getParent()));
job.setMapperClass(SharedCacheChecker.class);
job.setOutputFormatClass(NullOutputFormat.class);
FileInputFormat.setInputPaths(job, inputFile);
job.setMaxMapAttempts(1); // speed up failures
job.submit();
String trackingUrl = job.getTrackingURL();
String jobId = job.getJobID().toString();
Assert.assertTrue(job.waitForCompletion(true));
Assert.assertTrue("Tracking URL was " + trackingUrl
+ " but didn't Match Job ID " + jobId,
trackingUrl.endsWith(jobId.substring(jobId.lastIndexOf("_")) + "/"));
}
/**
* An identity mapper for testing the shared cache.
*/
public static class SharedCacheChecker extends
Mapper<LongWritable, Text, NullWritable, NullWritable> {
@Override
public void setup(Context context) throws IOException {
}
}
public static class ConfVerificationMapper extends SleepMapper {
@Override
protected void setup(Context context)
throws IOException, InterruptedException {
super.setup(context);
final Configuration conf = context.getConfiguration();
// check if the job classloader is enabled and verify the TCCL
if (conf.getBoolean(MRJobConfig.MAPREDUCE_JOB_CLASSLOADER, false)) {
ClassLoader tccl = Thread.currentThread().getContextClassLoader();
if (!(tccl instanceof ApplicationClassLoader)) {
throw new IOException("TCCL expected: " +
ApplicationClassLoader.class.getName() + ", actual: " +
tccl.getClass().getName());
}
}
final String ioSortMb = conf.get(MRJobConfig.IO_SORT_MB);
if (!TEST_IO_SORT_MB.equals(ioSortMb)) {
throw new IOException("io.sort.mb expected: " + TEST_IO_SORT_MB
+ ", actual: " + ioSortMb);
}
}
}
@Test
public void testSleepJobName() throws IOException {
SleepJob sleepJob = new SleepJob();
sleepJob.setConf(conf);
Job job1 = sleepJob.createJob(1, 1, 1, 1, 1, 1);
assertThat(job1.getJobName())
.withFailMessage("Wrong default name of sleep job.")
.isEqualTo(SleepJob.SLEEP_JOB_NAME);
String expectedJob2Name = SleepJob.SLEEP_JOB_NAME + " - test";
Job job2 = sleepJob.createJob(1, 1, 1, 1, 1, 1, "test");
assertThat(job2.getJobName())
.withFailMessage("Wrong name of sleep job.")
.isEqualTo(expectedJob2Name);
}
}