FederationInterceptor extends AbstractRequestInterceptor { @Override public void shutdown() { if (this.threadpool != null) { try { this.threadpool.shutdown(); } catch (Throwable ex) { } this.threadpool = null; } this.homeHeartbeartHandler.shutdown(); this.homeRMRelayer.shutdown(); for (AMRMClientRelayer relayer : this.secondaryRelayers.values()) { relayer.shutdown(); } super.shutdown(); } FederationInterceptor(); @Override void init(AMRMProxyApplicationContext appContext); @Override void recover(Map<String, byte[]> recoveredDataMap); @Override synchronized RegisterApplicationMasterResponse registerApplicationMaster(RegisterApplicationMasterRequest request); @Override AllocateResponse allocate(AllocateRequest request); @Override FinishApplicationMasterResponse finishApplicationMaster(
      FinishApplicationMasterRequest request); @Override void setNextInterceptor(RequestInterceptor next); @Override void shutdown(); @VisibleForTesting int getUnmanagedAMPoolSize(); @VisibleForTesting Map<SubClusterId, List<AllocateResponse>> getAsyncResponseSink(); static boolean isNullOrEmpty(Collection<T> c); static boolean isNullOrEmpty(Map<T1, T2> c); static final String NMSS_CLASS_PREFIX; static final String NMSS_REG_REQUEST_KEY; static final String NMSS_REG_RESPONSE_KEY; static final String NMSS_SECONDARY_SC_PREFIX; static final String STRING_TO_BYTE_FORMAT; }
@Test(timeout = 5000) public void testConcurrentRegister() throws InterruptedException, ExecutionException { ExecutorService threadpool = Executors.newCachedThreadPool(); ExecutorCompletionService<RegisterApplicationMasterResponse> compSvc = new ExecutorCompletionService<>(threadpool); Object syncObj = MockResourceManagerFacade.getSyncObj(); synchronized (syncObj) { LOG.info("Starting first register thread"); compSvc.submit(new ConcurrentRegisterAMCallable()); try { LOG.info("Test main starts waiting for the first thread to block"); syncObj.wait(); LOG.info("Test main wait finished"); } catch (Exception e) { LOG.info("Test main wait interrupted", e); } } LOG.info("Starting second register thread"); compSvc.submit(new ConcurrentRegisterAMCallable()); LOG.info("Let first blocked register thread move on"); synchronized (syncObj) { syncObj.notifyAll(); } RegisterApplicationMasterResponse response = compSvc.take().get(); Assert.assertNotNull(response); response = compSvc.take().get(); Assert.assertNotNull(response); threadpool.shutdown(); }