本文整理了Java中org.apache.twill.zookeeper.ZKClients.reWatchOnExpire()方法的一些代码示例，展示了ZKClients.reWatchOnExpire()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。ZKClients.reWatchOnExpire()方法的具体详情如下：
包路径：org.apache.twill.zookeeper.ZKClients
类名称：ZKClients
方法名：reWatchOnExpire

ZKClients.reWatchOnExpire介绍

[英]Creates a ZKClient that will perform auto re-watch on all existing watches when reconnection happens after session expiration. All org.apache.zookeeper.Watcherset through the returned ZKClient would not receive any connection events.
[中]创建一个ZKClient，当会话过期后重新连接时，该客户端将对所有现有手表执行自动重新监视。所有组织。阿帕奇。动物园管理员。通过返回的ZKClient的Watcherset不会接收任何连接事件。

代码示例

代码示例来源：origin: apache/phoenix

ZKClients.reWatchOnExpire(
  ZKClients.retryOnFailure(
     new TephraZKClientService(zkQuorumServersString, timeOut, null,

代码示例来源：origin: apache/phoenix

@Override
public PhoenixTransactionService getTransactionService(Configuration config, ConnectionInfo connInfo, int port) {
  config.setInt(TxConstants.Service.CFG_DATA_TX_BIND_PORT, port);
  int retryTimeOut = config.getInt(TxConstants.Service.CFG_DATA_TX_CLIENT_DISCOVERY_TIMEOUT_SEC, 
      TxConstants.Service.DEFAULT_DATA_TX_CLIENT_DISCOVERY_TIMEOUT_SEC);
  ZKClientService zkClient = ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(
     ZKClientService.Builder.of(connInfo.getZookeeperConnectionString())
      .setSessionTimeout(config.getInt(HConstants.ZK_SESSION_TIMEOUT,
          HConstants.DEFAULT_ZK_SESSION_TIMEOUT))
      .build(),
     RetryStrategies.exponentialDelay(500, retryTimeOut, TimeUnit.MILLISECONDS)
    )
   )
  );
  DiscoveryService discovery = new ZKDiscoveryService(zkClient);
  TransactionManager txManager = new TransactionManager(config, new HDFSTransactionStateStorage(config, 
      new SnapshotCodecProvider(config), new TxMetricsCollector()), new TxMetricsCollector());
  TransactionService txService = new TransactionService(config, zkClient, discovery, Providers.of(txManager));
  TephraTransactionService service = new TephraTransactionService(zkClient, txService);
  service.start();
  return service;
}

代码示例来源：origin: co.cask.tephra/tephra-core

@Provides
 @Singleton
 private ZKClientService provideZKClientService(Configuration conf) {
  String zkStr = conf.get(TxConstants.Service.CFG_DATA_TX_ZOOKEEPER_QUORUM);
  if (zkStr == null) {
   // Default to HBase one.
   zkStr = conf.get(TxConstants.HBase.ZOOKEEPER_QUORUM);
  }

  int timeOut = conf.getInt(TxConstants.HBase.ZK_SESSION_TIMEOUT, TxConstants.HBase.DEFAULT_ZK_SESSION_TIMEOUT);
  ZKClientService zkClientService = new TephraZKClientService(zkStr, timeOut, null,
                                ArrayListMultimap.<String, byte[]>create());
  return ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(zkClientService, RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
    )
   )
  );
 }
}

代码示例来源：origin: caskdata/tephra

@Provides
 @Singleton
 private ZKClientService provideZKClientService(Configuration conf) {
  String zkStr = conf.get(TxConstants.Service.CFG_DATA_TX_ZOOKEEPER_QUORUM);
  if (zkStr == null) {
   // Default to HBase one.
   zkStr = conf.get(TxConstants.HBase.ZOOKEEPER_QUORUM);
  }

  int timeOut = conf.getInt(TxConstants.HBase.ZK_SESSION_TIMEOUT, TxConstants.HBase.DEFAULT_ZK_SESSION_TIMEOUT);
  ZKClientService zkClientService = new TephraZKClientService(zkStr, timeOut, null,
                                ArrayListMultimap.<String, byte[]>create());
  return ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(zkClientService, RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
    )
   )
  );
 }
}

代码示例来源：origin: org.apache.tephra/tephra-core

@Provides
 @Singleton
 private ZKClientService provideZKClientService(Configuration conf) {
  String zkStr = conf.get(TxConstants.Service.CFG_DATA_TX_ZOOKEEPER_QUORUM);
  if (zkStr == null) {
   // Default to HBase one.
   zkStr = conf.get(TxConstants.HBase.ZOOKEEPER_QUORUM);
  }

  int timeOut = conf.getInt(TxConstants.HBase.ZK_SESSION_TIMEOUT, TxConstants.HBase.DEFAULT_ZK_SESSION_TIMEOUT);
  ZKClientService zkClientService = new TephraZKClientService(zkStr, timeOut, null,
                                ArrayListMultimap.<String, byte[]>create());
  return ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(zkClientService, RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
    )
   )
  );
 }
}

代码示例来源：origin: caskdata/coopr

ZKClients.reWatchOnExpire(
 ZKClients.retryOnFailure(zkClientService,
              RetryStrategies.limit(5, RetryStrategies.fixDelay(5, TimeUnit.SECONDS))));

代码示例来源：origin: caskdata/cdap

@Provides
 @Singleton
 private ZKClientService provideZKClientService(CConfiguration cConf) {
  String zookeeper = cConf.get(Constants.Zookeeper.QUORUM);
  Preconditions.checkNotNull(zookeeper, "Missing ZooKeeper configuration '%s'", Constants.Zookeeper.QUORUM);

  return ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(
     ZKClientService.Builder.of(cConf.get(Constants.Zookeeper.QUORUM))
      .setSessionTimeout(cConf.getInt(Constants.Zookeeper.CFG_SESSION_TIMEOUT_MILLIS,
                      Constants.Zookeeper.DEFAULT_SESSION_TIMEOUT_MILLIS))
      .build(),
     RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
    )
   )
  );
 }
}

代码示例来源：origin: co.cask.cdap/cdap-common

@Provides
 @Singleton
 private ZKClientService provideZKClientService(CConfiguration cConf) {
  String zookeeper = cConf.get(Constants.Zookeeper.QUORUM);
  Preconditions.checkNotNull(zookeeper, "Missing ZooKeeper configuration '%s'", Constants.Zookeeper.QUORUM);

  return ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(
     ZKClientService.Builder.of(cConf.get(Constants.Zookeeper.QUORUM))
      .setSessionTimeout(cConf.getInt(Constants.Zookeeper.CFG_SESSION_TIMEOUT_MILLIS,
                      Constants.Zookeeper.DEFAULT_SESSION_TIMEOUT_MILLIS))
      .build(),
     RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
    )
   )
  );
 }
}

代码示例来源：origin: co.cask.tigon/tigon-yarn

@Provides
 @Singleton
 private ZKClientService provideZKClientService(CConfiguration cConf) {
  String zookeeper = cConf.get(Constants.Zookeeper.QUORUM);
  Preconditions.checkNotNull(zookeeper, "Missing Zookeeper configuration '%s'", Constants.Zookeeper.QUORUM);

  return ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(
     ZKClientService.Builder.of(cConf.get(Constants.Zookeeper.QUORUM))
      .setSessionTimeout(cConf.getInt(Constants.Zookeeper.CFG_SESSION_TIMEOUT_MILLIS,
                      Constants.Zookeeper.DEFAULT_SESSION_TIMEOUT_MILLIS))
      .build(),
     RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
    )
   )
  );
 }
}

代码示例来源：origin: org.apache.twill/twill-yarn

private ZKClientService getZKClientService(String zkConnect) {
 return ZKClientServices.delegate(
  ZKClients.reWatchOnExpire(
   ZKClients.retryOnFailure(ZKClientService.Builder.of(zkConnect)
                 .setSessionTimeout(ZK_TIMEOUT)
                 .build(), RetryStrategies.exponentialDelay(100, 2000, TimeUnit.MILLISECONDS))));
}

代码示例来源：origin: apache/twill

private ZKClientService getZKClientService(String zkConnect) {
 return ZKClientServices.delegate(
  ZKClients.reWatchOnExpire(
   ZKClients.retryOnFailure(ZKClientService.Builder.of(zkConnect)
                 .setSessionTimeout(ZK_TIMEOUT)
                 .build(), RetryStrategies.exponentialDelay(100, 2000, TimeUnit.MILLISECONDS))));
}

代码示例来源：origin: org.apache.twill/twill-core

@Override
public void start() {
 Preconditions.checkNotNull(zkConnectStr);
 eventConverter = new LogEventConverter(hostname, runnableName);
 scheduler = Executors.newSingleThreadScheduledExecutor(Threads.createDaemonThreadFactory(PUBLISH_THREAD_NAME));
 zkClientService = ZKClientServices.delegate(
  ZKClients.reWatchOnExpire(
   ZKClients.retryOnFailure(ZKClientService.Builder.of(zkConnectStr).build(),
                RetryStrategies.fixDelay(1, TimeUnit.SECONDS))));
 kafkaClient = new ZKKafkaClientService(zkClientService);
 Futures.addCallback(Services.chainStart(zkClientService, kafkaClient),
           new FutureCallback<List<ListenableFuture<Service.State>>>() {
  @Override
  public void onSuccess(List<ListenableFuture<Service.State>> result) {
   for (ListenableFuture<Service.State> future : result) {
    Preconditions.checkState(Futures.getUnchecked(future) == Service.State.RUNNING,
                 "Service is not running.");
   }
   addInfo("Kafka client started: " + zkConnectStr);
   scheduler.scheduleWithFixedDelay(flushTask, 0, flushPeriod, TimeUnit.MILLISECONDS);
  }
  @Override
  public void onFailure(Throwable t) {
   // Fail to talk to kafka. Other than logging, what can be done?
   addError("Failed to start kafka appender.", t);
  }
 }, Threads.SAME_THREAD_EXECUTOR);
 super.start();
}

代码示例来源：origin: apache/twill

@Override
public void start() {
 Preconditions.checkNotNull(zkConnectStr);
 eventConverter = new LogEventConverter(hostname, runnableName);
 scheduler = Executors.newSingleThreadScheduledExecutor(Threads.createDaemonThreadFactory(PUBLISH_THREAD_NAME));
 zkClientService = ZKClientServices.delegate(
  ZKClients.reWatchOnExpire(
   ZKClients.retryOnFailure(ZKClientService.Builder.of(zkConnectStr).build(),
                RetryStrategies.fixDelay(1, TimeUnit.SECONDS))));
 kafkaClient = new ZKKafkaClientService(zkClientService);
 Futures.addCallback(Services.chainStart(zkClientService, kafkaClient),
           new FutureCallback<List<ListenableFuture<Service.State>>>() {
  @Override
  public void onSuccess(List<ListenableFuture<Service.State>> result) {
   for (ListenableFuture<Service.State> future : result) {
    Preconditions.checkState(Futures.getUnchecked(future) == Service.State.RUNNING,
                 "Service is not running.");
   }
   addInfo("Kafka client started: " + zkConnectStr);
   scheduler.scheduleWithFixedDelay(flushTask, 0, flushPeriod, TimeUnit.MILLISECONDS);
  }
  @Override
  public void onFailure(Throwable t) {
   // Fail to talk to kafka. Other than logging, what can be done?
   addError("Failed to start kafka appender.", t);
  }
 }, Threads.SAME_THREAD_EXECUTOR);
 super.start();
}

代码示例来源：origin: com.aliyun.phoenix/ali-phoenix-core

ZKClients.reWatchOnExpire(
  ZKClients.retryOnFailure(
     new TephraZKClientService(zkQuorumServersString, timeOut, null,

代码示例来源：origin: org.apache.phoenix/phoenix-core

ZKClients.reWatchOnExpire(
  ZKClients.retryOnFailure(
     new TephraZKClientService(zkQuorumServersString, timeOut, null,

代码示例来源：origin: caskdata/coopr

private ZKClientService getZKService(String connectString) {
  return ZKClientServices.delegate(
   ZKClients.namespace(
    ZKClients.reWatchOnExpire(
     ZKClients.retryOnFailure(
      ZKClientService.Builder.of(connectString)
       .setSessionTimeout(conf.getInt(Constants.ZOOKEEPER_SESSION_TIMEOUT_MILLIS))
       .build(),
      RetryStrategies.fixDelay(2, TimeUnit.SECONDS)
     )
    ), conf.get(Constants.ZOOKEEPER_NAMESPACE)
   )
  );
 }
}

代码示例来源：origin: apache/twill

/**
 * Creates a {@link ZKClientService}.
 */
protected final ZKClientService createZKClient() {
 TwillRuntimeSpecification twillRuntimeSpec = getTwillRuntimeSpecification();
 return ZKClientServices.delegate(
  ZKClients.namespace(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(
     ZKClientService.Builder.of(twillRuntimeSpec.getZkConnectStr()).build(),
     RetryStrategies.fixDelay(1, TimeUnit.SECONDS)
    )
   ), "/" + twillRuntimeSpec.getTwillAppName()
  )
 );
}

代码示例来源：origin: org.apache.twill/twill-yarn

/**
 * Creates a {@link ZKClientService}.
 */
protected final ZKClientService createZKClient() {
 TwillRuntimeSpecification twillRuntimeSpec = getTwillRuntimeSpecification();
 return ZKClientServices.delegate(
  ZKClients.namespace(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(
     ZKClientService.Builder.of(twillRuntimeSpec.getZkConnectStr()).build(),
     RetryStrategies.fixDelay(1, TimeUnit.SECONDS)
    )
   ), "/" + twillRuntimeSpec.getTwillAppName()
  )
 );
}

代码示例来源：origin: com.aliyun.phoenix/ali-phoenix-core

@Override
public PhoenixTransactionService getTransactionService(Configuration config, ConnectionInfo connInfo, int port) {
  config.setInt(TxConstants.Service.CFG_DATA_TX_BIND_PORT, port);
  int retryTimeOut = config.getInt(TxConstants.Service.CFG_DATA_TX_CLIENT_DISCOVERY_TIMEOUT_SEC, 
      TxConstants.Service.DEFAULT_DATA_TX_CLIENT_DISCOVERY_TIMEOUT_SEC);
  ZKClientService zkClient = ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(
     ZKClientService.Builder.of(connInfo.getZookeeperConnectionString())
      .setSessionTimeout(config.getInt(HConstants.ZK_SESSION_TIMEOUT,
          HConstants.DEFAULT_ZK_SESSION_TIMEOUT))
      .build(),
     RetryStrategies.exponentialDelay(500, retryTimeOut, TimeUnit.MILLISECONDS)
    )
   )
  );
  DiscoveryService discovery = new ZKDiscoveryService(zkClient);
  TransactionManager txManager = new TransactionManager(config, new HDFSTransactionStateStorage(config, 
      new SnapshotCodecProvider(config), new TxMetricsCollector()), new TxMetricsCollector());
  TransactionService txService = new TransactionService(config, zkClient, discovery, Providers.of(txManager));
  TephraTransactionService service = new TephraTransactionService(zkClient, txService);
  service.start();
  return service;
}

代码示例来源：origin: org.apache.phoenix/phoenix-core

@Override
public PhoenixTransactionService getTransactionService(Configuration config, ConnectionInfo connInfo) {
  ZKClientService zkClient = ZKClientServices.delegate(
   ZKClients.reWatchOnExpire(
    ZKClients.retryOnFailure(
     ZKClientService.Builder.of(connInfo.getZookeeperConnectionString())
      .setSessionTimeout(config.getInt(HConstants.ZK_SESSION_TIMEOUT,
          HConstants.DEFAULT_ZK_SESSION_TIMEOUT))
      .build(),
     RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
    )
   )
  );
  //zkClient.startAndWait();
  DiscoveryService discovery = new ZKDiscoveryService(zkClient);
  TransactionManager txManager = new TransactionManager(config, new HDFSTransactionStateStorage(config, new SnapshotCodecProvider(config), new TxMetricsCollector()), new TxMetricsCollector());
  TransactionService txService = new TransactionService(config, zkClient, discovery, Providers.of(txManager));
  TephraTransactionService service = new TephraTransactionService(zkClient, txService);
  //txService.startAndWait();            
  service.start();
  return service;
}