@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) {
      // replace the shuffle handler with one stubbed for testing
      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 {
          // Only set response headers and skip everything else
          // send some dummy value for content-length
          super.setResponseHeaders(response, keepAliveParam, 100);
        }
        @Override
        protected void verifyRequest(String appid,
            ChannelHandlerContext ctx, HttpRequest request,
            HttpResponse response, URL requestUri) throws IOException {
          // Do nothing.
        }
        @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 {
          // send a shuffle header
          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://127.0.0.1:"
                + 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) {
      // ignore
    }
    Assert.assertEquals("sendError called due to shuffle error",
        0, failures.size());
  } finally {
    shuffleHandler.stop();
    FileUtil.fullyDelete(ABS_LOG_DIR);
  }
}