@Test(timeout = 10000)
public void testContainerLogsForFinishedApps() throws Exception {

  String fileName = "syslog";
  String user = "user1";
  NodeId nodeId = NodeId.newInstance("test host", 100);
  NodeId nodeId2 = NodeId.newInstance("host2", 1234);
  ApplicationId appId = ApplicationId.newInstance(0, 1);
  ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(
      appId, 1);
  ContainerId containerId1 = ContainerId.newContainerId(appAttemptId, 1);
  ContainerId containerId100 = ContainerId.newContainerId(appAttemptId, 100);
  TestContainerLogsUtils.createContainerLogFileInRemoteFS(conf, fs,
      rootLogDir, appId, Collections.singletonMap(containerId1,
          "Hello." + containerId1),
      nodeId, fileName, user, true);
  TestContainerLogsUtils.createContainerLogFileInRemoteFS(conf, fs,
      rootLogDir, appId, Collections.singletonMap(containerId100,
          "Hello." + containerId100),
      nodeId2, fileName, user, false);
  // test whether we can find container log from remote diretory if
  // the containerInfo for this container could be fetched from AHS.
  WebResource r = resource();
  ClientResponse response = r.path("ws").path("v1")
      .path("applicationhistory").path("containerlogs")
      .path(containerId1.toString()).path(fileName)
      .queryParam("user.name", user)
      .accept(MediaType.TEXT_PLAIN)
      .get(ClientResponse.class);
  String responseText = response.getEntity(String.class);
  assertTrue(responseText.contains("Hello." + containerId1));
  // Do the same test with new API
  r = resource();
  response = r.path("ws").path("v1")
      .path("applicationhistory").path("containers")
      .path(containerId1.toString()).path("logs").path(fileName)
      .queryParam("user.name", user)
      .accept(MediaType.TEXT_PLAIN)
      .get(ClientResponse.class);
  responseText = response.getEntity(String.class);
  assertTrue(responseText.contains("Hello." + containerId1));
  // test whether we can find container log from remote diretory if
  // the containerInfo for this container could not be fetched from AHS.
  r = resource();
  response = r.path("ws").path("v1")
      .path("applicationhistory").path("containerlogs")
      .path(containerId100.toString()).path(fileName)
      .queryParam("user.name", user)
      .accept(MediaType.TEXT_PLAIN)
      .get(ClientResponse.class);
  responseText = response.getEntity(String.class);
  assertTrue(responseText.contains("Hello." + containerId100));
  // Do the same test with new API
  r = resource();
  response = r.path("ws").path("v1")
      .path("applicationhistory").path("containers")
      .path(containerId100.toString()).path("logs").path(fileName)
      .queryParam("user.name", user)
      .accept(MediaType.TEXT_PLAIN)
      .get(ClientResponse.class);
  responseText = response.getEntity(String.class);
  assertTrue(responseText.contains("Hello." + containerId100));
  // create an application which can not be found from AHS
  ApplicationId appId100 = ApplicationId.newInstance(0, 100);
  ApplicationAttemptId appAttemptId100 = ApplicationAttemptId.newInstance(
      appId100, 1);
  ContainerId containerId1ForApp100 = ContainerId.newContainerId(
      appAttemptId100, 1);
  TestContainerLogsUtils.createContainerLogFileInRemoteFS(conf, fs,
      rootLogDir, appId100,
      Collections.singletonMap(containerId1ForApp100,
          "Hello." + containerId1ForApp100),
      nodeId, fileName, user, true);
  r = resource();
  response = r.path("ws").path("v1")
      .path("applicationhistory").path("containerlogs")
      .path(containerId1ForApp100.toString()).path(fileName)
      .queryParam("user.name", user)
      .accept(MediaType.TEXT_PLAIN)
      .get(ClientResponse.class);
  responseText = response.getEntity(String.class);
  assertTrue(responseText.contains("Hello." + containerId1ForApp100));
  int fullTextSize = responseText.getBytes().length;
  String tailEndSeparator = StringUtils.repeat("*",
      "End of LogType:syslog".length() + 50) + "\n\n";
  int tailTextSize = "\nEnd of LogType:syslog\n".getBytes().length
      + tailEndSeparator.getBytes().length;
  String logMessage = "Hello." + containerId1ForApp100;
  int fileContentSize = logMessage.getBytes().length;
  // specify how many bytes we should get from logs
  // if we specify a position number, it would get the first n bytes from
  // container log
  r = resource();
  response = r.path("ws").path("v1")
      .path("applicationhistory").path("containerlogs")
      .path(containerId1ForApp100.toString()).path(fileName)
      .queryParam("user.name", user)
      .queryParam("size", "5")
      .accept(MediaType.TEXT_PLAIN)
      .get(ClientResponse.class);
  responseText = response.getEntity(String.class);
  assertEquals(responseText.getBytes().length,
      (fullTextSize - fileContentSize) + 5);
  assertTrue(fullTextSize >= responseText.getBytes().length);
  assertEquals(new String(responseText.getBytes(),
      (fullTextSize - fileContentSize - tailTextSize), 5),
      new String(logMessage.getBytes(), 0, 5));
  // specify how many bytes we should get from logs
  // if we specify a negative number, it would get the last n bytes from
  // container log
  r = resource();
  response = r.path("ws").path("v1")
      .path("applicationhistory").path("containerlogs")
      .path(containerId1ForApp100.toString()).path(fileName)
      .queryParam("user.name", user)
      .queryParam("size", "-5")
      .accept(MediaType.TEXT_PLAIN)
      .get(ClientResponse.class);
  responseText = response.getEntity(String.class);
  assertEquals(responseText.getBytes().length,
      (fullTextSize - fileContentSize) + 5);
  assertTrue(fullTextSize >= responseText.getBytes().length);
  assertEquals(new String(responseText.getBytes(),
      (fullTextSize - fileContentSize - tailTextSize), 5),
      new String(logMessage.getBytes(), fileContentSize - 5, 5));
  // specify the bytes which is larger than the actual file size,
  // we would get the full logs
  r = resource();
  response = r.path("ws").path("v1")
      .path("applicationhistory").path("containerlogs")
      .path(containerId1ForApp100.toString()).path(fileName)
      .queryParam("user.name", user)
      .queryParam("size", "10000")
      .accept(MediaType.TEXT_PLAIN)
      .get(ClientResponse.class);
  responseText = response.getEntity(String.class);
  assertThat(responseText.getBytes()).hasSize(fullTextSize);
  r = resource();
  response = r.path("ws").path("v1")
      .path("applicationhistory").path("containerlogs")
      .path(containerId1ForApp100.toString()).path(fileName)
      .queryParam("user.name", user)
      .queryParam("size", "-10000")
      .accept(MediaType.TEXT_PLAIN)
      .get(ClientResponse.class);
  responseText = response.getEntity(String.class);
  assertThat(responseText.getBytes()).hasSize(fullTextSize);
}