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()); recordJobShuffleInfo(jobId, user, 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 MAX_WEIGHT; static final int DEFAULT_MAX_WEIGHT; static final String EXPIRE_AFTER_ACCESS_MINUTES; static final int DEFAULT_EXPIRE_AFTER_ACCESS_MINUTES; static final String CONCURRENCY_LEVEL; static final int DEFAULT_CONCURRENCY_LEVEL; static final HttpResponseStatus TOO_MANY_REQ_STATUS; static final long FETCH_RETRY_DELAY; static final String RETRY_AFTER_HEADER; static final String MAPREDUCE_SHUFFLE_SERVICEID; static final String SHUFFLE_PORT_CONFIG_KEY; static final int DEFAULT_SHUFFLE_PORT; static final String SHUFFLE_LISTEN_QUEUE_SIZE; static final int DEFAULT_SHUFFLE_LISTEN_QUEUE_SIZE; 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; static final String SHUFFLE_MAX_SESSION_OPEN_FILES; static final int DEFAULT_SHUFFLE_MAX_SESSION_OPEN_FILES; }
@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); conf.set(YarnConfiguration.NM_LOCAL_DIRS, ABS_LOG_DIR.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(ABS_LOG_DIR, 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 { } }; } }; AuxiliaryLocalPathHandler pathHandler = new TestAuxiliaryLocalPathHandler(); shuffleHandler.setAuxiliaryLocalPathHandler(pathHandler); 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(); FileUtil.fullyDelete(ABS_LOG_DIR); } }
@Test(timeout = 100000) public void testGetMapOutputInfo() throws Exception { final ArrayList<Throwable> failures = new ArrayList<Throwable>(1); 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, "simple"); UserGroupInformation.setConfiguration(conf); conf.set(YarnConfiguration.NM_LOCAL_DIRS, ABS_LOG_DIR.getAbsolutePath()); ApplicationId appId = ApplicationId.newInstance(12345, 1); String appAttemptId = "attempt_12345_1_m_1_0"; String user = "randomUser"; String reducerId = "0"; List<File> fileMap = new ArrayList<File>(); createShuffleHandlerFiles(ABS_LOG_DIR, user, appId.toString(), appAttemptId, conf, fileMap); AuxiliaryLocalPathHandler pathHandler = new TestAuxiliaryLocalPathHandler(); ShuffleHandler shuffleHandler = new ShuffleHandler() { @Override protected Shuffle getShuffle(Configuration conf) { return new Shuffle(conf) { @Override protected void populateHeaders(List<String> mapIds, String outputBaseStr, String user, int reduce, HttpRequest request, HttpResponse response, boolean keepAliveParam, Map<String, MapOutputInfo> infoMap) throws IOException { super.setResponseHeaders(response, keepAliveParam, 100); } @Override protected void verifyRequest(String appid, ChannelHandlerContext ctx, HttpRequest request, HttpResponse response, URL requestUri) throws IOException { } @Override protected void sendError(ChannelHandlerContext ctx, String message, HttpResponseStatus status) { if (failures.size() == 0) { failures.add(new Error(message)); ctx.getChannel().close(); } } @Override protected ChannelFuture sendMapOutput(ChannelHandlerContext ctx, Channel ch, String user, String mapId, int reduce, MapOutputInfo info) throws IOException { ShuffleHeader header = new ShuffleHeader("attempt_12345_1_m_1_0", 5678, 5678, 1); DataOutputBuffer dob = new DataOutputBuffer(); header.write(dob); return ch.write(wrappedBuffer(dob.getData(), 0, dob.getLength())); } }; } }; shuffleHandler.setAuxiliaryLocalPathHandler(pathHandler); 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(); try { DataInputStream is = new DataInputStream(conn.getInputStream()); ShuffleHeader header = new ShuffleHeader(); header.readFields(is); is.close(); } catch (EOFException e) { } Assert.assertEquals("sendError called due to shuffle error", 0, failures.size()); } finally { shuffleHandler.stop(); FileUtil.fullyDelete(ABS_LOG_DIR); } }