RouterClientRMService extends AbstractService implements ApplicationClientProtocol { @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); } RouterClientRMService(); @VisibleForTesting Server getServer(); @Override GetNewApplicationResponse getNewApplication(
      GetNewApplicationRequest request); @Override SubmitApplicationResponse submitApplication(
      SubmitApplicationRequest request); @Override KillApplicationResponse forceKillApplication(
      KillApplicationRequest request); @Override GetClusterMetricsResponse getClusterMetrics(
      GetClusterMetricsRequest request); @Override GetClusterNodesResponse getClusterNodes(GetClusterNodesRequest request); @Override GetQueueInfoResponse getQueueInfo(GetQueueInfoRequest request); @Override GetQueueUserAclsInfoResponse getQueueUserAcls(
      GetQueueUserAclsInfoRequest request); @Override MoveApplicationAcrossQueuesResponse moveApplicationAcrossQueues(
      MoveApplicationAcrossQueuesRequest request); @Override GetNewReservationResponse getNewReservation(
      GetNewReservationRequest request); @Override ReservationSubmissionResponse submitReservation(
      ReservationSubmissionRequest request); @Override ReservationListResponse listReservations(
      ReservationListRequest request); @Override ReservationUpdateResponse updateReservation(
      ReservationUpdateRequest request); @Override ReservationDeleteResponse deleteReservation(
      ReservationDeleteRequest request); @Override GetNodesToLabelsResponse getNodeToLabels(
      GetNodesToLabelsRequest request); @Override GetLabelsToNodesResponse getLabelsToNodes(
      GetLabelsToNodesRequest request); @Override GetClusterNodeLabelsResponse getClusterNodeLabels(
      GetClusterNodeLabelsRequest request); @Override GetApplicationReportResponse getApplicationReport(
      GetApplicationReportRequest request); @Override GetApplicationsResponse getApplications(GetApplicationsRequest request); @Override GetApplicationAttemptReportResponse getApplicationAttemptReport(
      GetApplicationAttemptReportRequest request); @Override GetApplicationAttemptsResponse getApplicationAttempts(
      GetApplicationAttemptsRequest request); @Override GetContainerReportResponse getContainerReport(
      GetContainerReportRequest request); @Override GetContainersResponse getContainers(GetContainersRequest request); @Override GetDelegationTokenResponse getDelegationToken(
      GetDelegationTokenRequest request); @Override RenewDelegationTokenResponse renewDelegationToken(
      RenewDelegationTokenRequest request); @Override CancelDelegationTokenResponse cancelDelegationToken(
      CancelDelegationTokenRequest request); @Override FailApplicationAttemptResponse failApplicationAttempt(
      FailApplicationAttemptRequest request); @Override UpdateApplicationPriorityResponse updateApplicationPriority(
      UpdateApplicationPriorityRequest request); @Override SignalContainerResponse signalToContainer(
      SignalContainerRequest request); @Override UpdateApplicationTimeoutsResponse updateApplicationTimeouts(
      UpdateApplicationTimeoutsRequest request); @Override GetAllResourceProfilesResponse getResourceProfiles(
      GetAllResourceProfilesRequest request); @Override GetResourceProfileResponse getResourceProfile(
      GetResourceProfileRequest request); @Override GetAllResourceTypeInfoResponse getResourceTypeInfo(
      GetAllResourceTypeInfoRequest request); @Override GetAttributesToNodesResponse getAttributesToNodes(
      GetAttributesToNodesRequest request); @Override GetClusterNodeAttributesResponse getClusterNodeAttributes(
      GetClusterNodeAttributesRequest request); @Override GetNodesToAttributesResponse getNodesToAttributes(
      GetNodesToAttributesRequest request);  }
@Test public void testClientPipelineConcurrent() throws InterruptedException { final String user = "test1"; class ClientTestThread extends Thread { private ClientRequestInterceptor interceptor; @Override public void run() { try { interceptor = pipeline(); } catch (IOException | InterruptedException e) { e.printStackTrace(); } } private ClientRequestInterceptor pipeline() throws IOException, InterruptedException { return UserGroupInformation.createRemoteUser(user).doAs( new PrivilegedExceptionAction<ClientRequestInterceptor>() { @Override public ClientRequestInterceptor run() throws Exception { RequestInterceptorChainWrapper wrapper = getRouterClientRMService().getInterceptorChain(); ClientRequestInterceptor interceptor = wrapper.getRootInterceptor(); Assert.assertNotNull(interceptor); LOG.info("init client 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); }