RouterRMAdminService extends AbstractService implements ResourceManagerAdministrationProtocol { @VisibleForTesting protected RequestInterceptorChainWrapper getInterceptorChain() throws IOException { String user = UserGroupInformation.getCurrentUser().getUserName(); RequestInterceptorChainWrapper chain = userPipelineMap.get(user); if (chain != null && chain.getRootInterceptor() != null) { return chain; } return initializePipeline(user); } RouterRMAdminService(); @VisibleForTesting Server getServer(); @Override String[] getGroupsForUser(String user); @Override RefreshQueuesResponse refreshQueues(RefreshQueuesRequest request); @Override RefreshNodesResponse refreshNodes(RefreshNodesRequest request); @Override RefreshSuperUserGroupsConfigurationResponse refreshSuperUserGroupsConfiguration(
      RefreshSuperUserGroupsConfigurationRequest request); @Override RefreshUserToGroupsMappingsResponse refreshUserToGroupsMappings(
      RefreshUserToGroupsMappingsRequest request); @Override RefreshAdminAclsResponse refreshAdminAcls(
      RefreshAdminAclsRequest request); @Override RefreshServiceAclsResponse refreshServiceAcls(
      RefreshServiceAclsRequest request); @Override UpdateNodeResourceResponse updateNodeResource(
      UpdateNodeResourceRequest request); @Override RefreshNodesResourcesResponse refreshNodesResources(
      RefreshNodesResourcesRequest request); @Override AddToClusterNodeLabelsResponse addToClusterNodeLabels(
      AddToClusterNodeLabelsRequest request); @Override RemoveFromClusterNodeLabelsResponse removeFromClusterNodeLabels(
      RemoveFromClusterNodeLabelsRequest request); @Override ReplaceLabelsOnNodeResponse replaceLabelsOnNode(
      ReplaceLabelsOnNodeRequest request); @Override CheckForDecommissioningNodesResponse checkForDecommissioningNodes(
      CheckForDecommissioningNodesRequest checkForDecommissioningNodesRequest); @Override RefreshClusterMaxPriorityResponse refreshClusterMaxPriority(
      RefreshClusterMaxPriorityRequest request); @Override NodesToAttributesMappingResponse mapAttributesToNodes(
      NodesToAttributesMappingRequest request);  }
@Test public void testRMAdminPipelineConcurrent() throws InterruptedException { final String user = "test1"; class ClientTestThread extends Thread { private RMAdminRequestInterceptor interceptor; @Override public void run() { try { interceptor = pipeline(); } catch (IOException | InterruptedException e) { e.printStackTrace(); } } private RMAdminRequestInterceptor pipeline() throws IOException, InterruptedException { return UserGroupInformation.createRemoteUser(user).doAs( new PrivilegedExceptionAction<RMAdminRequestInterceptor>() { @Override public RMAdminRequestInterceptor run() throws Exception { RequestInterceptorChainWrapper wrapper = getRouterRMAdminService().getInterceptorChain(); RMAdminRequestInterceptor interceptor = wrapper.getRootInterceptor(); Assert.assertNotNull(interceptor); LOG.info("init rm admin interceptor success for user" + user); return interceptor; } }); } } ClientTestThread client1 = new ClientTestThread(); ClientTestThread client2 = new ClientTestThread(); client1.start(); client2.start(); client1.join(); client2.join(); Assert.assertNotNull(client1.interceptor); Assert.assertNotNull(client2.interceptor); Assert.assertTrue(client1.interceptor == client2.interceptor); }