gerashegalov commented on code in PR #8854:
URL: https://github.com/apache/hadoop/pull/8854#discussion_r4503462904
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java:
##########
@@ -533,6 +536,96 @@ public void testGetContainerReport() throws YarnException,
IOException {
}
}
+ @Test
+ public void testGetContainerReportForContainerFromPreviousAttempt()
Review Comment:
This test fakes the problem by mocking an empty attempts map, so it doesn't
show how a real RM gets there. The name is also misleading: a container from a
previous attempt already works today, as long as that attempt is still in
memory.
Could we add an end-to-end test in `TestAMRestart`? The steps are:
1. Turn on keepContainers and work-preserving recovery. Set max attempts to
2 and a short failure validity interval.
2. Start a container in attempt 1.
3. Fail attempts 1, 2 and 3, waiting past the interval each time, so attempt
1 is deleted from the store.
4. Restart the RM and have the NM report the container again.
5. Check that `getContainerReport` returns it.
I tried this locally. It passes with your change and fails on trunk with
`ApplicationAttemptNotFoundException`, for both the Capacity and Fair
schedulers, in about 5 seconds each. Before the restart it passes on both.
Here's a shortened version of the test I ran, if it helps. I haven't re-run
this shorter version, and it needs an import for `GetContainerReportRequest`:
<details><summary>Test sketch</summary>
```java
@ParameterizedTest(name = "{0}")
@MethodSource("getParameters")
@Timeout(value = 120)
public void testContainerReportForAttemptRemovedFromStateStore(
SchedulerType type) throws Exception {
initTestAMRestart(type);
getConf().setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, 2);
getConf().setBoolean(YarnConfiguration.RECOVERY_ENABLED, true);
getConf().setBoolean(
YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_ENABLED, true);
getConf().setLong(
YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_SCHEDULING_WAIT_MS, 0);
getConf().set(YarnConfiguration.RM_STORE,
MemoryRMStateStore.class.getName());
final long window = 1000;
MockRM rm1 = new MockRM(getConf());
rm1.start();
MemoryRMStateStore memStore = (MemoryRMStateStore) rm1.getRMStateStore();
MockNM nm1 =
new MockNM("127.0.0.1:1234", 8000, rm1.getResourceTrackerService());
nm1.registerNode();
RMApp app1 = MockRMAppSubmitter.submit(rm1,
MockRMAppSubmissionData.Builder.createWithMemory(200, rm1)
.withAppName("")
.withUser(UserGroupInformation.getCurrentUser().getShortUserName())
.withKeepContainers(true)
.withAttemptFailuresValidityInterval(window)
.build());
MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nm1);
ApplicationAttemptId attempt1 = am1.getApplicationAttemptId();
allocateContainers(nm1, am1, 1);
nm1.nodeHeartbeat(attempt1, 2, ContainerState.RUNNING);
ContainerId containerId2 = ContainerId.newContainerId(attempt1, 2);
rm1.waitForState(nm1, containerId2, RMContainerState.RUNNING);
// Fail attempts 1-3, each outside the validity window. Failing attempt 3
// deletes attempt 1 from the state store (removeExcessAttempts).
MockAM am = am1;
for (int i = 1; i <= 3; i++) {
nm1.nodeHeartbeat(am.getApplicationAttemptId(), 1,
ContainerState.COMPLETE);
rm1.waitForState(am.getApplicationAttemptId(), RMAppAttemptState.FAILED);
rm1.waitForState(app1.getApplicationId(), RMAppState.ACCEPTED);
if (i < 3) {
Thread.sleep(window + 500);
am = MockRM.launchAndRegisterAM(app1, rm1, nm1);
}
}
GenericTestUtils.waitFor(() -> !memStore.getState().getApplicationState()
.get(app1.getApplicationId()).attempts.containsKey(attempt1),
100, 10000);
MockRM rm2 = new MockRM(getConf(), memStore);
rm2.start();
nm1.setResourceTrackerService(rm2.getResourceTrackerService());
nm1.registerNode(Lists.newArrayList(TestRMRestart.createNMContainerStatus(
attempt1, 2, ContainerState.RUNNING)), null);
rm2.waitForState(nm1, containerId2, RMContainerState.RUNNING);
assertFalse(rm2.getRMContext().getRMApps().get(app1.getApplicationId())
.getAppAttempts().containsKey(attempt1));
assertEquals(containerId2, rm2.getClientRMService().getContainerReport(
GetContainerReportRequest.newInstance(containerId2))
.getContainerReport().getContainerId());
rm2.stop();
rm1.stop();
}
```
</details>
If you keep the mocked test too, maybe rename it to something like
`testGetContainerReportWhenAttemptMissingWithKeepContainers`.
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java:
##########
@@ -533,6 +536,96 @@ public void testGetContainerReport() throws YarnException,
IOException {
}
}
+ @Test
+ public void testGetContainerReportForContainerFromPreviousAttempt()
+ throws Exception {
+ ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(123456, 1), 1);
+ ContainerId containerId = ContainerId.newContainerId(attemptId, 1);
+ ClientRMService rmService = createRMServiceForContainerReport(
+ containerId, true, false, true);
+
+ GetContainerReportResponse response = rmService.getContainerReport(
+ GetContainerReportRequest.newInstance(containerId));
+
+ assertEquals(containerId, response.getContainerReport().getContainerId());
+ }
+
+ @Test
+ public void
testGetContainerReportRejectsMissingAttemptWithoutContainerTransfer()
+ throws Exception {
+ ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(123456, 1), 1);
+ ContainerId containerId = ContainerId.newContainerId(attemptId, 1);
+ ClientRMService rmService = createRMServiceForContainerReport(
+ containerId, false, false, true);
+
+ assertThrows(ApplicationAttemptNotFoundException.class,
+ () -> rmService.getContainerReport(
+ GetContainerReportRequest.newInstance(containerId)));
+ }
+
+ @Test
+ public void testGetContainerReportRejectsMissingRetainedContainer()
+ throws Exception {
+ ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(123456, 1), 1);
+ ContainerId containerId = ContainerId.newContainerId(attemptId, 1);
+ ClientRMService rmService = createRMServiceForContainerReport(
+ containerId, true, false, false);
+
+ assertThrows(ContainerNotFoundException.class,
+ () -> rmService.getContainerReport(
+ GetContainerReportRequest.newInstance(containerId)));
+ }
+
+ private ClientRMService createRMServiceForContainerReport(
+ ContainerId containerId, boolean keepContainersAcrossAttempts,
+ boolean includeAttempt, boolean includeContainer) {
+ ApplicationAttemptId attemptId = containerId.getApplicationAttemptId();
+ ApplicationId applicationId = attemptId.getApplicationId();
+ ResourceScheduler scheduler = mock(ResourceScheduler.class);
+ RMContext rmContext = mock(RMContext.class);
+ RMApp application = mock(RMApp.class);
+
+ ConcurrentHashMap<ApplicationId, RMApp> applications =
+ new ConcurrentHashMap<>();
+ applications.put(applicationId, application);
+ when(rmContext.getRMApps()).thenReturn(applications);
+ when(rmContext.getScheduler()).thenReturn(scheduler);
+ when(application.getApplicationId()).thenReturn(applicationId);
+ when(application.getUser()).thenReturn("test-user");
+
+ ApplicationSubmissionContext submissionContext =
+ mock(ApplicationSubmissionContext.class);
+ when(submissionContext.getKeepContainersAcrossApplicationAttempts())
+ .thenReturn(keepContainersAcrossAttempts);
+ when(application.getApplicationSubmissionContext())
+ .thenReturn(submissionContext);
+
+ Map<ApplicationAttemptId, RMAppAttempt> attempts = new HashMap<>();
+ if (includeAttempt) {
Review Comment:
Every test passes `includeAttempt = false`, so this branch never runs. Could
we add the normal case too, with keepContainers on and the attempt still
present (`true, true, true`)? That's what most keepContainers apps actually hit.
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java:
##########
@@ -533,6 +536,96 @@ public void testGetContainerReport() throws YarnException,
IOException {
}
}
+ @Test
+ public void testGetContainerReportForContainerFromPreviousAttempt()
+ throws Exception {
+ ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(123456, 1), 1);
+ ContainerId containerId = ContainerId.newContainerId(attemptId, 1);
+ ClientRMService rmService = createRMServiceForContainerReport(
+ containerId, true, false, true);
+
+ GetContainerReportResponse response = rmService.getContainerReport(
+ GetContainerReportRequest.newInstance(containerId));
+
+ assertEquals(containerId, response.getContainerReport().getContainerId());
+ }
+
+ @Test
+ public void
testGetContainerReportRejectsMissingAttemptWithoutContainerTransfer()
+ throws Exception {
+ ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(123456, 1), 1);
+ ContainerId containerId = ContainerId.newContainerId(attemptId, 1);
+ ClientRMService rmService = createRMServiceForContainerReport(
+ containerId, false, false, true);
+
+ assertThrows(ApplicationAttemptNotFoundException.class,
+ () -> rmService.getContainerReport(
+ GetContainerReportRequest.newInstance(containerId)));
+ }
+
+ @Test
+ public void testGetContainerReportRejectsMissingRetainedContainer()
+ throws Exception {
+ ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(123456, 1), 1);
+ ContainerId containerId = ContainerId.newContainerId(attemptId, 1);
+ ClientRMService rmService = createRMServiceForContainerReport(
+ containerId, true, false, false);
+
+ assertThrows(ContainerNotFoundException.class,
+ () -> rmService.getContainerReport(
+ GetContainerReportRequest.newInstance(containerId)));
+ }
+
+ private ClientRMService createRMServiceForContainerReport(
+ ContainerId containerId, boolean keepContainersAcrossAttempts,
+ boolean includeAttempt, boolean includeContainer) {
+ ApplicationAttemptId attemptId = containerId.getApplicationAttemptId();
+ ApplicationId applicationId = attemptId.getApplicationId();
+ ResourceScheduler scheduler = mock(ResourceScheduler.class);
+ RMContext rmContext = mock(RMContext.class);
+ RMApp application = mock(RMApp.class);
+
+ ConcurrentHashMap<ApplicationId, RMApp> applications =
+ new ConcurrentHashMap<>();
+ applications.put(applicationId, application);
+ when(rmContext.getRMApps()).thenReturn(applications);
+ when(rmContext.getScheduler()).thenReturn(scheduler);
+ when(application.getApplicationId()).thenReturn(applicationId);
+ when(application.getUser()).thenReturn("test-user");
+
+ ApplicationSubmissionContext submissionContext =
+ mock(ApplicationSubmissionContext.class);
+ when(submissionContext.getKeepContainersAcrossApplicationAttempts())
+ .thenReturn(keepContainersAcrossAttempts);
+ when(application.getApplicationSubmissionContext())
+ .thenReturn(submissionContext);
+
+ Map<ApplicationAttemptId, RMAppAttempt> attempts = new HashMap<>();
+ if (includeAttempt) {
+ attempts.put(attemptId, mock(RMAppAttempt.class));
+ }
+ when(application.getAppAttempts()).thenReturn(attempts);
+
+ if (includeContainer) {
+ RMContainer rmContainer = mock(RMContainer.class);
+ ContainerReport containerReport = mock(ContainerReport.class);
+ when(containerReport.getContainerId()).thenReturn(containerId);
+ when(rmContainer.createContainerReport()).thenReturn(containerReport);
+ when(scheduler.getRMContainer(containerId)).thenReturn(rmContainer);
+ }
+
+ ApplicationACLsManager applicationACLsManager =
+ mock(ApplicationACLsManager.class);
+ QueueACLsManager queueACLsManager = mock(QueueACLsManager.class);
+ when(queueACLsManager.checkAccess(any(UserGroupInformation.class),
Review Comment:
Access is always granted here, so nothing checks that the new branch still
rejects a user without access. Could the helper take a flag to deny access,
plus a test that a keepContainers app still throws the "does not have
privilege" error?
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java:
##########
@@ -512,11 +512,14 @@ public GetContainerReportResponse getContainerReport(
ApplicationAccessType.VIEW_APP, application);
GetContainerReportResponse response = null;
if (allowAccess) {
- RMAppAttempt appAttempt = application.getAppAttempts().get(appAttemptId);
- if (appAttempt == null) {
- throw new ApplicationAttemptNotFoundException(
- "ApplicationAttempt with id '" + appAttemptId +
- "' doesn't exist in RM.");
+ if (!application.getApplicationSubmissionContext()
Review Comment:
For keepContainers apps, this changes which error callers see. Previously, a
missing attempt always gave `ApplicationAttemptNotFoundException`. Now it gives
`ContainerNotFoundException` when the container is gone, including for an
attempt id that never existed.
I think that's an improvement. `YarnClientImpl#getContainerReport` only
falls back to the history server on `ApplicationNotFoundException` or
`ContainerNotFoundException`, so finished containers of old attempts can now be
found there. But it's a behavior change, so it's worth saying in the
description.
Related: `testGetContainerReportRejectsMissingRetainedContainer` (L569) is
described as "preserving" `ContainerNotFoundException`, but on trunk the same
input throws `ApplicationAttemptNotFoundException`. The test checks new
behavior, so the name and description should say so.
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java:
##########
@@ -533,6 +536,96 @@ public void testGetContainerReport() throws YarnException,
IOException {
}
}
+ @Test
+ public void testGetContainerReportForContainerFromPreviousAttempt()
+ throws Exception {
+ ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(123456, 1), 1);
+ ContainerId containerId = ContainerId.newContainerId(attemptId, 1);
+ ClientRMService rmService = createRMServiceForContainerReport(
+ containerId, true, false, true);
+
+ GetContainerReportResponse response = rmService.getContainerReport(
+ GetContainerReportRequest.newInstance(containerId));
+
+ assertEquals(containerId, response.getContainerReport().getContainerId());
+ }
+
+ @Test
+ public void
testGetContainerReportRejectsMissingAttemptWithoutContainerTransfer()
+ throws Exception {
+ ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(123456, 1), 1);
+ ContainerId containerId = ContainerId.newContainerId(attemptId, 1);
+ ClientRMService rmService = createRMServiceForContainerReport(
+ containerId, false, false, true);
+
+ assertThrows(ApplicationAttemptNotFoundException.class,
+ () -> rmService.getContainerReport(
+ GetContainerReportRequest.newInstance(containerId)));
+ }
+
+ @Test
+ public void testGetContainerReportRejectsMissingRetainedContainer()
+ throws Exception {
+ ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(
+ ApplicationId.newInstance(123456, 1), 1);
+ ContainerId containerId = ContainerId.newContainerId(attemptId, 1);
+ ClientRMService rmService = createRMServiceForContainerReport(
+ containerId, true, false, false);
+
+ assertThrows(ContainerNotFoundException.class,
+ () -> rmService.getContainerReport(
+ GetContainerReportRequest.newInstance(containerId)));
+ }
+
+ private ClientRMService createRMServiceForContainerReport(
Review Comment:
Optional: three positional booleans are hard to read at the call sites, and
the three tests repeat the same id setup. A `@ParameterizedTest` with
`@CsvSource` (or named constants) would make the cases clearer and make it easy
to add the two from the comments above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
