private static ClassLoader createJobClassLoader(final String appClasspath, final String[] systemClasses) throws IOException { try { return AccessController.doPrivileged( new PrivilegedExceptionAction<ClassLoader>() { @Override public ClassLoader run() throws MalformedURLException { return new ApplicationClassLoader(appClasspath, MRApps.class.getClassLoader(), Arrays.asList(systemClasses)); } }); } catch (PrivilegedActionException e) { Throwable t = e.getCause(); if (t instanceof MalformedURLException) { throw (MalformedURLException) t; } throw new IOException(e); } }
@Test public void testSystemClasses() { final List<String> systemClasses = Arrays.asList(StringUtils.getTrimmedStrings( ApplicationClassLoader.SYSTEM_CLASSES_DEFAULT)); for (String defaultXml : DEFAULT_XMLS) { assertTrue(defaultXml + " must be system resource", ApplicationClassLoader.isSystemClass(defaultXml, systemClasses)); } for (String klass : SYS_CLASSES) { assertTrue(klass + " must be system class", ApplicationClassLoader.isSystemClass(klass, systemClasses)); } assertFalse("/fake/Klass must not be a system class", ApplicationClassLoader.isSystemClass("/fake/Klass", systemClasses)); }
@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); } }
/** * 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"); } } }
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); }
