1. 包列表
  2. 类列表
  3. org.apache.hadoop.io.retry.RetryUtils

Java 类org.apache.hadoop.io.retry.RetryUtils 实例源码

项目：hadoop-plus    文件：WebHdfsFileSystem.java    
@Override
public synchronized void initialize(URI uri, Configuration conf
    ) throws IOException {
  super.initialize(uri, conf);
  setConf(conf);
  ugi = UserGroupInformation.getCurrentUser();
  try {
    this.uri = new URI(uri.getScheme(), uri.getAuthority(), null, null, null);
  } catch (URISyntaxException e) {
    throw new IllegalArgumentException(e);
  }
  this.nnAddr = NetUtils.createSocketAddr(uri.getAuthority(), getDefaultPort());
  this.retryPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class);
  this.workingDir = getHomeDirectory();

  if (UserGroupInformation.isSecurityEnabled()) {
    initDelegationToken();
  }
}
项目：hops    文件：WebHdfsFileSystem.java    
@Override
public synchronized void initialize(URI uri, Configuration conf)
    throws IOException {
  super.initialize(uri, conf);
  setConf(conf);
  ugi = UserGroupInformation.getCurrentUser();
  try {
    this.uri = new URI(uri.getScheme(), uri.getAuthority(), null, null, null);
  } catch (URISyntaxException e) {
    throw new IllegalArgumentException(e);
  }
  this.nnAddr =
      NetUtils.createSocketAddr(uri.getAuthority(), getDefaultPort());
  this.retryPolicy = RetryUtils.getDefaultRetryPolicy(conf,
      DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY,
      DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
      DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
      DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
      SafeModeException.class);
  this.workingDir = getHomeDirectory();

  if (UserGroupInformation.isSecurityEnabled()) {
    initDelegationToken();
  }
}
项目：hadoop-TCP    文件：WebHdfsFileSystem.java    
@Override
public synchronized void initialize(URI uri, Configuration conf
    ) throws IOException {
  super.initialize(uri, conf);
  setConf(conf);
  ugi = UserGroupInformation.getCurrentUser();
  try {
    this.uri = new URI(uri.getScheme(), uri.getAuthority(), null, null, null);
  } catch (URISyntaxException e) {
    throw new IllegalArgumentException(e);
  }
  this.nnAddr = NetUtils.createSocketAddr(uri.getAuthority(), getDefaultPort());
  this.retryPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf, 
          DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_KEY,
          DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
          DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_KEY,
          DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class);
  this.workingDir = getHomeDirectory();

  if (UserGroupInformation.isSecurityEnabled()) {
    initDelegationToken();
  }
}
项目：hadoop-on-lustre    文件：JobClient.java    
private static JobSubmissionProtocol createRPCProxy(InetSocketAddress addr,
    Configuration conf) throws IOException {

  JobSubmissionProtocol rpcJobSubmitClient = 
      (JobSubmissionProtocol)RPC.getProxy(
          JobSubmissionProtocol.class,
          JobSubmissionProtocol.versionID, addr, 
          UserGroupInformation.getCurrentUser(), conf,
          NetUtils.getSocketFactory(conf, JobSubmissionProtocol.class), 
          0,
          RetryUtils.getMultipleLinearRandomRetry(
              conf,
              MAPREDUCE_CLIENT_RETRY_POLICY_ENABLED_KEY,
              MAPREDUCE_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
              MAPREDUCE_CLIENT_RETRY_POLICY_SPEC_KEY,
              MAPREDUCE_CLIENT_RETRY_POLICY_SPEC_DEFAULT
              )
          );

  return rpcJobSubmitClient;
}
项目：hadoop-on-lustre    文件：WebHdfsFileSystem.java    
@Override
public synchronized void initialize(URI uri, Configuration conf
    ) throws IOException {
  super.initialize(uri, conf);
  setConf(conf);

  this.nnAddr = NetUtils.createSocketAddr(uri.getAuthority(), getDefaultPort());
  this.retryPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class
          );
  this.workingDir = getHomeDirectory();

  if (UserGroupInformation.isSecurityEnabled()) {
    initDelegationToken();
  }
}
项目：hardfs    文件：WebHdfsFileSystem.java    
@Override
public synchronized void initialize(URI uri, Configuration conf
    ) throws IOException {
  super.initialize(uri, conf);
  setConf(conf);
  ugi = UserGroupInformation.getCurrentUser();
  try {
    this.uri = new URI(uri.getScheme(), uri.getAuthority(), null, null, null);
  } catch (URISyntaxException e) {
    throw new IllegalArgumentException(e);
  }
  this.nnAddr = NetUtils.createSocketAddr(uri.getAuthority(), getDefaultPort());
  this.retryPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf, 
          DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_KEY,
          DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
          DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_KEY,
          DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class);
  this.workingDir = getHomeDirectory();

  if (UserGroupInformation.isSecurityEnabled()) {
    initDelegationToken();
  }
}
项目：hortonworks-extension    文件：JobClient.java    
private static JobSubmissionProtocol createRPCProxy(InetSocketAddress addr,
    Configuration conf) throws IOException {

  JobSubmissionProtocol rpcJobSubmitClient = 
      (JobSubmissionProtocol)RPC.getProxy(
          JobSubmissionProtocol.class,
          JobSubmissionProtocol.versionID, addr, 
          UserGroupInformation.getCurrentUser(), conf,
          NetUtils.getSocketFactory(conf, JobSubmissionProtocol.class), 
          0,
          RetryUtils.getMultipleLinearRandomRetry(
              conf,
              MAPREDUCE_CLIENT_RETRY_POLICY_ENABLED_KEY,
              MAPREDUCE_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
              MAPREDUCE_CLIENT_RETRY_POLICY_SPEC_KEY,
              MAPREDUCE_CLIENT_RETRY_POLICY_SPEC_DEFAULT
              )
          );

  return rpcJobSubmitClient;
}
项目：hortonworks-extension    文件：WebHdfsFileSystem.java    
@Override
public synchronized void initialize(URI uri, Configuration conf
    ) throws IOException {
  super.initialize(uri, conf);
  setConf(conf);

  this.nnAddr = NetUtils.createSocketAddr(uri.getAuthority(), getDefaultPort());
  this.retryPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class
          );
  this.workingDir = getHomeDirectory();

  if (UserGroupInformation.isSecurityEnabled()) {
    initDelegationToken();
  }
}
项目：hortonworks-extension    文件：JobClient.java    
private static JobSubmissionProtocol createRPCProxy(InetSocketAddress addr,
    Configuration conf) throws IOException {

  JobSubmissionProtocol rpcJobSubmitClient = 
      (JobSubmissionProtocol)RPC.getProxy(
          JobSubmissionProtocol.class,
          JobSubmissionProtocol.versionID, addr, 
          UserGroupInformation.getCurrentUser(), conf,
          NetUtils.getSocketFactory(conf, JobSubmissionProtocol.class), 
          0,
          RetryUtils.getMultipleLinearRandomRetry(
              conf,
              MAPREDUCE_CLIENT_RETRY_POLICY_ENABLED_KEY,
              MAPREDUCE_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
              MAPREDUCE_CLIENT_RETRY_POLICY_SPEC_KEY,
              MAPREDUCE_CLIENT_RETRY_POLICY_SPEC_DEFAULT
              )
          );

  return rpcJobSubmitClient;
}
项目：hortonworks-extension    文件：WebHdfsFileSystem.java    
@Override
public synchronized void initialize(URI uri, Configuration conf
    ) throws IOException {
  super.initialize(uri, conf);
  setConf(conf);

  this.nnAddr = NetUtils.createSocketAddr(uri.getAuthority(), getDefaultPort());
  this.retryPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class
          );
  this.workingDir = getHomeDirectory();

  if (UserGroupInformation.isSecurityEnabled()) {
    initDelegationToken();
  }
}
项目：hadoop    文件：NameNodeProxies.java    
private static ClientProtocol createNNProxyWithClientProtocol(
    InetSocketAddress address, Configuration conf, UserGroupInformation ugi,
    boolean withRetries, AtomicBoolean fallbackToSimpleAuth)
    throws IOException {
  RPC.setProtocolEngine(conf, ClientNamenodeProtocolPB.class, ProtobufRpcEngine.class);

  final RetryPolicy defaultPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class);

  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).getProxy();

  if (withRetries) { // create the proxy with retries

    Map<String, RetryPolicy> methodNameToPolicyMap 
               = new HashMap<String, RetryPolicy>();

    ClientProtocol translatorProxy =
      new ClientNamenodeProtocolTranslatorPB(proxy);
    return (ClientProtocol) RetryProxy.create(
        ClientProtocol.class,
        new DefaultFailoverProxyProvider<ClientProtocol>(
            ClientProtocol.class, translatorProxy),
        methodNameToPolicyMap,
        defaultPolicy);
  } else {
    return new ClientNamenodeProtocolTranslatorPB(proxy);
  }
}
项目：aliyun-oss-hadoop-fs    文件：NameNodeProxiesClient.java    
public static ClientProtocol createNonHAProxyWithClientProtocol(
    InetSocketAddress address, Configuration conf, UserGroupInformation ugi,
    boolean withRetries, AtomicBoolean fallbackToSimpleAuth)
    throws IOException {
  RPC.setProtocolEngine(conf, ClientNamenodeProtocolPB.class,
      ProtobufRpcEngine.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).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);
  }
}
项目：big-c    文件：NameNodeProxies.java    
private static ClientProtocol createNNProxyWithClientProtocol(
    InetSocketAddress address, Configuration conf, UserGroupInformation ugi,
    boolean withRetries, AtomicBoolean fallbackToSimpleAuth)
    throws IOException {
  RPC.setProtocolEngine(conf, ClientNamenodeProtocolPB.class, ProtobufRpcEngine.class);

  final RetryPolicy defaultPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class);

  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).getProxy();

  if (withRetries) { // create the proxy with retries

    Map<String, RetryPolicy> methodNameToPolicyMap 
               = new HashMap<String, RetryPolicy>();

    ClientProtocol translatorProxy =
      new ClientNamenodeProtocolTranslatorPB(proxy);
    return (ClientProtocol) RetryProxy.create(
        ClientProtocol.class,
        new DefaultFailoverProxyProvider<ClientProtocol>(
            ClientProtocol.class, translatorProxy),
        methodNameToPolicyMap,
        defaultPolicy);
  } else {
    return new ClientNamenodeProtocolTranslatorPB(proxy);
  }
}
项目：hadoop-on-lustre    文件：JobClient.java    
private static JobSubmissionProtocol createProxy(
    JobSubmissionProtocol rpcJobSubmitClient,
    Configuration conf) throws IOException {

  RetryPolicy defaultPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf,
          MAPREDUCE_CLIENT_RETRY_POLICY_ENABLED_KEY,
          MAPREDUCE_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
          MAPREDUCE_CLIENT_RETRY_POLICY_SPEC_KEY,
          MAPREDUCE_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class
          ); 

  /* 
   * Method specific retry policies for killJob and killTask...
   * 
   * No retries on any exception including 
   * ConnectionException and SafeModeException
   */
  Map<String,RetryPolicy> methodNameToPolicyMap = 
      new HashMap<String,RetryPolicy>();
  methodNameToPolicyMap.put("killJob", RetryPolicies.TRY_ONCE_THEN_FAIL);
  methodNameToPolicyMap.put("killTask", RetryPolicies.TRY_ONCE_THEN_FAIL);

  return (JobSubmissionProtocol) RetryProxy.create(JobSubmissionProtocol.class,
      rpcJobSubmitClient, defaultPolicy, methodNameToPolicyMap);
}
项目：hadoop-on-lustre    文件：DFSClient.java    
private static ClientProtocol createRPCNamenode(InetSocketAddress nameNodeAddr,
  Configuration conf, UserGroupInformation ugi) 
throws IOException {
return (ClientProtocol)RPC.getProxy(ClientProtocol.class,
    ClientProtocol.versionID, nameNodeAddr, ugi, conf,
    NetUtils.getSocketFactory(conf, ClientProtocol.class), 0,
    RetryUtils.getMultipleLinearRandomRetry(
            conf, 
            DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY,
            DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
            DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
            DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT
            ));  
}
项目：hadoop-on-lustre    文件：DFSClient.java    
private static ClientProtocol createNamenode(ClientProtocol rpcNamenode,
    Configuration conf) throws IOException {
  //default policy
  final RetryPolicy defaultPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class
          );

  //create policy
  RetryPolicy createPolicy = RetryPolicies.retryUpToMaximumCountWithFixedSleep(
      5, LEASE_SOFTLIMIT_PERIOD, TimeUnit.MILLISECONDS);

  Map<Class<? extends Exception>,RetryPolicy> remoteExceptionToPolicyMap =
    new HashMap<Class<? extends Exception>, RetryPolicy>();
  remoteExceptionToPolicyMap.put(AlreadyBeingCreatedException.class, createPolicy);

  Map<Class<? extends Exception>,RetryPolicy> exceptionToPolicyMap =
    new HashMap<Class<? extends Exception>, RetryPolicy>();
  exceptionToPolicyMap.put(RemoteException.class, 
      RetryPolicies.retryByRemoteException(
          defaultPolicy, remoteExceptionToPolicyMap));
  RetryPolicy methodPolicy = RetryPolicies.retryByException(
      defaultPolicy, exceptionToPolicyMap);
  Map<String,RetryPolicy> methodNameToPolicyMap = new HashMap<String,RetryPolicy>();

  methodNameToPolicyMap.put("create", methodPolicy);

  return (ClientProtocol) RetryProxy.create(ClientProtocol.class,
      rpcNamenode, defaultPolicy, methodNameToPolicyMap);
}
项目：hortonworks-extension    文件：JobClient.java    
private static JobSubmissionProtocol createProxy(
    JobSubmissionProtocol rpcJobSubmitClient,
    Configuration conf) throws IOException {

  RetryPolicy defaultPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf,
          MAPREDUCE_CLIENT_RETRY_POLICY_ENABLED_KEY,
          MAPREDUCE_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
          MAPREDUCE_CLIENT_RETRY_POLICY_SPEC_KEY,
          MAPREDUCE_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class
          ); 

  /* 
   * Method specific retry policies for killJob and killTask...
   * 
   * No retries on any exception including 
   * ConnectionException and SafeModeException
   */
  Map<String,RetryPolicy> methodNameToPolicyMap = 
      new HashMap<String,RetryPolicy>();
  methodNameToPolicyMap.put("killJob", RetryPolicies.TRY_ONCE_THEN_FAIL);
  methodNameToPolicyMap.put("killTask", RetryPolicies.TRY_ONCE_THEN_FAIL);

  return (JobSubmissionProtocol) RetryProxy.create(JobSubmissionProtocol.class,
      rpcJobSubmitClient, defaultPolicy, methodNameToPolicyMap);
}
项目：hortonworks-extension    文件：DFSClient.java    
private static ClientProtocol createRPCNamenode(InetSocketAddress nameNodeAddr,
  Configuration conf, UserGroupInformation ugi) 
throws IOException {
return (ClientProtocol)RPC.getProxy(ClientProtocol.class,
    ClientProtocol.versionID, nameNodeAddr, ugi, conf,
    NetUtils.getSocketFactory(conf, ClientProtocol.class), 0,
    RetryUtils.getMultipleLinearRandomRetry(
            conf, 
            DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY,
            DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
            DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
            DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT
            ));  
}
项目：hortonworks-extension    文件：DFSClient.java    
private static ClientProtocol createNamenode(ClientProtocol rpcNamenode,
    Configuration conf) throws IOException {
  //default policy
  final RetryPolicy defaultPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class
          );

  //create policy
  RetryPolicy createPolicy = RetryPolicies.retryUpToMaximumCountWithFixedSleep(
      5, LEASE_SOFTLIMIT_PERIOD, TimeUnit.MILLISECONDS);

  Map<Class<? extends Exception>,RetryPolicy> remoteExceptionToPolicyMap =
    new HashMap<Class<? extends Exception>, RetryPolicy>();
  remoteExceptionToPolicyMap.put(AlreadyBeingCreatedException.class, createPolicy);

  Map<Class<? extends Exception>,RetryPolicy> exceptionToPolicyMap =
    new HashMap<Class<? extends Exception>, RetryPolicy>();
  exceptionToPolicyMap.put(RemoteException.class, 
      RetryPolicies.retryByRemoteException(
          defaultPolicy, remoteExceptionToPolicyMap));
  RetryPolicy methodPolicy = RetryPolicies.retryByException(
      defaultPolicy, exceptionToPolicyMap);
  Map<String,RetryPolicy> methodNameToPolicyMap = new HashMap<String,RetryPolicy>();

  methodNameToPolicyMap.put("create", methodPolicy);

  return (ClientProtocol) RetryProxy.create(ClientProtocol.class,
      rpcNamenode, defaultPolicy, methodNameToPolicyMap);
}
项目：hortonworks-extension    文件：JobClient.java    
private static JobSubmissionProtocol createProxy(
    JobSubmissionProtocol rpcJobSubmitClient,
    Configuration conf) throws IOException {

  RetryPolicy defaultPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf,
          MAPREDUCE_CLIENT_RETRY_POLICY_ENABLED_KEY,
          MAPREDUCE_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
          MAPREDUCE_CLIENT_RETRY_POLICY_SPEC_KEY,
          MAPREDUCE_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class
          ); 

  /* 
   * Method specific retry policies for killJob and killTask...
   * 
   * No retries on any exception including 
   * ConnectionException and SafeModeException
   */
  Map<String,RetryPolicy> methodNameToPolicyMap = 
      new HashMap<String,RetryPolicy>();
  methodNameToPolicyMap.put("killJob", RetryPolicies.TRY_ONCE_THEN_FAIL);
  methodNameToPolicyMap.put("killTask", RetryPolicies.TRY_ONCE_THEN_FAIL);

  return (JobSubmissionProtocol) RetryProxy.create(JobSubmissionProtocol.class,
      rpcJobSubmitClient, defaultPolicy, methodNameToPolicyMap);
}
项目：hortonworks-extension    文件：DFSClient.java    
private static ClientProtocol createRPCNamenode(InetSocketAddress nameNodeAddr,
  Configuration conf, UserGroupInformation ugi) 
throws IOException {
return (ClientProtocol)RPC.getProxy(ClientProtocol.class,
    ClientProtocol.versionID, nameNodeAddr, ugi, conf,
    NetUtils.getSocketFactory(conf, ClientProtocol.class), 0,
    RetryUtils.getMultipleLinearRandomRetry(
            conf, 
            DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY,
            DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
            DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
            DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT
            ));  
}
项目：hortonworks-extension    文件：DFSClient.java    
private static ClientProtocol createNamenode(ClientProtocol rpcNamenode,
    Configuration conf) throws IOException {
  //default policy
  final RetryPolicy defaultPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class
          );

  //create policy
  RetryPolicy createPolicy = RetryPolicies.retryUpToMaximumCountWithFixedSleep(
      5, LEASE_SOFTLIMIT_PERIOD, TimeUnit.MILLISECONDS);

  Map<Class<? extends Exception>,RetryPolicy> remoteExceptionToPolicyMap =
    new HashMap<Class<? extends Exception>, RetryPolicy>();
  remoteExceptionToPolicyMap.put(AlreadyBeingCreatedException.class, createPolicy);

  Map<Class<? extends Exception>,RetryPolicy> exceptionToPolicyMap =
    new HashMap<Class<? extends Exception>, RetryPolicy>();
  exceptionToPolicyMap.put(RemoteException.class, 
      RetryPolicies.retryByRemoteException(
          defaultPolicy, remoteExceptionToPolicyMap));
  RetryPolicy methodPolicy = RetryPolicies.retryByException(
      defaultPolicy, exceptionToPolicyMap);
  Map<String,RetryPolicy> methodNameToPolicyMap = new HashMap<String,RetryPolicy>();

  methodNameToPolicyMap.put("create", methodPolicy);

  return (ClientProtocol) RetryProxy.create(ClientProtocol.class,
      rpcNamenode, defaultPolicy, methodNameToPolicyMap);
}
项目：hadoop    文件：WebHdfsFileSystem.java    
@Override
public synchronized void initialize(URI uri, Configuration conf
    ) throws IOException {
  super.initialize(uri, conf);
  setConf(conf);
  /** set user pattern based on configuration file */
  UserParam.setUserPattern(conf.get(
      DFSConfigKeys.DFS_WEBHDFS_USER_PATTERN_KEY,
      DFSConfigKeys.DFS_WEBHDFS_USER_PATTERN_DEFAULT));

  connectionFactory = URLConnectionFactory
      .newDefaultURLConnectionFactory(conf);

  ugi = UserGroupInformation.getCurrentUser();
  this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());
  this.nnAddrs = resolveNNAddr();

  boolean isHA = HAUtil.isClientFailoverConfigured(conf, this.uri);
  boolean isLogicalUri = isHA && HAUtil.isLogicalUri(conf, this.uri);
  // In non-HA or non-logical URI case, the code needs to call
  // getCanonicalUri() in order to handle the case where no port is
  // specified in the URI
  this.tokenServiceName = isLogicalUri ?
      HAUtil.buildTokenServiceForLogicalUri(uri, getScheme())
      : SecurityUtil.buildTokenService(getCanonicalUri());

  if (!isHA) {
    this.retryPolicy =
        RetryUtils.getDefaultRetryPolicy(
            conf,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_KEY,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_KEY,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
            SafeModeException.class);
  } else {

    int maxFailoverAttempts = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_MAX_ATTEMPTS_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_MAX_ATTEMPTS_DEFAULT);
    int maxRetryAttempts = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_MAX_ATTEMPTS_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_MAX_ATTEMPTS_DEFAULT);
    int failoverSleepBaseMillis = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_BASE_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_BASE_DEFAULT);
    int failoverSleepMaxMillis = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_MAX_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_MAX_DEFAULT);

    this.retryPolicy = RetryPolicies
        .failoverOnNetworkException(RetryPolicies.TRY_ONCE_THEN_FAIL,
            maxFailoverAttempts, maxRetryAttempts, failoverSleepBaseMillis,
            failoverSleepMaxMillis);
  }

  this.workingDir = getHomeDirectory();
  this.canRefreshDelegationToken = UserGroupInformation.isSecurityEnabled();
  this.disallowFallbackToInsecureCluster = !conf.getBoolean(
      CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_KEY,
      CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_DEFAULT);
  this.delegationToken = null;
}
项目：aliyun-oss-hadoop-fs    文件：WebHdfsFileSystem.java    
@Override
public synchronized void initialize(URI uri, Configuration conf
) throws IOException {
  super.initialize(uri, conf);
  setConf(conf);
  /** set user pattern based on configuration file */
  UserParam.setUserPattern(conf.get(
      HdfsClientConfigKeys.DFS_WEBHDFS_USER_PATTERN_KEY,
      HdfsClientConfigKeys.DFS_WEBHDFS_USER_PATTERN_DEFAULT));

  boolean isOAuth = conf.getBoolean(
      HdfsClientConfigKeys.DFS_WEBHDFS_OAUTH_ENABLED_KEY,
      HdfsClientConfigKeys.DFS_WEBHDFS_OAUTH_ENABLED_DEFAULT);

  if(isOAuth) {
    LOG.debug("Enabling OAuth2 in WebHDFS");
    connectionFactory = URLConnectionFactory
        .newOAuth2URLConnectionFactory(conf);
  } else {
    LOG.debug("Not enabling OAuth2 in WebHDFS");
    connectionFactory = URLConnectionFactory
        .newDefaultURLConnectionFactory(conf);
  }


  ugi = UserGroupInformation.getCurrentUser();
  this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());
  this.nnAddrs = resolveNNAddr();

  boolean isHA = HAUtilClient.isClientFailoverConfigured(conf, this.uri);
  boolean isLogicalUri = isHA && HAUtilClient.isLogicalUri(conf, this.uri);
  // In non-HA or non-logical URI case, the code needs to call
  // getCanonicalUri() in order to handle the case where no port is
  // specified in the URI
  this.tokenServiceName = isLogicalUri ?
      HAUtilClient.buildTokenServiceForLogicalUri(uri, getScheme())
      : SecurityUtil.buildTokenService(getCanonicalUri());

  if (!isHA) {
    this.retryPolicy =
        RetryUtils.getDefaultRetryPolicy(
            conf,
            HdfsClientConfigKeys.HttpClient.RETRY_POLICY_ENABLED_KEY,
            HdfsClientConfigKeys.HttpClient.RETRY_POLICY_ENABLED_DEFAULT,
            HdfsClientConfigKeys.HttpClient.RETRY_POLICY_SPEC_KEY,
            HdfsClientConfigKeys.HttpClient.RETRY_POLICY_SPEC_DEFAULT,
            HdfsConstants.SAFEMODE_EXCEPTION_CLASS_NAME);
  } else {

    int maxFailoverAttempts = conf.getInt(
        HdfsClientConfigKeys.HttpClient.FAILOVER_MAX_ATTEMPTS_KEY,
        HdfsClientConfigKeys.HttpClient.FAILOVER_MAX_ATTEMPTS_DEFAULT);
    int maxRetryAttempts = conf.getInt(
        HdfsClientConfigKeys.HttpClient.RETRY_MAX_ATTEMPTS_KEY,
        HdfsClientConfigKeys.HttpClient.RETRY_MAX_ATTEMPTS_DEFAULT);
    int failoverSleepBaseMillis = conf.getInt(
        HdfsClientConfigKeys.HttpClient.FAILOVER_SLEEPTIME_BASE_KEY,
        HdfsClientConfigKeys.HttpClient.FAILOVER_SLEEPTIME_BASE_DEFAULT);
    int failoverSleepMaxMillis = conf.getInt(
        HdfsClientConfigKeys.HttpClient.FAILOVER_SLEEPTIME_MAX_KEY,
        HdfsClientConfigKeys.HttpClient.FAILOVER_SLEEPTIME_MAX_DEFAULT);

    this.retryPolicy = RetryPolicies
        .failoverOnNetworkException(RetryPolicies.TRY_ONCE_THEN_FAIL,
            maxFailoverAttempts, maxRetryAttempts, failoverSleepBaseMillis,
            failoverSleepMaxMillis);
  }

  this.workingDir = makeQualified(new Path(getHomeDirectoryString(ugi)));
  this.canRefreshDelegationToken = UserGroupInformation.isSecurityEnabled();
  this.disallowFallbackToInsecureCluster = !conf.getBoolean(
      CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_KEY,
      CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_DEFAULT);
  this.delegationToken = null;
}
项目：big-c    文件：WebHdfsFileSystem.java    
@Override
public synchronized void initialize(URI uri, Configuration conf
    ) throws IOException {
  super.initialize(uri, conf);
  setConf(conf);
  /** set user pattern based on configuration file */
  UserParam.setUserPattern(conf.get(
      DFSConfigKeys.DFS_WEBHDFS_USER_PATTERN_KEY,
      DFSConfigKeys.DFS_WEBHDFS_USER_PATTERN_DEFAULT));

  connectionFactory = URLConnectionFactory
      .newDefaultURLConnectionFactory(conf);

  ugi = UserGroupInformation.getCurrentUser();
  this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());
  this.nnAddrs = resolveNNAddr();

  boolean isHA = HAUtil.isClientFailoverConfigured(conf, this.uri);
  boolean isLogicalUri = isHA && HAUtil.isLogicalUri(conf, this.uri);
  // In non-HA or non-logical URI case, the code needs to call
  // getCanonicalUri() in order to handle the case where no port is
  // specified in the URI
  this.tokenServiceName = isLogicalUri ?
      HAUtil.buildTokenServiceForLogicalUri(uri, getScheme())
      : SecurityUtil.buildTokenService(getCanonicalUri());

  if (!isHA) {
    this.retryPolicy =
        RetryUtils.getDefaultRetryPolicy(
            conf,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_KEY,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_KEY,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
            SafeModeException.class);
  } else {

    int maxFailoverAttempts = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_MAX_ATTEMPTS_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_MAX_ATTEMPTS_DEFAULT);
    int maxRetryAttempts = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_MAX_ATTEMPTS_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_MAX_ATTEMPTS_DEFAULT);
    int failoverSleepBaseMillis = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_BASE_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_BASE_DEFAULT);
    int failoverSleepMaxMillis = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_MAX_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_MAX_DEFAULT);

    this.retryPolicy = RetryPolicies
        .failoverOnNetworkException(RetryPolicies.TRY_ONCE_THEN_FAIL,
            maxFailoverAttempts, maxRetryAttempts, failoverSleepBaseMillis,
            failoverSleepMaxMillis);
  }

  this.workingDir = getHomeDirectory();
  this.canRefreshDelegationToken = UserGroupInformation.isSecurityEnabled();
  this.disallowFallbackToInsecureCluster = !conf.getBoolean(
      CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_KEY,
      CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_DEFAULT);
  this.delegationToken = null;
}
项目：hadoop-2.6.0-cdh5.4.3    文件：NameNodeProxies.java    
private static ClientProtocol createNNProxyWithClientProtocol(
    InetSocketAddress address, Configuration conf, UserGroupInformation ugi,
    boolean withRetries, AtomicBoolean fallbackToSimpleAuth)
    throws IOException {
  RPC.setProtocolEngine(conf, ClientNamenodeProtocolPB.class, ProtobufRpcEngine.class);

  final RetryPolicy defaultPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class);

  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).getProxy();

  if (withRetries) { // create the proxy with retries

    RetryPolicy createPolicy = RetryPolicies
        .retryUpToMaximumCountWithFixedSleep(5,
            HdfsConstants.LEASE_SOFTLIMIT_PERIOD, TimeUnit.MILLISECONDS);

    Map<Class<? extends Exception>, RetryPolicy> remoteExceptionToPolicyMap 
               = new HashMap<Class<? extends Exception>, RetryPolicy>();
    remoteExceptionToPolicyMap.put(AlreadyBeingCreatedException.class,
        createPolicy);

    RetryPolicy methodPolicy = RetryPolicies.retryByRemoteException(
        defaultPolicy, remoteExceptionToPolicyMap);
    Map<String, RetryPolicy> methodNameToPolicyMap 
               = new HashMap<String, RetryPolicy>();

    methodNameToPolicyMap.put("create", methodPolicy);

    ClientProtocol translatorProxy =
      new ClientNamenodeProtocolTranslatorPB(proxy);
    return (ClientProtocol) RetryProxy.create(
        ClientProtocol.class,
        new DefaultFailoverProxyProvider<ClientProtocol>(
            ClientProtocol.class, translatorProxy),
        methodNameToPolicyMap,
        defaultPolicy);
  } else {
    return new ClientNamenodeProtocolTranslatorPB(proxy);
  }
}
项目：hadoop-2.6.0-cdh5.4.3    文件：WebHdfsFileSystem.java    
@Override
public synchronized void initialize(URI uri, Configuration conf
    ) throws IOException {
  super.initialize(uri, conf);
  setConf(conf);
  /** set user pattern based on configuration file */
  UserParam.setUserPattern(conf.get(
      DFSConfigKeys.DFS_WEBHDFS_USER_PATTERN_KEY,
      DFSConfigKeys.DFS_WEBHDFS_USER_PATTERN_DEFAULT));

  connectionFactory = URLConnectionFactory
      .newDefaultURLConnectionFactory(conf);

  ugi = UserGroupInformation.getCurrentUser();
  this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());
  this.nnAddrs = resolveNNAddr();

  boolean isHA = HAUtil.isClientFailoverConfigured(conf, this.uri);
  boolean isLogicalUri = isHA && HAUtil.isLogicalUri(conf, this.uri);
  // In non-HA or non-logical URI case, the code needs to call
  // getCanonicalUri() in order to handle the case where no port is
  // specified in the URI
  this.tokenServiceName = isLogicalUri ?
      HAUtil.buildTokenServiceForLogicalUri(uri, getScheme())
      : SecurityUtil.buildTokenService(getCanonicalUri());

  if (!isHA) {
    this.retryPolicy =
        RetryUtils.getDefaultRetryPolicy(
            conf,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_KEY,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_KEY,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
            SafeModeException.class);
  } else {

    int maxFailoverAttempts = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_MAX_ATTEMPTS_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_MAX_ATTEMPTS_DEFAULT);
    int maxRetryAttempts = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_MAX_ATTEMPTS_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_MAX_ATTEMPTS_DEFAULT);
    int failoverSleepBaseMillis = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_BASE_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_BASE_DEFAULT);
    int failoverSleepMaxMillis = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_MAX_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_MAX_DEFAULT);

    this.retryPolicy = RetryPolicies
        .failoverOnNetworkException(RetryPolicies.TRY_ONCE_THEN_FAIL,
            maxFailoverAttempts, maxRetryAttempts, failoverSleepBaseMillis,
            failoverSleepMaxMillis);
  }

  this.workingDir = getHomeDirectory();
  this.canRefreshDelegationToken = UserGroupInformation.isSecurityEnabled();
  this.disallowFallbackToInsecureCluster = !conf.getBoolean(
      CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_KEY,
      CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_DEFAULT);
  this.delegationToken = null;
}
项目：FlexMap    文件：NameNodeProxies.java    
private static ClientProtocol createNNProxyWithClientProtocol(
    InetSocketAddress address, Configuration conf, UserGroupInformation ugi,
    boolean withRetries, AtomicBoolean fallbackToSimpleAuth)
    throws IOException {
  RPC.setProtocolEngine(conf, ClientNamenodeProtocolPB.class, ProtobufRpcEngine.class);

  final RetryPolicy defaultPolicy = 
      RetryUtils.getDefaultRetryPolicy(
          conf, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_KEY, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_ENABLED_DEFAULT, 
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_KEY,
          DFSConfigKeys.DFS_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
          SafeModeException.class);

  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).getProxy();

  if (withRetries) { // create the proxy with retries

    RetryPolicy createPolicy = RetryPolicies
        .retryUpToMaximumCountWithFixedSleep(5,
            HdfsConstants.LEASE_SOFTLIMIT_PERIOD, TimeUnit.MILLISECONDS);

    Map<Class<? extends Exception>, RetryPolicy> remoteExceptionToPolicyMap 
               = new HashMap<Class<? extends Exception>, RetryPolicy>();
    remoteExceptionToPolicyMap.put(AlreadyBeingCreatedException.class,
        createPolicy);

    RetryPolicy methodPolicy = RetryPolicies.retryByRemoteException(
        defaultPolicy, remoteExceptionToPolicyMap);
    Map<String, RetryPolicy> methodNameToPolicyMap 
               = new HashMap<String, RetryPolicy>();

    methodNameToPolicyMap.put("create", methodPolicy);

    ClientProtocol translatorProxy =
      new ClientNamenodeProtocolTranslatorPB(proxy);
    return (ClientProtocol) RetryProxy.create(
        ClientProtocol.class,
        new DefaultFailoverProxyProvider<ClientProtocol>(
            ClientProtocol.class, translatorProxy),
        methodNameToPolicyMap,
        defaultPolicy);
  } else {
    return new ClientNamenodeProtocolTranslatorPB(proxy);
  }
}
项目：FlexMap    文件：WebHdfsFileSystem.java    
@Override
public synchronized void initialize(URI uri, Configuration conf
    ) throws IOException {
  super.initialize(uri, conf);
  setConf(conf);
  /** set user pattern based on configuration file */
  UserParam.setUserPattern(conf.get(
      DFSConfigKeys.DFS_WEBHDFS_USER_PATTERN_KEY,
      DFSConfigKeys.DFS_WEBHDFS_USER_PATTERN_DEFAULT));

  connectionFactory = URLConnectionFactory
      .newDefaultURLConnectionFactory(conf);

  ugi = UserGroupInformation.getCurrentUser();
  this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());
  this.nnAddrs = resolveNNAddr();

  boolean isHA = HAUtil.isClientFailoverConfigured(conf, this.uri);
  boolean isLogicalUri = isHA && HAUtil.isLogicalUri(conf, this.uri);
  // In non-HA or non-logical URI case, the code needs to call
  // getCanonicalUri() in order to handle the case where no port is
  // specified in the URI
  this.tokenServiceName = isLogicalUri ?
      HAUtil.buildTokenServiceForLogicalUri(uri, getScheme())
      : SecurityUtil.buildTokenService(getCanonicalUri());

  if (!isHA) {
    this.retryPolicy =
        RetryUtils.getDefaultRetryPolicy(
            conf,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_KEY,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_KEY,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
            SafeModeException.class);
  } else {

    int maxFailoverAttempts = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_MAX_ATTEMPTS_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_MAX_ATTEMPTS_DEFAULT);
    int maxRetryAttempts = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_MAX_ATTEMPTS_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_MAX_ATTEMPTS_DEFAULT);
    int failoverSleepBaseMillis = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_BASE_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_BASE_DEFAULT);
    int failoverSleepMaxMillis = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_MAX_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_MAX_DEFAULT);

    this.retryPolicy = RetryPolicies
        .failoverOnNetworkException(RetryPolicies.TRY_ONCE_THEN_FAIL,
            maxFailoverAttempts, maxRetryAttempts, failoverSleepBaseMillis,
            failoverSleepMaxMillis);
  }

  this.workingDir = getHomeDirectory();
  this.canRefreshDelegationToken = UserGroupInformation.isSecurityEnabled();
  this.disallowFallbackToInsecureCluster = !conf.getBoolean(
      CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_KEY,
      CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_DEFAULT);
  this.delegationToken = null;
}
项目：hadoop-on-lustre2    文件：WebHdfsFileSystem.java    
@Override
public synchronized void initialize(URI uri, Configuration conf
    ) throws IOException {
  super.initialize(uri, conf);
  setConf(conf);
  /** set user pattern based on configuration file */
  UserParam.setUserPattern(conf.get(
      DFSConfigKeys.DFS_WEBHDFS_USER_PATTERN_KEY,
      DFSConfigKeys.DFS_WEBHDFS_USER_PATTERN_DEFAULT));

  connectionFactory = URLConnectionFactory
      .newDefaultURLConnectionFactory(conf);

  ugi = UserGroupInformation.getCurrentUser();
  this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());
  this.nnAddrs = DFSUtil.resolveWebHdfsUri(this.uri, conf);

  boolean isHA = HAUtil.isLogicalUri(conf, this.uri);
  // In non-HA case, the code needs to call getCanonicalUri() in order to
  // handle the case where no port is specified in the URI
  this.tokenServiceName = isHA ? HAUtil.buildTokenServiceForLogicalUri(uri)
      : SecurityUtil.buildTokenService(getCanonicalUri());
  initializeTokenAspect();

  if (!isHA) {
    this.retryPolicy =
        RetryUtils.getDefaultRetryPolicy(
            conf,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_KEY,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_ENABLED_DEFAULT,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_KEY,
            DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT,
            SafeModeException.class);
  } else {

    int maxFailoverAttempts = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_MAX_ATTEMPTS_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_MAX_ATTEMPTS_DEFAULT);
    int maxRetryAttempts = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_MAX_ATTEMPTS_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_RETRY_MAX_ATTEMPTS_DEFAULT);
    int failoverSleepBaseMillis = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_BASE_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_BASE_DEFAULT);
    int failoverSleepMaxMillis = conf.getInt(
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_MAX_KEY,
        DFSConfigKeys.DFS_HTTP_CLIENT_FAILOVER_SLEEPTIME_MAX_DEFAULT);

    this.retryPolicy = RetryPolicies
        .failoverOnNetworkException(RetryPolicies.TRY_ONCE_THEN_FAIL,
            maxFailoverAttempts, maxRetryAttempts, failoverSleepBaseMillis,
            failoverSleepMaxMillis);
  }

  this.workingDir = getHomeDirectory();

  if (UserGroupInformation.isSecurityEnabled()) {
    tokenAspect.initDelegationToken(ugi);
  }
}