org.apache.hadoop.fs.ParentNotDirectoryException#com.google.protobuf.BlockingService源码实例Demo

/**
 * Starts an RPC server, if configured.
 *
 * @param conf configuration
 * @param addr configured address of RPC server
 * @param protocol RPC protocol provided by RPC server
 * @param instance RPC protocol implementation instance
 * @param handlerCount RPC server handler count
 * @return RPC server
 * @throws IOException if there is an I/O error while creating RPC server
 */
private static RPC.Server startRpcServer(Configuration conf,
    InetSocketAddress addr, Class<?>
    protocol, BlockingService instance, int handlerCount)
    throws IOException {
  RPC.Server rpcServer = new RPC.Builder(conf)
      .setProtocol(protocol)
      .setInstance(instance)
      .setBindAddress(addr.getHostString())
      .setPort(addr.getPort())
      .setNumHandlers(handlerCount)
      .setVerbose(false)
      .setSecretManager(null)
      .build();

  HddsServerUtil.addPBProtocol(conf, protocol, instance, rpcServer);
  return rpcServer;
}
 

/**
 * Start Datanode RPC server.
 */
public static RPC.Server startScmRpcServer(ConfigurationSource configuration,
    StorageContainerDatanodeProtocol server,
    InetSocketAddress rpcServerAddresss, int handlerCount) throws
    IOException {

  Configuration hadoopConfig =
      LegacyHadoopConfigurationSource.asHadoopConfiguration(configuration);
  RPC.setProtocolEngine(hadoopConfig,
      StorageContainerDatanodeProtocolPB.class,
      ProtobufRpcEngine.class);

  BlockingService scmDatanodeService =
      StorageContainerDatanodeProtocolService.
          newReflectiveBlockingService(
              new StorageContainerDatanodeProtocolServerSideTranslatorPB(
                  server, Mockito.mock(ProtocolMessageMetrics.class)));

  RPC.Server scmServer = startRpcServer(hadoopConfig, rpcServerAddresss,
      StorageContainerDatanodeProtocolPB.class, scmDatanodeService,
      handlerCount);

  scmServer.start();
  return scmServer;
}
 

/**
 * Starts an RPC server, if configured.
 *
 * @param conf configuration
 * @param addr configured address of RPC server
 * @param protocol RPC protocol provided by RPC server
 * @param instance RPC protocol implementation instance
 * @param handlerCount RPC server handler count
 * @return RPC server
 * @throws IOException if there is an I/O error while creating RPC server
 */
public static RPC.Server startRpcServer(
    OzoneConfiguration conf,
    InetSocketAddress addr,
    Class<?> protocol,
    BlockingService instance,
    int handlerCount)
    throws IOException {
  RPC.Server rpcServer =
      new RPC.Builder(conf)
          .setProtocol(protocol)
          .setInstance(instance)
          .setBindAddress(addr.getHostString())
          .setPort(addr.getPort())
          .setNumHandlers(handlerCount)
          .setVerbose(false)
          .setSecretManager(null)
          .build();

  HddsServerUtil.addPBProtocol(conf, protocol, instance, rpcServer);
  return rpcServer;
}
 

/**
 * Starts an RPC server, if configured.
 *
 * @param conf         configuration
 * @param addr         configured address of RPC server
 * @param protocol     RPC protocol provided by RPC server
 * @param instance     RPC protocol implementation instance
 * @param handlerCount RPC server handler count
 * @return RPC server
 * @throws IOException if there is an I/O error while creating RPC server
 */
private RPC.Server startRpcServer(OzoneConfiguration conf,
    InetSocketAddress addr, Class<?> protocol, BlockingService instance,
    int handlerCount) throws IOException {
  RPC.Server rpcServer = new RPC.Builder(conf)
      .setProtocol(protocol)
      .setInstance(instance)
      .setBindAddress(addr.getHostString())
      .setPort(addr.getPort())
      .setNumHandlers(handlerCount)
      .setVerbose(false)
      .setSecretManager(delegationTokenMgr)
      .build();

  HddsServerUtil.addPBProtocol(conf, protocol, instance, rpcServer);

  if (conf.getBoolean(CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION,
      false)) {
    rpcServer.refreshServiceAcl(conf, OMPolicyProvider.getInstance());
  }
  return rpcServer;
}
 

/**
 * Creates a new instance of rpc server. If an earlier instance is already
 * running then returns the same.
 */
private RPC.Server getRpcServer(OzoneConfiguration conf) throws IOException {
  if (isOmRpcServerRunning) {
    return omRpcServer;
  }

  InetSocketAddress omNodeRpcAddr = OmUtils.getOmAddress(conf);

  final int handlerCount = conf.getInt(OZONE_OM_HANDLER_COUNT_KEY,
      OZONE_OM_HANDLER_COUNT_DEFAULT);
  RPC.setProtocolEngine(configuration, OzoneManagerProtocolPB.class,
      ProtobufRpcEngine.class);
  this.omServerProtocol = new OzoneManagerProtocolServerSideTranslatorPB(
      this, omRatisServer, omClientProtocolMetrics, isRatisEnabled);

  BlockingService omService = newReflectiveBlockingService(omServerProtocol);

  return startRpcServer(configuration, omNodeRpcAddr,
      OzoneManagerProtocolPB.class, omService,
      handlerCount);
}
 

private static Server createMockDatanode(BlockTokenSecretManager sm,
    Token<BlockTokenIdentifier> token, Configuration conf)
    throws IOException, ServiceException {
  ClientDatanodeProtocolPB mockDN = mock(ClientDatanodeProtocolPB.class);

  BlockTokenIdentifier id = sm.createIdentifier();
  id.readFields(new DataInputStream(new ByteArrayInputStream(token
      .getIdentifier())));
  
  doAnswer(new GetLengthAnswer(sm, id)).when(mockDN)
      .getReplicaVisibleLength(any(RpcController.class),
          any(GetReplicaVisibleLengthRequestProto.class));

  RPC.setProtocolEngine(conf, ClientDatanodeProtocolPB.class,
      ProtobufRpcEngine.class);
  BlockingService service = ClientDatanodeProtocolService
      .newReflectiveBlockingService(mockDN);
  return new RPC.Builder(conf).setProtocol(ClientDatanodeProtocolPB.class)
      .setInstance(service).setBindAddress(ADDRESS).setPort(0)
      .setNumHandlers(5).setVerbose(true).setSecretManager(sm).build();
}
 

ZKFCRpcServer(Configuration conf,
    InetSocketAddress bindAddr,
    ZKFailoverController zkfc,
    PolicyProvider policy) throws IOException {
  this.zkfc = zkfc;
  
  RPC.setProtocolEngine(conf, ZKFCProtocolPB.class,
      ProtobufRpcEngine.class);
  ZKFCProtocolServerSideTranslatorPB translator =
      new ZKFCProtocolServerSideTranslatorPB(this);
  BlockingService service = ZKFCProtocolService
      .newReflectiveBlockingService(translator);
  this.server = new RPC.Builder(conf).setProtocol(ZKFCProtocolPB.class)
      .setInstance(service).setBindAddress(bindAddr.getHostName())
      .setPort(bindAddr.getPort()).setNumHandlers(HANDLER_COUNT)
      .setVerbose(false).build();
  
  // set service-level authorization security policy
  if (conf.getBoolean(
      CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION, false)) {
    server.refreshServiceAcl(conf, policy);
  }

}
 

private InetSocketAddress startAndGetRPCServerAddress(InetSocketAddress serverAddress) {
  Configuration conf = new Configuration();

  try {
    RPC.setProtocolEngine(conf,
        HAServiceProtocolPB.class, ProtobufRpcEngine.class);
    HAServiceProtocolServerSideTranslatorPB haServiceProtocolXlator =
        new HAServiceProtocolServerSideTranslatorPB(new MockHAProtocolImpl());
    BlockingService haPbService = HAServiceProtocolService
        .newReflectiveBlockingService(haServiceProtocolXlator);

    Server server = new RPC.Builder(conf)
        .setProtocol(HAServiceProtocolPB.class)
        .setInstance(haPbService)
        .setBindAddress(serverAddress.getHostName())
        .setPort(serverAddress.getPort()).build();
    server.start();
    return NetUtils.getConnectAddress(server);
  } catch (IOException e) {
    return null;
  }
}
 

@Before
public  void setUp() throws IOException { // Setup server for both protocols
  conf = new Configuration();
  conf.setInt(CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH, 1024);
  // Set RPC engine to protobuf RPC engine
  RPC.setProtocolEngine(conf, TestRpcService.class, ProtobufRpcEngine.class);

  // Create server side implementation
  PBServerImpl serverImpl = new PBServerImpl();
  BlockingService service = TestProtobufRpcProto
      .newReflectiveBlockingService(serverImpl);

  // Get RPC server for server side implementation
  server = new RPC.Builder(conf).setProtocol(TestRpcService.class)
      .setInstance(service).setBindAddress(ADDRESS).setPort(PORT).build();
  addr = NetUtils.getConnectAddress(server);
  
  // now the second protocol
  PBServer2Impl server2Impl = new PBServer2Impl();
  BlockingService service2 = TestProtobufRpc2Proto
      .newReflectiveBlockingService(server2Impl);
  
  server.addProtocol(RPC.RpcKind.RPC_PROTOCOL_BUFFER, TestRpcService2.class,
      service2);
  server.start();
}
 

@Before
public void setUp() throws Exception {
  // create a server with two handlers
  server = new RPC.Builder(conf).setProtocol(Foo0.class)
      .setInstance(new Foo0Impl()).setBindAddress(ADDRESS).setPort(0)
      .setNumHandlers(2).setVerbose(false).build();
  server.addProtocol(RPC.RpcKind.RPC_WRITABLE, Foo1.class, new Foo1Impl());
  server.addProtocol(RPC.RpcKind.RPC_WRITABLE, Bar.class, new BarImpl());
  server.addProtocol(RPC.RpcKind.RPC_WRITABLE, Mixin.class, new BarImpl());
  
  
  // Add Protobuf server
  // Create server side implementation
  PBServerImpl pbServerImpl = 
      new PBServerImpl();
  BlockingService service = TestProtobufRpcProto
      .newReflectiveBlockingService(pbServerImpl);
  server.addProtocol(RPC.RpcKind.RPC_PROTOCOL_BUFFER, TestRpcService.class,
      service);
  server.start();
  addr = NetUtils.getConnectAddress(server);
}
 

private static Server createMockDatanode(BlockTokenSecretManager sm,
    Token<BlockTokenIdentifier> token, Configuration conf)
    throws IOException, ServiceException {
  ClientDatanodeProtocolPB mockDN = mock(ClientDatanodeProtocolPB.class);

  BlockTokenIdentifier id = sm.createIdentifier();
  id.readFields(new DataInputStream(new ByteArrayInputStream(token
      .getIdentifier())));
  
  doAnswer(new GetLengthAnswer(sm, id)).when(mockDN)
      .getReplicaVisibleLength(any(RpcController.class),
          any(GetReplicaVisibleLengthRequestProto.class));

  RPC.setProtocolEngine(conf, ClientDatanodeProtocolPB.class,
      ProtobufRpcEngine.class);
  BlockingService service = ClientDatanodeProtocolService
      .newReflectiveBlockingService(mockDN);
  return new RPC.Builder(conf).setProtocol(ClientDatanodeProtocolPB.class)
      .setInstance(service).setBindAddress(ADDRESS).setPort(0)
      .setNumHandlers(5).setVerbose(true).setSecretManager(sm).build();
}
 

ZKFCRpcServer(Configuration conf,
    InetSocketAddress bindAddr,
    ZKFailoverController zkfc,
    PolicyProvider policy) throws IOException {
  this.zkfc = zkfc;
  
  RPC.setProtocolEngine(conf, ZKFCProtocolPB.class,
      ProtobufRpcEngine.class);
  ZKFCProtocolServerSideTranslatorPB translator =
      new ZKFCProtocolServerSideTranslatorPB(this);
  BlockingService service = ZKFCProtocolService
      .newReflectiveBlockingService(translator);
  this.server = new RPC.Builder(conf).setProtocol(ZKFCProtocolPB.class)
      .setInstance(service).setBindAddress(bindAddr.getHostName())
      .setPort(bindAddr.getPort()).setNumHandlers(HANDLER_COUNT)
      .setVerbose(false).build();
  
  // set service-level authorization security policy
  if (conf.getBoolean(
      CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION, false)) {
    server.refreshServiceAcl(conf, policy);
  }

}
 

private InetSocketAddress startAndGetRPCServerAddress(InetSocketAddress serverAddress) {
  Configuration conf = new Configuration();

  try {
    RPC.setProtocolEngine(conf,
        HAServiceProtocolPB.class, ProtobufRpcEngine.class);
    HAServiceProtocolServerSideTranslatorPB haServiceProtocolXlator =
        new HAServiceProtocolServerSideTranslatorPB(new MockHAProtocolImpl());
    BlockingService haPbService = HAServiceProtocolService
        .newReflectiveBlockingService(haServiceProtocolXlator);

    Server server = new RPC.Builder(conf)
        .setProtocol(HAServiceProtocolPB.class)
        .setInstance(haPbService)
        .setBindAddress(serverAddress.getHostName())
        .setPort(serverAddress.getPort()).build();
    server.start();
    return NetUtils.getConnectAddress(server);
  } catch (IOException e) {
    return null;
  }
}
 

@Before
public  void setUp() throws IOException { // Setup server for both protocols
  conf = new Configuration();
  conf.setInt(CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH, 1024);
  // Set RPC engine to protobuf RPC engine
  RPC.setProtocolEngine(conf, TestRpcService.class, ProtobufRpcEngine.class);

  // Create server side implementation
  PBServerImpl serverImpl = new PBServerImpl();
  BlockingService service = TestProtobufRpcProto
      .newReflectiveBlockingService(serverImpl);

  // Get RPC server for server side implementation
  server = new RPC.Builder(conf).setProtocol(TestRpcService.class)
      .setInstance(service).setBindAddress(ADDRESS).setPort(PORT).build();
  addr = NetUtils.getConnectAddress(server);
  
  // now the second protocol
  PBServer2Impl server2Impl = new PBServer2Impl();
  BlockingService service2 = TestProtobufRpc2Proto
      .newReflectiveBlockingService(server2Impl);
  
  server.addProtocol(RPC.RpcKind.RPC_PROTOCOL_BUFFER, TestRpcService2.class,
      service2);
  server.start();
}
 

@Before
public void setUp() throws Exception {
  // create a server with two handlers
  server = new RPC.Builder(conf).setProtocol(Foo0.class)
      .setInstance(new Foo0Impl()).setBindAddress(ADDRESS).setPort(0)
      .setNumHandlers(2).setVerbose(false).build();
  server.addProtocol(RPC.RpcKind.RPC_WRITABLE, Foo1.class, new Foo1Impl());
  server.addProtocol(RPC.RpcKind.RPC_WRITABLE, Bar.class, new BarImpl());
  server.addProtocol(RPC.RpcKind.RPC_WRITABLE, Mixin.class, new BarImpl());
  
  
  // Add Protobuf server
  // Create server side implementation
  PBServerImpl pbServerImpl = 
      new PBServerImpl();
  BlockingService service = TestProtobufRpcProto
      .newReflectiveBlockingService(pbServerImpl);
  server.addProtocol(RPC.RpcKind.RPC_PROTOCOL_BUFFER, TestRpcService.class,
      service);
  server.start();
  addr = NetUtils.getConnectAddress(server);
}
 

/**
 * Handle the blocking RPC request by forwarding it to the correct
 * service/method.
 *
 * @throws RpcException If there was some error executing the RPC.
 */
public SocketRpcProtos.Response doBlockingRpc(
    SocketRpcProtos.Request rpcRequest) throws RpcException {
  // Get the service, first try BlockingService
  BlockingService blockingService = blockingServiceMap.get(
      rpcRequest.getServiceName());
  if (blockingService != null) {
    return forwardToBlockingService(rpcRequest, blockingService);
  }

  // Now try Service
  Service service = serviceMap.get(rpcRequest.getServiceName());
  if (service == null) {
    throw new RpcException(ErrorReason.SERVICE_NOT_FOUND,
        "Could not find service: " + rpcRequest.getServiceName(), null);
  }

  // Call service using an instant callback
  Callback<Message> callback = new Callback<Message>();
  SocketRpcController socketController = new SocketRpcController();
  forwardToService(rpcRequest, callback, service, socketController);

  // Build and return response (callback invocation is optional)
  return createRpcResponse(callback.response, callback.invoked,
      socketController);
}
 

public BlockingRpcServer(final Class<?> protocol,
                         final Object instance,
                         final InetSocketAddress bindAddress,
                         final int threads)
    throws Exception {

  super(protocol.getSimpleName(), bindAddress);

  String serviceClassName = protocol.getName() + "$" +
      protocol.getSimpleName() + "Service";
  Class<?> serviceClass = Class.forName(serviceClassName);
  Class<?> interfaceClass = Class.forName(serviceClassName +
      "$BlockingInterface");
  Method method = serviceClass.getMethod(
      "newReflectiveBlockingService", interfaceClass);

  this.service = (BlockingService) method.invoke(null, instance);
  this.initializer = new ProtoServerChannelInitializer(new ServerHandler(), RpcRequest.getDefaultInstance());

  super.init(this.initializer, threads);
}
 

public BlockingRpcServer(final Class<?> protocol,
                         final Object instance,
                         final InetSocketAddress bindAddress,
                         final int workerNum)
    throws Exception {

  super(protocol.getSimpleName(), bindAddress);

  String serviceClassName = protocol.getName() + "$" +
      protocol.getSimpleName() + "Service";
  Class<?> serviceClass = Class.forName(serviceClassName);
  Class<?> interfaceClass = Class.forName(serviceClassName +
      "$BlockingInterface");
  Method method = serviceClass.getMethod(
      "newReflectiveBlockingService", interfaceClass);

  this.service = (BlockingService) method.invoke(null, instance);
  this.pipeline = new ProtoPipelineFactory(new ServerHandler(),
      RpcRequest.getDefaultInstance());

  super.init(this.pipeline, workerNum);
}
 

SCMSecurityProtocolServer(OzoneConfiguration conf,
    CertificateServer certificateServer) throws IOException {
  this.certificateServer = certificateServer;

  final int handlerCount =
      conf.getInt(ScmConfigKeys.OZONE_SCM_SECURITY_HANDLER_COUNT_KEY,
          ScmConfigKeys.OZONE_SCM_SECURITY_HANDLER_COUNT_DEFAULT);
  rpcAddress = HddsServerUtil
      .getScmSecurityInetAddress(conf);
  // SCM security service RPC service.
  RPC.setProtocolEngine(conf, SCMSecurityProtocolPB.class,
      ProtobufRpcEngine.class);
  metrics = new ProtocolMessageMetrics("ScmSecurityProtocol",
      "SCM Security protocol metrics",
      SCMSecurityProtocolProtos.Type.values());
  BlockingService secureProtoPbService =
      SCMSecurityProtocolProtos.SCMSecurityProtocolService
          .newReflectiveBlockingService(
              new SCMSecurityProtocolServerSideTranslatorPB(this, metrics));
  this.rpcServer =
      StorageContainerManager.startRpcServer(
          conf,
          rpcAddress,
          SCMSecurityProtocolPB.class,
          secureProtoPbService,
          handlerCount);
  if (conf.getBoolean(
      CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION, false)) {
    rpcServer.refreshServiceAcl(conf, SCMPolicyProvider.getInstance());
  }
}
 

public SCMClientProtocolServer(OzoneConfiguration conf,
    StorageContainerManager scm) throws IOException {
  this.scm = scm;
  this.conf = conf;
  safeModePrecheck = new SafeModePrecheck(conf);
  final int handlerCount =
      conf.getInt(OZONE_SCM_HANDLER_COUNT_KEY,
          OZONE_SCM_HANDLER_COUNT_DEFAULT);
  RPC.setProtocolEngine(conf, StorageContainerLocationProtocolPB.class,
      ProtobufRpcEngine.class);

  protocolMetrics = ProtocolMessageMetrics
      .create("ScmContainerLocationProtocol",
          "SCM ContainerLocation protocol metrics",
          StorageContainerLocationProtocolProtos.Type.values());

  // SCM Container Service RPC
  BlockingService storageProtoPbService =
      newReflectiveBlockingService(
          new StorageContainerLocationProtocolServerSideTranslatorPB(this,
              protocolMetrics));

  final InetSocketAddress scmAddress = HddsServerUtil
      .getScmClientBindAddress(conf);
  clientRpcServer =
      startRpcServer(
          conf,
          scmAddress,
          StorageContainerLocationProtocolPB.class,
          storageProtoPbService,
          handlerCount);
  clientRpcAddress =
      updateRPCListenAddress(conf, OZONE_SCM_CLIENT_ADDRESS_KEY,
          scmAddress, clientRpcServer);
  if (conf.getBoolean(CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION,
      false)) {
    clientRpcServer.refreshServiceAcl(conf, SCMPolicyProvider.getInstance());
  }
}
 

private Server createServer(Class<?> pbProtocol, InetSocketAddress addr, Configuration conf, 
    SecretManager<? extends TokenIdentifier> secretManager, int numHandlers, 
    BlockingService blockingService, String portRangeConfig) throws IOException {
  RPC.setProtocolEngine(conf, pbProtocol, ProtobufRpcEngine.class);
  RPC.Server server = new RPC.Builder(conf).setProtocol(pbProtocol)
      .setInstance(blockingService).setBindAddress(addr.getHostName())
      .setPort(addr.getPort()).setNumHandlers(numHandlers).setVerbose(false)
      .setSecretManager(secretManager).setPortRangeConfig(portRangeConfig)
      .build();
  LOG.info("Adding protocol "+pbProtocol.getCanonicalName()+" to the server");
  server.addProtocol(RPC.RpcKind.RPC_PROTOCOL_BUFFER, pbProtocol, blockingService);
  return server;
}
 

protected void startServer() throws Exception {
  Configuration conf = getConfig();
  YarnRPC rpc = YarnRPC.create(conf);
  this.server = (Server) rpc.getServer(
      ResourceManagerAdministrationProtocol.class, this, masterServiceBindAddress,
      conf, null,
      conf.getInt(YarnConfiguration.RM_ADMIN_CLIENT_THREAD_COUNT,
          YarnConfiguration.DEFAULT_RM_ADMIN_CLIENT_THREAD_COUNT));

  // Enable service authorization?
  if (conf.getBoolean(
      CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION,
      false)) {
    refreshServiceAcls(
        getConfiguration(conf,
            YarnConfiguration.HADOOP_POLICY_CONFIGURATION_FILE),
        RMPolicyProvider.getInstance());
  }

  if (rmContext.isHAEnabled()) {
    RPC.setProtocolEngine(conf, HAServiceProtocolPB.class,
        ProtobufRpcEngine.class);

    HAServiceProtocolServerSideTranslatorPB haServiceProtocolXlator =
        new HAServiceProtocolServerSideTranslatorPB(this);
    BlockingService haPbService =
        HAServiceProtocolProtos.HAServiceProtocolService
            .newReflectiveBlockingService(haServiceProtocolXlator);
    server.addProtocol(RPC.RpcKind.RPC_PROTOCOL_BUFFER,
        HAServiceProtocol.class, haPbService);
  }

  this.server.start();
  conf.updateConnectAddr(YarnConfiguration.RM_BIND_HOST,
                         YarnConfiguration.RM_ADMIN_ADDRESS,
                         YarnConfiguration.DEFAULT_RM_ADMIN_ADDRESS,
                         server.getListenerAddress());
}
 

private BackupNodeRpcServer(Configuration conf, BackupNode nn)
    throws IOException {
  super(conf, nn);
  JournalProtocolServerSideTranslatorPB journalProtocolTranslator = 
      new JournalProtocolServerSideTranslatorPB(this);
  BlockingService service = JournalProtocolService
      .newReflectiveBlockingService(journalProtocolTranslator);
  DFSUtil.addPBProtocol(conf, JournalProtocolPB.class, service,
      this.clientRpcServer);
}
 

JournalNodeRpcServer(Configuration conf, JournalNode jn) throws IOException {
  this.jn = jn;
  
  Configuration confCopy = new Configuration(conf);
  
  // Ensure that nagling doesn't kick in, which could cause latency issues.
  confCopy.setBoolean(
      CommonConfigurationKeysPublic.IPC_SERVER_TCPNODELAY_KEY,
      true);
  
  InetSocketAddress addr = getAddress(confCopy);
  RPC.setProtocolEngine(confCopy, QJournalProtocolPB.class,
      ProtobufRpcEngine.class);
  QJournalProtocolServerSideTranslatorPB translator =
      new QJournalProtocolServerSideTranslatorPB(this);
  BlockingService service = QJournalProtocolService
      .newReflectiveBlockingService(translator);
  
  this.server = new RPC.Builder(confCopy)
    .setProtocol(QJournalProtocolPB.class)
    .setInstance(service)
    .setBindAddress(addr.getHostName())
    .setPort(addr.getPort())
    .setNumHandlers(HANDLER_COUNT)
    .setVerbose(false)
    .build();

  // set service-level authorization security policy
  if (confCopy.getBoolean(
    CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION, false)) {
        server.refreshServiceAcl(confCopy, new HDFSPolicyProvider());
  }
}
 

private void initProtocolMetaInfo(Configuration conf) {
  RPC.setProtocolEngine(conf, ProtocolMetaInfoPB.class,
      ProtobufRpcEngine.class);
  ProtocolMetaInfoServerSideTranslatorPB xlator = 
      new ProtocolMetaInfoServerSideTranslatorPB(this);
  BlockingService protocolInfoBlockingService = ProtocolInfoService
      .newReflectiveBlockingService(xlator);
  addProtocol(RpcKind.RPC_PROTOCOL_BUFFER, ProtocolMetaInfoPB.class,
      protocolInfoBlockingService);
}
 

private Server startServer(MyOptions opts) throws IOException {
  if (opts.serverThreads <= 0) {
    return null;
  }
  conf.setInt(CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_KEY,
      opts.serverReaderThreads);
  
  RPC.Server server;
  // Get RPC server for server side implementation
  if (opts.rpcEngine == ProtobufRpcEngine.class) {
    // Create server side implementation
    PBServerImpl serverImpl = new PBServerImpl();
    BlockingService service = TestProtobufRpcProto
        .newReflectiveBlockingService(serverImpl);

    server = new RPC.Builder(conf).setProtocol(TestRpcService.class)
        .setInstance(service).setBindAddress(opts.host).setPort(opts.getPort())
        .setNumHandlers(opts.serverThreads).setVerbose(false).build();
  } else if (opts.rpcEngine == WritableRpcEngine.class) {
    server = new RPC.Builder(conf).setProtocol(TestProtocol.class)
        .setInstance(new TestRPC.TestImpl()).setBindAddress(opts.host)
        .setPort(opts.getPort()).setNumHandlers(opts.serverThreads)
        .setVerbose(false).build();
  } else {
    throw new RuntimeException("Bad engine: " + opts.rpcEngine);
  }
  server.start();
  return server;
}
 

public void start() throws IOException {
    int rpcHandlerCount = conf.getInt(ProxyConfig.PROXY_HANDLER_COUNT, ProxyConfig.PROXY_HANDLER_COUNT_DEFAULT);
    RPC.setProtocolEngine(conf, ClientNamenodeProtocolPB.class,
            ProtobufRpcEngine.class);
    RPC.setProtocolEngine(conf, NamenodeProtocolPB.class,
            ProtobufRpcEngine.class);

    this.protocol = (ClientProtocol) Proxy.newProxyInstance(
            this.getClass().getClassLoader(),
            new Class[]{ClientProtocol.class},
            this.invocationHandler);

    ClientNamenodeProtocolPB proxy = new ClientNamenodeProtocolServerSideTranslatorPB(this.protocol);
    BlockingService clientNNPbService = ClientNamenodeProtocolProtos.ClientNamenodeProtocol.
            newReflectiveBlockingService(proxy);

    int port = conf.getInt(ProxyConfig.RPC_PORT, ProxyConfig.RPC_PORT_DEFAULT);

    this.rpcServer = new RPC.Builder(conf)
            .setProtocol(org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolPB.class)
            .setInstance(clientNNPbService).setBindAddress("0.0.0.0")
            .setPort(port).setNumHandlers(rpcHandlerCount)
            .setVerbose(false).build();
    this.rpcServer.start();

    InetSocketAddress listenAddr = rpcServer.getListenerAddress();
    rpcAddress = new InetSocketAddress("0.0.0.0", listenAddr.getPort());
}
 

private Server createServer(Class<?> pbProtocol, InetSocketAddress addr, Configuration conf, 
    SecretManager<? extends TokenIdentifier> secretManager, int numHandlers, 
    BlockingService blockingService, String portRangeConfig) throws IOException {
  RPC.setProtocolEngine(conf, pbProtocol, ProtobufRpcEngine.class);
  RPC.Server server = new RPC.Builder(conf).setProtocol(pbProtocol)
      .setInstance(blockingService).setBindAddress(addr.getHostName())
      .setPort(addr.getPort()).setNumHandlers(numHandlers).setVerbose(false)
      .setSecretManager(secretManager).setPortRangeConfig(portRangeConfig)
      .build();
  LOG.info("Adding protocol "+pbProtocol.getCanonicalName()+" to the server");
  server.addProtocol(RPC.RpcKind.RPC_PROTOCOL_BUFFER, pbProtocol, blockingService);
  return server;
}
 

protected void startServer() throws Exception {
  Configuration conf = getConfig();
  YarnRPC rpc = YarnRPC.create(conf);
  this.server = (Server) rpc.getServer(
      ResourceManagerAdministrationProtocol.class, this, masterServiceBindAddress,
      conf, null,
      conf.getInt(YarnConfiguration.RM_ADMIN_CLIENT_THREAD_COUNT,
          YarnConfiguration.DEFAULT_RM_ADMIN_CLIENT_THREAD_COUNT));

  // Enable service authorization?
  if (conf.getBoolean(
      CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION,
      false)) {
    refreshServiceAcls(
        getConfiguration(conf,
            YarnConfiguration.HADOOP_POLICY_CONFIGURATION_FILE),
        RMPolicyProvider.getInstance());
  }

  if (rmContext.isHAEnabled()) {
    RPC.setProtocolEngine(conf, HAServiceProtocolPB.class,
        ProtobufRpcEngine.class);

    HAServiceProtocolServerSideTranslatorPB haServiceProtocolXlator =
        new HAServiceProtocolServerSideTranslatorPB(this);
    BlockingService haPbService =
        HAServiceProtocolProtos.HAServiceProtocolService
            .newReflectiveBlockingService(haServiceProtocolXlator);
    server.addProtocol(RPC.RpcKind.RPC_PROTOCOL_BUFFER,
        HAServiceProtocol.class, haPbService);
  }

  this.server.start();
  conf.updateConnectAddr(YarnConfiguration.RM_BIND_HOST,
                         YarnConfiguration.RM_ADMIN_ADDRESS,
                         YarnConfiguration.DEFAULT_RM_ADMIN_ADDRESS,
                         server.getListenerAddress());
}
 

private BackupNodeRpcServer(Configuration conf, BackupNode nn)
    throws IOException {
  super(conf, nn);
  JournalProtocolServerSideTranslatorPB journalProtocolTranslator = 
      new JournalProtocolServerSideTranslatorPB(this);
  BlockingService service = JournalProtocolService
      .newReflectiveBlockingService(journalProtocolTranslator);
  DFSUtil.addPBProtocol(conf, JournalProtocolPB.class, service,
      this.clientRpcServer);
}