ShuffleHandler extends AuxiliaryService { @Override public void initializeApplication(ApplicationInitializationContext context) { String user = context.getUser(); ApplicationId appId = context.getApplicationId(); ByteBuffer secret = context.getApplicationDataForService(); try { Token<JobTokenIdentifier> jt = deserializeServiceData(secret); JobID jobId = new JobID(Long.toString(appId.getClusterTimestamp()), appId.getId()); userRsrc.put(jobId.toString(), user); LOG.info("Added token for " + jobId.toString()); secretManager.addTokenForJob(jobId.toString(), jt); } catch (IOException e) { LOG.error("Error during initApp", e); } } ShuffleHandler(MetricsSystem ms);  ShuffleHandler(); static ByteBuffer serializeMetaData(int port); static int deserializeMetaData(ByteBuffer meta); static ByteBuffer serializeServiceData(Token<JobTokenIdentifier> jobToken); @Override void initializeApplication(ApplicationInitializationContext context); @Override void stopApplication(ApplicationTerminationContext context); @Override synchronized ByteBuffer getMetaData(); static final String SHUFFLE_MANAGE_OS_CACHE; static final boolean DEFAULT_SHUFFLE_MANAGE_OS_CACHE; static final String SHUFFLE_READAHEAD_BYTES; static final int DEFAULT_SHUFFLE_READAHEAD_BYTES; static final String MAPREDUCE_SHUFFLE_SERVICEID; static final String SHUFFLE_PORT_CONFIG_KEY; static final int DEFAULT_SHUFFLE_PORT; static final String SHUFFLE_CONNECTION_KEEP_ALIVE_ENABLED; static final boolean DEFAULT_SHUFFLE_CONNECTION_KEEP_ALIVE_ENABLED; static final String SHUFFLE_CONNECTION_KEEP_ALIVE_TIME_OUT; static final int DEFAULT_SHUFFLE_CONNECTION_KEEP_ALIVE_TIME_OUT; static final String SHUFFLE_MAPOUTPUT_META_INFO_CACHE_SIZE; static final int DEFAULT_SHUFFLE_MAPOUTPUT_META_INFO_CACHE_SIZE; static final String CONNECTION_CLOSE; static final String SUFFLE_SSL_FILE_BUFFER_SIZE_KEY; static final int DEFAULT_SUFFLE_SSL_FILE_BUFFER_SIZE; static final String MAX_SHUFFLE_CONNECTIONS; static final int DEFAULT_MAX_SHUFFLE_CONNECTIONS; static final String MAX_SHUFFLE_THREADS; static final int DEFAULT_MAX_SHUFFLE_THREADS; static final String SHUFFLE_BUFFER_SIZE; static final int DEFAULT_SHUFFLE_BUFFER_SIZE; static final String SHUFFLE_TRANSFERTO_ALLOWED; static final boolean DEFAULT_SHUFFLE_TRANSFERTO_ALLOWED; static final boolean WINDOWS_DEFAULT_SHUFFLE_TRANSFERTO_ALLOWED; }
@Test(timeout = 100000) public void testMapFileAccess() throws IOException { assumeTrue(NativeIO.isAvailable()); Configuration conf = new Configuration(); conf.setInt(ShuffleHandler.SHUFFLE_PORT_CONFIG_KEY, 0); conf.setInt(ShuffleHandler.MAX_SHUFFLE_CONNECTIONS, 3); conf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION, "kerberos"); UserGroupInformation.setConfiguration(conf); File absLogDir = new File("target", TestShuffleHandler.class.getSimpleName() + "LocDir").getAbsoluteFile(); conf.set(YarnConfiguration.NM_LOCAL_DIRS, absLogDir.getAbsolutePath()); ApplicationId appId = ApplicationId.newInstance(12345, 1); LOG.info(appId.toString()); String appAttemptId = "attempt_12345_1_m_1_0"; String user = "randomUser"; String reducerId = "0"; List<File> fileMap = new ArrayList<File>(); createShuffleHandlerFiles(absLogDir, user, appId.toString(), appAttemptId, conf, fileMap); ShuffleHandler shuffleHandler = new ShuffleHandler() { @Override protected Shuffle getShuffle(Configuration conf) { return new Shuffle(conf) { @Override protected void verifyRequest(String appid, ChannelHandlerContext ctx, HttpRequest request, HttpResponse response, URL requestUri) throws IOException { } }; } }; shuffleHandler.init(conf); try { shuffleHandler.start(); DataOutputBuffer outputBuffer = new DataOutputBuffer(); outputBuffer.reset(); Token<JobTokenIdentifier> jt = new Token<JobTokenIdentifier>("identifier".getBytes(), "password".getBytes(), new Text(user), new Text("shuffleService")); jt.write(outputBuffer); shuffleHandler .initializeApplication(new ApplicationInitializationContext(user, appId, ByteBuffer.wrap(outputBuffer.getData(), 0, outputBuffer.getLength()))); URL url = new URL( "http: + shuffleHandler.getConfig().get( ShuffleHandler.SHUFFLE_PORT_CONFIG_KEY) + "/mapOutput?job=job_12345_0001&reduce=" + reducerId + "&map=attempt_12345_1_m_1_0"); HttpURLConnection conn = (HttpURLConnection) url.openConnection(); conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_NAME, ShuffleHeader.DEFAULT_HTTP_HEADER_NAME); conn.setRequestProperty(ShuffleHeader.HTTP_HEADER_VERSION, ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION); conn.connect(); byte[] byteArr = new byte[10000]; try { DataInputStream is = new DataInputStream(conn.getInputStream()); is.readFully(byteArr); } catch (EOFException e) { } FileInputStream is = new FileInputStream(fileMap.get(0)); String owner = NativeIO.POSIX.getFstat(is.getFD()).getOwner(); is.close(); String message = "Owner '" + owner + "' for path " + fileMap.get(0).getAbsolutePath() + " did not match expected owner '" + user + "'"; Assert.assertTrue((new String(byteArr)).contains(message)); } finally { shuffleHandler.stop(); } }