ActivitiesManager extends AbstractService { @VisibleForTesting public int getAppActivitiesMaxQueueLength() { return appActivitiesMaxQueueLength; } ActivitiesManager(RMContext rmContext); AppActivitiesInfo getAppActivitiesInfo(ApplicationId applicationId,
      Set<Integer> requestPriorities, Set<Long> allocationRequestIds,
      RMWSConsts.ActivitiesGroupBy groupBy, int limit, boolean summarize,
      double maxTimeInSeconds); ActivitiesInfo getActivitiesInfo(String nodeId,
      RMWSConsts.ActivitiesGroupBy groupBy); List<ActivitiesInfo> recordAndGetBulkActivitiesInfo(
      int activitiesCount, RMWSConsts.ActivitiesGroupBy groupBy); void recordNextNodeUpdateActivities(String nodeId); void turnOnAppActivitiesRecording(ApplicationId applicationId,
      double maxTime); boolean isRecordingMultiNodes(); NodeId getRecordingNodeId(SchedulerNode node); Optional<DiagnosticsCollector> getOptionalDiagnosticsCollector(); String getResourceDiagnostics(ResourceCalculator rc, Resource required,
      Resource available); static String getDiagnostics(Optional<DiagnosticsCollector> dcOpt); @VisibleForTesting int getAppActivitiesMaxQueueLength(); static final NodeId EMPTY_NODE_ID; static final char DIAGNOSTICS_DETAILS_SEPARATOR; static final String EMPTY_DIAGNOSTICS; }
@Test (timeout = 10000) public void testAppActivitiesMaxQueueLengthUpdate() throws TimeoutException, InterruptedException { Configuration conf = new Configuration(); int configuredAppActivitiesMaxQueueLength = 1; conf.setInt(YarnConfiguration. RM_ACTIVITIES_MANAGER_APP_ACTIVITIES_MAX_QUEUE_LENGTH, configuredAppActivitiesMaxQueueLength); conf.setInt(YarnConfiguration.RM_ACTIVITIES_MANAGER_CLEANUP_INTERVAL_MS, 500); ConcurrentMap<NodeId, RMNode> mockNodes = new ConcurrentHashMap<>(); int numNodes = 5; for (int i = 0; i < numNodes; i++) { mockNodes.put(NodeId.newInstance("node" + i, 0), mock(RMNode.class)); } CapacityScheduler cs = Mockito.mock(CapacityScheduler.class); RMContext mockRMContext = Mockito.mock(RMContext.class); Mockito.when(mockRMContext.getRMNodes()).thenReturn(mockNodes); Mockito.when(mockRMContext.getYarnConfiguration()).thenReturn(conf); Mockito.when(mockRMContext.getScheduler()).thenReturn(cs); Mockito.when(cs.isMultiNodePlacementEnabled()).thenReturn(false); int numAsyncSchedulerThreads = 3; Mockito.when(cs.getNumAsyncSchedulerThreads()) .thenReturn(numAsyncSchedulerThreads); ActivitiesManager newActivitiesManager = new ActivitiesManager(mockRMContext); Assert.assertEquals(1, newActivitiesManager.getAppActivitiesMaxQueueLength()); newActivitiesManager.init(conf); newActivitiesManager.start(); GenericTestUtils.waitFor( () -> newActivitiesManager.getAppActivitiesMaxQueueLength() == numNodes * numAsyncSchedulerThreads, 100, 3000); Assert.assertEquals(15, newActivitiesManager.getAppActivitiesMaxQueueLength()); Mockito.when(cs.getNumAsyncSchedulerThreads()).thenReturn(0); GenericTestUtils.waitFor( () -> newActivitiesManager.getAppActivitiesMaxQueueLength() == numNodes * 1.2, 100, 3000); Assert.assertEquals(6, newActivitiesManager.getAppActivitiesMaxQueueLength()); Mockito.when(cs.isMultiNodePlacementEnabled()).thenReturn(true); GenericTestUtils.waitFor( () -> newActivitiesManager.getAppActivitiesMaxQueueLength() == configuredAppActivitiesMaxQueueLength, 100, 3000); Assert.assertEquals(1, newActivitiesManager.getAppActivitiesMaxQueueLength()); }