YarnClusterResourceManager extends ClusterResourceManager implements AMRMClientAsync.CallbackHandler, NMClientAsync.CallbackHandler { @Override public boolean isResourceExpired(SamzaResource resource) { Duration yarnAllocatedResourceExpiry = Duration.ofMillis(YarnConfiguration.DEFAULT_RM_CONTAINER_ALLOC_EXPIRY_INTERVAL_MS) .minus(Duration.ofSeconds(30)); return System.currentTimeMillis() - resource.getTimestamp() > yarnAllocatedResourceExpiry.toMillis(); } YarnClusterResourceManager(AMRMClientAsync amClientAsync, NMClientAsync nmClientAsync, Callback callback,
      YarnAppState yarnAppState, SamzaYarnAppMasterLifecycle lifecycle, SamzaYarnAppMasterService service,
      SamzaAppMasterMetrics metrics, YarnConfiguration yarnConfiguration, Config config);  YarnClusterResourceManager(Config config, JobModelManager jobModelManager,
      ClusterResourceManager.Callback callback, SamzaApplicationState samzaAppState); @Override void start(); @Override void requestResources(SamzaResourceRequest resourceRequest); @Override void releaseResources(SamzaResource resource); @Override void launchStreamProcessor(SamzaResource resource, CommandBuilder builder); void stopStreamProcessor(SamzaResource resource); @Override void cancelResourceRequest(SamzaResourceRequest request); @Override void stop(SamzaApplicationState.SamzaAppStatus status); @Override void onContainersCompleted(List<ContainerStatus> statuses); @Override void onContainersAllocated(List<Container> containers); @Override void onShutdownRequest(); @Override void onNodesUpdated(List<NodeReport> updatedNodes); @Override float getProgress(); @Override void onError(Throwable e); @Override void onContainerStarted(ContainerId containerId, Map<String, ByteBuffer> allServiceResponse); @Override void onContainerStatusReceived(ContainerId containerId, ContainerStatus containerStatus); @Override void onContainerStopped(ContainerId containerId); @Override void onStartContainerError(ContainerId containerId, Throwable t); @Override void onGetContainerStatusError(ContainerId containerId, Throwable t); @Override void onStopContainerError(ContainerId containerId, Throwable t); @Override boolean isResourceExpired(SamzaResource resource); void runProcessor(String processorId, Container container, CommandBuilder cmdBuilder);  }
@Test public void testAllocatedResourceExpiryForYarn() { YarnConfiguration yarnConfiguration = mock(YarnConfiguration.class); SamzaAppMasterMetrics metrics = mock(SamzaAppMasterMetrics.class); Config config = mock(Config.class); AMRMClientAsync asyncClient = mock(AMRMClientAsync.class); YarnAppState yarnAppState = new YarnAppState(0, mock(ContainerId.class), "host", 8080, 8081); SamzaYarnAppMasterLifecycle lifecycle = mock(SamzaYarnAppMasterLifecycle.class); SamzaYarnAppMasterService service = mock(SamzaYarnAppMasterService.class); NMClientAsync asyncNMClient = mock(NMClientAsync.class); ClusterResourceManager.Callback callback = mock(ClusterResourceManager.Callback.class); YarnClusterResourceManager yarnClusterResourceManager = new YarnClusterResourceManager(asyncClient, asyncNMClient, callback, yarnAppState, lifecycle, service, metrics, yarnConfiguration, config); SamzaResource allocatedResource = mock(SamzaResource.class); when(allocatedResource.getTimestamp()).thenReturn(System.currentTimeMillis() - Duration.ofMinutes(10).toMillis()); Assert.assertTrue(yarnClusterResourceManager.isResourceExpired(allocatedResource)); }