Hadoop RPC 代码分析

if (proxyInfo != null) {
    // ...
} else if (rpcNamenode != null) {
    // This case is used for testing.
} else {
    Preconditions.checkArgument(nameNodeUri != null,
                                "null URI");
    proxyInfo = NameNodeProxiesClient.createProxyWithClientProtocol(
        conf,
        nameNodeUri, nnFallbackToSimpleAuth);
    this.dtService = proxyInfo.getDelegationTokenService();
    this.namenode = proxyInfo.getProxy();
}

public static ProxyAndInfo<ClientProtocol> createProxyWithClientProtocol(
    Configuration conf, URI nameNodeUri, AtomicBoolean fallbackToSimpleAuth)
    throws IOException {
    AbstractNNFailoverProxyProvider<ClientProtocol> failoverProxyProvider =
        createFailoverProxyProvider(conf, nameNodeUri, ClientProtocol.class,
                                    true, fallbackToSimpleAuth);

    if (failoverProxyProvider == null) {
        InetSocketAddress nnAddr = DFSUtilClient.getNNAddress(nameNodeUri);
        Text dtService = SecurityUtil.buildTokenService(nnAddr);
        ClientProtocol proxy = createNonHAProxyWithClientProtocol(
            nnAddr, conf,
            UserGroupInformation.getCurrentUser(), true, fallbackToSimpleAuth);
        return new ProxyAndInfo<>(proxy, dtService, nnAddr);
    } else {
        return createHAProxy(conf, nameNodeUri, ClientProtocol.class,
                             failoverProxyProvider);
    }
}

public static ClientProtocol createNonHAProxyWithClientProtocol(
    InetSocketAddress address, Configuration conf, UserGroupInformation ugi,
    boolean withRetries, AtomicBoolean fallbackToSimpleAuth)
    throws IOException {
    return createProxyWithAlignmentContext(address, conf, ugi, withRetries,
                                           fallbackToSimpleAuth, null);
}

public static ClientProtocol createProxyWithAlignmentContext(
    InetSocketAddress address, Configuration conf, UserGroupInformation ugi,
    boolean withRetries, AtomicBoolean fallbackToSimpleAuth,
    AlignmentContext alignmentContext)
    throws IOException {
    RPC.setProtocolEngine(conf, ClientNamenodeProtocolPB.class,
                          ProtobufRpcEngine2.class);

    final RetryPolicy defaultPolicy =
        RetryUtils.getDefaultRetryPolicy(
            conf,
            HdfsClientConfigKeys.Retry.POLICY_ENABLED_KEY,
            HdfsClientConfigKeys.Retry.POLICY_ENABLED_DEFAULT,
            HdfsClientConfigKeys.Retry.POLICY_SPEC_KEY,
            HdfsClientConfigKeys.Retry.POLICY_SPEC_DEFAULT,
            SafeModeException.class.getName());

    final long version = RPC.getProtocolVersion(ClientNamenodeProtocolPB.class);
    ClientNamenodeProtocolPB proxy = RPC.getProtocolProxy(
        ClientNamenodeProtocolPB.class, version, address, ugi, conf,
        NetUtils.getDefaultSocketFactory(conf),
        org.apache.hadoop.ipc.Client.getTimeout(conf), defaultPolicy,
        fallbackToSimpleAuth, alignmentContext).getProxy();

    if (withRetries) { // create the proxy with retries
        Map<String, RetryPolicy> methodNameToPolicyMap = new HashMap<>();
        ClientProtocol translatorProxy =
            new ClientNamenodeProtocolTranslatorPB(proxy);
        return (ClientProtocol) RetryProxy.create(
            ClientProtocol.class,
            new DefaultFailoverProxyProvider<>(ClientProtocol.class,
                                               translatorProxy),
            methodNameToPolicyMap,
            defaultPolicy);
    } else {
        return new ClientNamenodeProtocolTranslatorPB(proxy);
    }
}

/**
 * This class forwards NN's ClientProtocol calls as RPC calls to the NN server
 * while translating from the parameter types used in ClientProtocol to the
 * new PB types.
 */

public final FSDataOutputStream create(final Path f,
    final EnumSet<CreateFlag> createFlag, Options.CreateOpts... opts)
    throws AccessControlException, FileAlreadyExistsException,
    FileNotFoundException, ParentNotDirectoryException,
    UnsupportedFileSystemException, UnresolvedLinkException, IOException {
  checkPath(f);
  int bufferSize = -1;
  short replication = -1;
  long blockSize = -1;
  int bytesPerChecksum = -1;
  ChecksumOpt checksumOpt = null;
  FsPermission permission = null;
  Progressable progress = null;
  Boolean createParent = null;

  for (CreateOpts iOpt : opts) {
    if (CreateOpts.BlockSize.class.isInstance(iOpt)) {
      if (blockSize != -1) {
        throw new HadoopIllegalArgumentException(
            "BlockSize option is set multiple times");
      }
      blockSize = ((CreateOpts.BlockSize) iOpt).getValue();
    } else if (CreateOpts.BufferSize.class.isInstance(iOpt)) {
      if (bufferSize != -1) {
        throw new HadoopIllegalArgumentException(
            "BufferSize option is set multiple times");
      }
      bufferSize = ((CreateOpts.BufferSize) iOpt).getValue();
    } else if (...) {
      // ...
    } else {
      throw new HadoopIllegalArgumentException("Unkown CreateOpts of type " +
          iOpt.getClass().getName());
    }
  }

  // ...

  return this.createInternal(f, createFlag, permission, bufferSize,
    replication, blockSize, progress, checksumOpt, createParent);
}

@Override
public HdfsDataOutputStream createInternal(Path f,
    EnumSet<CreateFlag> createFlag, FsPermission absolutePermission,
    int bufferSize, short replication, long blockSize, Progressable progress,
    ChecksumOpt checksumOpt, boolean createParent) throws IOException {

  final DFSOutputStream dfsos = dfs.primitiveCreate(getUriPath(f),
    absolutePermission, createFlag, createParent, replication, blockSize,
    progress, bufferSize, checksumOpt);
  return dfs.createWrappedOutputStream(dfsos, statistics,
      dfsos.getInitialLen());
}

public DFSOutputStream primitiveCreate(String src, FsPermission absPermission,
    EnumSet<CreateFlag> flag, boolean createParent, short replication,
    long blockSize, Progressable progress, int buffersize,
    ChecksumOpt checksumOpt) throws IOException {
  checkOpen();
  CreateFlag.validate(flag);
  DFSOutputStream result = primitiveAppend(src, flag, progress);
  if (result == null) {
    DataChecksum checksum = dfsClientConf.createChecksum(checksumOpt);
    result = DFSOutputStream.newStreamForCreate(this, src, absPermission,
        flag, createParent, replication, blockSize, progress, checksum,
        null, null, null);
  }
  beginFileLease(result.getFileId(), result);
  return result;
}

static DFSOutputStream newStreamForCreate(DFSClient dfsClient, String src,
    FsPermission masked, EnumSet<CreateFlag> flag, boolean createParent,
    short replication, long blockSize, Progressable progress,
    DataChecksum checksum, String[] favoredNodes, String ecPolicyName,
    String storagePolicy)
    throws IOException {
  try (TraceScope ignored =
           dfsClient.newPathTraceScope("newStreamForCreate", src)) {
    HdfsFileStatus stat = null;

    // Retry the create if we get a RetryStartFileException up to a maximum
    // number of times
    boolean shouldRetry = true;
    int retryCount = CREATE_RETRY_COUNT;
    while (shouldRetry) {
      shouldRetry = false;
      try {
        stat = dfsClient.namenode.create(src, masked, dfsClient.clientName,
            new EnumSetWritable<>(flag), createParent, replication,
            blockSize, SUPPORTED_CRYPTO_VERSIONS, ecPolicyName,
            storagePolicy);
        break;
      } catch (RemoteException re) {
        IOException e = re.unwrapRemoteException(
            AccessControlException.class,
            DSQuotaExceededException.class,
            QuotaByStorageTypeExceededException.class,
            FileAlreadyExistsException.class,
            FileNotFoundException.class,
            ParentNotDirectoryException.class,
            NSQuotaExceededException.class,
            RetryStartFileException.class,
            SafeModeException.class,
            UnresolvedPathException.class,
            SnapshotAccessControlException.class,
            UnknownCryptoProtocolVersionException.class);
        if (e instanceof RetryStartFileException) {
          if (retryCount > 0) {
            shouldRetry = true;
            retryCount--;
          } else {
            throw new IOException("Too many retries because of encryption" +
                " zone operations", e);
          }
        } else {
          throw e;
        }
      }
    }
    Preconditions.checkNotNull(stat, "HdfsFileStatus should not be null!");
    final DFSOutputStream out;
    if(stat.getErasureCodingPolicy() != null) {
      out = new DFSStripedOutputStream(dfsClient, src, stat,
          flag, progress, checksum, favoredNodes);
    } else {
      out = new DFSOutputStream(dfsClient, src, stat,
          flag, progress, checksum, favoredNodes, true);
    }
    out.start();
    return out;
  }
}

// ClientNamenodeProtocolTranslatorPB.java
@Override
public HdfsFileStatus create(String src, FsPermission masked,
    String clientName, EnumSetWritable<CreateFlag> flag,
    boolean createParent, short replication, long blockSize,
    CryptoProtocolVersion[] supportedVersions, String ecPolicyName,
    String storagePolicy)
    throws IOException {
    CreateRequestProto.Builder builder = CreateRequestProto.newBuilder()
        .setSrc(src)
        .setMasked(PBHelperClient.convert(masked))
        .setClientName(clientName)
        .setCreateFlag(PBHelperClient.convertCreateFlag(flag))
        .setCreateParent(createParent)
        .setReplication(replication)
        .setBlockSize(blockSize);
    if (ecPolicyName != null) {
        builder.setEcPolicyName(ecPolicyName);
    }
    if (storagePolicy != null) {
        builder.setStoragePolicy(storagePolicy);
    }
    FsPermission unmasked = masked.getUnmasked();
    if (unmasked != null) {
        builder.setUnmasked(PBHelperClient.convert(unmasked));
    }
    builder.addAllCryptoProtocolVersion(
        PBHelperClient.convert(supportedVersions));
    CreateRequestProto req = builder.build();
    try {
        // NameNodeProxiesClient.java init rpcProxy from RPC.getProtocolProxy.
        CreateResponseProto res = rpcProxy.create(null, req);
        return res.hasFs() ? PBHelperClient.convert(res.getFs()) : null;
    } catch (ServiceException e) {
        throw ProtobufHelper.getRemoteException(e);
    }
}

// bind to invoke
@Override
public Message invoke(Object proxy, final Method method, Object[] args)
    throws ServiceException {
    long startTime = 0;
    if (LOG.isDebugEnabled()) {
      startTime = Time.now();
    }

    if (args.length != 2) { // RpcController + Message
      throw new ServiceException(
          "Too many or few parameters for request. Method: ["
          + method.getName() + "]" + ", Expected: 2, Actual: "
          + args.length);
    }
    if (args[1] == null) {
      throw new ServiceException("null param while calling Method: ["
          + method.getName() + "]");
    }

    // if Tracing is on then start a new span for this rpc.
    // guard it in the if statement to make sure there isn't
    // any extra string manipulation.
    Tracer tracer = Tracer.curThreadTracer();
    TraceScope traceScope = null;
    if (tracer != null) {
      traceScope = tracer.newScope(RpcClientUtil.methodToTraceString(method));
    }

    if (LOG.isTraceEnabled()) {
      LOG.trace(Thread.currentThread().getId() + ": Call -> " +
          remoteId + ": " + method.getName() +
          " {" + TextFormat.shortDebugString((Message) args[1]) + "}");
    }


    final Message theRequest = (Message) args[1];
    final RpcWritable.Buffer val;
    try {
      val = (RpcWritable.Buffer) client.call(RPC.RpcKind.RPC_PROTOCOL_BUFFER,
          constructRpcRequest(method, theRequest), remoteId,
          fallbackToSimpleAuth, alignmentContext);

    } catch (Throwable e) {
      if (LOG.isTraceEnabled()) {
        LOG.trace(Thread.currentThread().getId() + ": Exception <- " +
            remoteId + ": " + method.getName() +
              " {" + e + "}");
      }
      if (traceScope != null) {
        traceScope.addTimelineAnnotation("Call got exception: " +
            e.toString());
      }
      throw new ServiceException(e);
    } finally {
      if (traceScope != null) {
        traceScope.close();
      }
    }

    if (LOG.isDebugEnabled()) {
      long callTime = Time.now() - startTime;
      LOG.debug("Call: " + method.getName() + " took " + callTime + "ms");
    }

    if (Client.isAsynchronousMode()) {
      final AsyncGet<RpcWritable.Buffer, IOException> arr
          = Client.getAsyncRpcResponse();
      final AsyncGet<Message, Exception> asyncGet =
          new AsyncGet<Message, Exception>() {
            @Override
            public Message get(long timeout, TimeUnit unit) throws Exception {
              return getReturnMessage(method, arr.get(timeout, unit));
            }

            @Override
            public boolean isDone() {
          return arr.isDone();
        }
      };
      ASYNC_RETURN_MESSAGE.set(asyncGet);
      return null;
    } else {
      return getReturnMessage(method, val);
    }
  }

@Override
@SuppressWarnings("unchecked")
public <T> ProtocolProxy<T> getProxy(
    Class<T> protocol, long clientVersion,
    ConnectionId connId, Configuration conf, SocketFactory factory)
    throws IOException {
    final Invoker invoker = new Invoker(protocol, connId, conf, factory);
    return new ProtocolProxy<T>(
        protocol, (T) Proxy.newProxyInstance(
                          protocol.getClassLoader(),
                          new Class[] {protocol}, invoker), false);
}

  // NameNodeRpcServer.java
  public NameNodeRpcServer(Configuration conf, NameNode nn)
      throws IOException {
    ClientNamenodeProtocolServerSideTranslatorPB
       clientProtocolServerTranslator =
         new ClientNamenodeProtocolServerSideTranslatorPB(this);
     BlockingService clientNNPbService = ClientNamenodeProtocol.
         newReflectiveBlockingService(clientProtocolServerTranslator);

    InetSocketAddress serviceRpcAddr = nn.getServiceRpcServerAddress(conf);
    if (serviceRpcAddr != null) {
      int serviceHandlerCount =
        conf.getInt(DFS_NAMENODE_SERVICE_HANDLER_COUNT_KEY,
                    DFS_NAMENODE_SERVICE_HANDLER_COUNT_DEFAULT);
      serviceRpcServer = new RPC.Builder(conf)
          .setProtocol(
              org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolPB.class)
          .setInstance(clientNNPbService)
          .setBindAddress(bindHost)
          .setPort(serviceRpcAddr.getPort())
          .setNumHandlers(serviceHandlerCount)
          .setVerbose(false)
          .setSecretManager(namesystem.getDelegationTokenSecretManager())
          .build();

      // Update the address with the correct port
      InetSocketAddress listenAddr = serviceRpcServer.getListenerAddress();
      serviceRPCAddress = new InetSocketAddress(
            serviceRpcAddr.getHostName(), listenAddr.getPort());
      nn.setRpcServiceServerAddress(conf, serviceRPCAddress);
    }
  }

  // ClientNamenodeProtocolServerSideTranslatorPB.java
  @Override
  public CreateResponseProto create(RpcController controller,
      CreateRequestProto req) throws ServiceException {
    try {
      FsPermission masked = req.hasUnmasked() ?
          FsCreateModes.create(PBHelperClient.convert(req.getMasked()),
              PBHelperClient.convert(req.getUnmasked())) :
          PBHelperClient.convert(req.getMasked());
      HdfsFileStatus result = server.create(req.getSrc(),
          masked, req.getClientName(),
          PBHelperClient.convertCreateFlag(req.getCreateFlag()), req.getCreateParent(),
          (short) req.getReplication(), req.getBlockSize(),
          PBHelperClient.convertCryptoProtocolVersions(
              req.getCryptoProtocolVersionList()),
          req.getEcPolicyName(), req.getStoragePolicy());

      if (result != null) {
        return CreateResponseProto.newBuilder().setFs(PBHelperClient.convert(result))
            .build();
      }
      return VOID_CREATE_RESPONSE;
    } catch (IOException e) {
      throw new ServiceException(e);
    }
  }

// hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java
// Optional.
@Override // ClientProtocol
  public HdfsFileStatus create(String src, FsPermission masked,
      String clientName, EnumSetWritable<CreateFlag> flag,
      boolean createParent, short replication, long blockSize,
      CryptoProtocolVersion[] supportedVersions, String ecPolicyName,
      String storagePolicy)
      throws IOException {
    return clientProto.create(src, masked, clientName, flag, createParent,
        replication, blockSize, supportedVersions, ecPolicyName, storagePolicy);
  }

  // hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
  @Override // ClientProtocol
  public HdfsFileStatus create(String src, FsPermission masked,
      String clientName, EnumSetWritable<CreateFlag> flag,
      boolean createParent, short replication, long blockSize,
      CryptoProtocolVersion[] supportedVersions, String ecPolicyName,
      String storagePolicy)
      throws IOException {
    checkNNStartup();
    String clientMachine = getClientMachine();
    if (stateChangeLog.isDebugEnabled()) {
      stateChangeLog.debug("*DIR* NameNode.create: file "
          +src+" for "+clientName+" at "+clientMachine);
    }
    if (!checkPathLength(src)) {
      throw new IOException("create: Pathname too long.  Limit "
          + MAX_PATH_LENGTH + " characters, " + MAX_PATH_DEPTH + " levels.");
    }
    namesystem.checkOperation(OperationCategory.WRITE);
    CacheEntryWithPayload cacheEntry = RetryCache.waitForCompletion(retryCache, null);
    if (cacheEntry != null && cacheEntry.isSuccess()) {
      return (HdfsFileStatus) cacheEntry.getPayload();
    }

    HdfsFileStatus status = null;
    try {
      PermissionStatus perm = new PermissionStatus(getRemoteUser()
          .getShortUserName(), null, masked);
      status = namesystem.startFile(src, perm, clientName, clientMachine,
          flag.get(), createParent, replication, blockSize, supportedVersions,
          ecPolicyName, storagePolicy, cacheEntry != null);
    } finally {
      RetryCache.setState(cacheEntry, status != null, status);
    }

    metrics.incrFilesCreated();
    metrics.incrCreateFileOps();
    return status;
  }

// hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
/**
   * Create a new file entry in the namespace.
   *
   * For description of parameters and exceptions thrown see
   * {@link ClientProtocol#create}, except it returns valid file status upon
   * success
   */
  HdfsFileStatus startFile(String src, PermissionStatus permissions,
      String holder, String clientMachine, EnumSet<CreateFlag> flag,
      boolean createParent, short replication, long blockSize,
      CryptoProtocolVersion[] supportedVersions, String ecPolicyName,
      String storagePolicy, boolean logRetryCache) throws IOException {

    HdfsFileStatus status;
    try {
      status = startFileInt(src, permissions, holder, clientMachine, flag,
          createParent, replication, blockSize, supportedVersions, ecPolicyName,
          storagePolicy, logRetryCache);
    } catch (AccessControlException e) {
      logAuditEvent(false, "create", src);
      throw e;
    }
    logAuditEvent(true, "create", src, status);
    return status;
  }