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

ZkCoordinationUtilsFactory介绍

暂无

代码示例

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

private ZkUtils getZkUtilsWithNewClient() {
  ZkClient zkClient = ZkCoordinationUtilsFactory
    .createZkClient(testZkConnectionString, SESSION_TIMEOUT_MS, CONNECTION_TIMEOUT_MS);
  return new ZkUtils(
    KEY_BUILDER,
    zkClient,
    CONNECTION_TIMEOUT_MS, SESSION_TIMEOUT_MS, new NoOpMetricsRegistry());
 }
}

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

/**
 * helper method to create zkClient
 * @param connectString - zkConnect string
 * @param sessionTimeoutMS - session timeout
 * @param connectionTimeoutMs - connection timeout
 * @return zkClient object
 */
public static ZkClient createZkClient(String connectString, int sessionTimeoutMS, int connectionTimeoutMs) {
 ZkClient zkClient;
 try {
  zkClient = new ZkClient(connectString, sessionTimeoutMS, connectionTimeoutMs, new SerializableSerializer(), connectionTimeoutMs);
 } catch (Exception e) {
  // ZkClient constructor may throw a variety of different exceptions, not all of them Zk based.
  throw new SamzaException("zkClient failed to connect to ZK at :" + connectString, e);
 }
 // make sure the namespace in zk exists (if specified)
 validateZkNameSpace(connectString, zkClient);
 return zkClient;
}

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

private ZkUtils getZkUtilsWithNewClient(String processorId) {
  ZkClient zkClient = ZkCoordinationUtilsFactory
    .createZkClient(testZkConnectionString, SESSION_TIMEOUT_MS, CONNECTION_TIMEOUT_MS);
  return new ZkUtils(KEY_BUILDER, zkClient, CONNECTION_TIMEOUT_MS, SESSION_TIMEOUT_MS, new NoOpMetricsRegistry());
 }
}

代码示例来源：origin: org.apache.samza/samza-core_2.10

/**
 * helper method to create zkClient
 * @param connectString - zkConnect string
 * @param sessionTimeoutMS - session timeout
 * @param connectionTimeoutMs - connection timeout
 * @return zkClient object
 */
public static ZkClient createZkClient(String connectString, int sessionTimeoutMS, int connectionTimeoutMs) {
 ZkClient zkClient;
 try {
  zkClient = new ZkClient(connectString, sessionTimeoutMS, connectionTimeoutMs, new SerializableSerializer(), connectionTimeoutMs);
 } catch (Exception e) {
  // ZkClient constructor may throw a variety of different exceptions, not all of them Zk based.
  throw new SamzaException("zkClient failed to connect to ZK at :" + connectString, e);
 }
 // make sure the namespace in zk exists (if specified)
 validateZkNameSpace(connectString, zkClient);
 return zkClient;
}

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

private ZkUtils getZkUtils(Config config, MetricsRegistry metricsRegistry, String coordinatorZkBasePath) {
 ZkConfig zkConfig = new ZkConfig(config);
 ZkKeyBuilder keyBuilder = new ZkKeyBuilder(coordinatorZkBasePath);
 ZkClient zkClient = ZkCoordinationUtilsFactory
   .createZkClient(zkConfig.getZkConnect(), zkConfig.getZkSessionTimeoutMs(), zkConfig.getZkConnectionTimeoutMs());
 return new ZkUtils(keyBuilder, zkClient, zkConfig.getZkConnectionTimeoutMs(), zkConfig.getZkSessionTimeoutMs(), metricsRegistry);
}

代码示例来源：origin: org.apache.samza/samza-core_2.11

/**
 * helper method to create zkClient
 * @param connectString - zkConnect string
 * @param sessionTimeoutMS - session timeout
 * @param connectionTimeoutMs - connection timeout
 * @return zkClient object
 */
public static ZkClient createZkClient(String connectString, int sessionTimeoutMS, int connectionTimeoutMs) {
 ZkClient zkClient;
 try {
  zkClient = new ZkClient(connectString, sessionTimeoutMS, connectionTimeoutMs, new SerializableSerializer(), connectionTimeoutMs);
 } catch (Exception e) {
  // ZkClient constructor may throw a variety of different exceptions, not all of them Zk based.
  throw new SamzaException("zkClient failed to connect to ZK at :" + connectString, e);
 }
 // make sure the namespace in zk exists (if specified)
 validateZkNameSpace(connectString, zkClient);
 return zkClient;
}

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

private ZkUtils getZkUtils(Config config, MetricsRegistry metricsRegistry, String coordinatorZkBasePath) {
 ZkConfig zkConfig = new ZkConfig(config);
 ZkKeyBuilder keyBuilder = new ZkKeyBuilder(coordinatorZkBasePath);
 ZkClient zkClient = ZkCoordinationUtilsFactory
   .createZkClient(zkConfig.getZkConnect(), zkConfig.getZkSessionTimeoutMs(), zkConfig.getZkConnectionTimeoutMs());
 return new ZkUtils(keyBuilder, zkClient, zkConfig.getZkConnectionTimeoutMs(), zkConfig.getZkSessionTimeoutMs(), metricsRegistry);
}

代码示例来源：origin: org.apache.samza/samza-core_2.12

/**
 * helper method to create zkClient
 * @param connectString - zkConnect string
 * @param sessionTimeoutMS - session timeout
 * @param connectionTimeoutMs - connection timeout
 * @return zkClient object
 */
public static ZkClient createZkClient(String connectString, int sessionTimeoutMS, int connectionTimeoutMs) {
 ZkClient zkClient;
 try {
  zkClient = new ZkClient(connectString, sessionTimeoutMS, connectionTimeoutMs, new SerializableSerializer(), connectionTimeoutMs);
 } catch (Exception e) {
  // ZkClient constructor may throw a variety of different exceptions, not all of them Zk based.
  throw new SamzaException("zkClient failed to connect to ZK at :" + connectString, e);
 }
 // make sure the namespace in zk exists (if specified)
 validateZkNameSpace(connectString, zkClient);
 return zkClient;
}

代码示例来源：origin: org.apache.samza/samza-core_2.12

private ZkUtils getZkUtils(Config config, MetricsRegistry metricsRegistry, String coordinatorZkBasePath) {
 ZkConfig zkConfig = new ZkConfig(config);
 ZkKeyBuilder keyBuilder = new ZkKeyBuilder(coordinatorZkBasePath);
 ZkClient zkClient = ZkCoordinationUtilsFactory
   .createZkClient(zkConfig.getZkConnect(), zkConfig.getZkSessionTimeoutMs(), zkConfig.getZkConnectionTimeoutMs());
 return new ZkUtils(keyBuilder, zkClient, zkConfig.getZkConnectionTimeoutMs(), zkConfig.getZkSessionTimeoutMs(), metricsRegistry);
}

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

/**
 * helper method to create zkClient
 * @param connectString - zkConnect string
 * @param sessionTimeoutMS - session timeout
 * @param connectionTimeoutMs - connection timeout
 * @return zkClient object
 */
public static ZkClient createZkClient(String connectString, int sessionTimeoutMS, int connectionTimeoutMs) {
 ZkClient zkClient;
 try {
  zkClient = new ZkClient(connectString, sessionTimeoutMS, connectionTimeoutMs, new SerializableSerializer(), connectionTimeoutMs);
 } catch (Exception e) {
  // ZkClient constructor may throw a variety of different exceptions, not all of them Zk based.
  throw new SamzaException("zkClient failed to connect to ZK at :" + connectString, e);
 }
 // make sure the namespace in zk exists (if specified)
 validateZkNameSpace(connectString, zkClient);
 return zkClient;
}

代码示例来源：origin: org.apache.samza/samza-core_2.10

private ZkUtils getZkUtils(Config config, MetricsRegistry metricsRegistry, String coordinatorZkBasePath) {
 ZkConfig zkConfig = new ZkConfig(config);
 ZkKeyBuilder keyBuilder = new ZkKeyBuilder(coordinatorZkBasePath);
 ZkClient zkClient = ZkCoordinationUtilsFactory
   .createZkClient(zkConfig.getZkConnect(), zkConfig.getZkSessionTimeoutMs(), zkConfig.getZkConnectionTimeoutMs());
 return new ZkUtils(keyBuilder, zkClient, zkConfig.getZkConnectionTimeoutMs(), zkConfig.getZkSessionTimeoutMs(), metricsRegistry);
}

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

@Test
public void testValidateFailZkNameSpaceEmptyPath() {
 // should succeed, because no namespace provided
 String zkConnect = "127.0.0.1:" + zkServer.getPort() + "";
 initZk(zkConnect);
 ZkCoordinationUtilsFactory.validateZkNameSpace(zkConnect, zkClient);
 tearDownZk();
}

代码示例来源：origin: org.apache.samza/samza-core_2.11

private ZkUtils getZkUtils(Config config, MetricsRegistry metricsRegistry, String coordinatorZkBasePath) {
 ZkConfig zkConfig = new ZkConfig(config);
 ZkKeyBuilder keyBuilder = new ZkKeyBuilder(coordinatorZkBasePath);
 ZkClient zkClient = ZkCoordinationUtilsFactory
   .createZkClient(zkConfig.getZkConnect(), zkConfig.getZkSessionTimeoutMs(), zkConfig.getZkConnectionTimeoutMs());
 return new ZkUtils(keyBuilder, zkClient, zkConfig.getZkConnectionTimeoutMs(), zkConfig.getZkSessionTimeoutMs(), metricsRegistry);
}

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

@Test
public void testValidateFailZkNameSpace2LevelPath() {
 try {
  String zkConnect = "127.0.0.1:" + zkServer.getPort() + "/zkNameSpace/xyz";
  initZk(zkConnect);
  ZkCoordinationUtilsFactory.validateZkNameSpace(zkConnect, zkClient);
  Assert.fail("2.Should fail with exception, because namespace doesn't exist");
 } catch (SamzaException e) {
  // expected
 } finally {
  tearDownZk();
 }
}

代码示例来源：origin: org.apache.samza/samza-core_2.11

public CoordinationUtils getCoordinationUtils(String groupId, String participantId, Config config) {
 ZkConfig zkConfig = new ZkConfig(config);
 ZkClient zkClient =
   createZkClient(zkConfig.getZkConnect(), zkConfig.getZkSessionTimeoutMs(), zkConfig.getZkConnectionTimeoutMs());
 ZkUtils zkUtils = new ZkUtils(new ZkKeyBuilder(groupId), zkClient, zkConfig.getZkConnectionTimeoutMs(), zkConfig.getZkSessionTimeoutMs(),
   new NoOpMetricsRegistry());
 return new ZkCoordinationUtils(participantId, zkConfig, zkUtils);
}

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

@Test
public void testValidateFailZkNameSpace1LevelPath() {
 try {
  String zkConnect = "127.0.0.1:" + zkServer.getPort() + "/zkNameSpace";
  initZk(zkConnect);
  ZkCoordinationUtilsFactory.validateZkNameSpace(zkConnect, zkClient);
  Assert.fail("1.Should fail with exception, because namespace doesn't exist");
 } catch (SamzaException e) {
  // expected
 } finally {
  tearDownZk();
 }
}

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

public CoordinationUtils getCoordinationUtils(String groupId, String participantId, Config config) {
 ZkConfig zkConfig = new ZkConfig(config);
 ZkClient zkClient =
   createZkClient(zkConfig.getZkConnect(), zkConfig.getZkSessionTimeoutMs(), zkConfig.getZkConnectionTimeoutMs());
 ZkUtils zkUtils = new ZkUtils(new ZkKeyBuilder(groupId), zkClient, zkConfig.getZkConnectionTimeoutMs(), zkConfig.getZkSessionTimeoutMs(),
   new NoOpMetricsRegistry());
 return new ZkCoordinationUtils(participantId, zkConfig, zkUtils);
}

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

private void testDoNotFailIfNameSpacePresent(String zkNameSpace) {
 String zkConnect = "127.0.0.1:" + zkServer.getPort() + zkNameSpace;
 createNamespace(zkNameSpace);
 initZk(zkConnect);
 ZkCoordinationUtilsFactory.validateZkNameSpace(zkConnect, zkClient);
 zkClient.createPersistent("/test");
 zkClient.createPersistent("/test/test1");
 // test if the new root exists
 Assert.assertTrue(zkClient.exists("/"));
 Assert.assertTrue(zkClient.exists("/test"));
 Assert.assertTrue(zkClient.exists("/test/test1"));
}

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

public CoordinationUtils getCoordinationUtils(String groupId, String participantId, Config config) {
 ZkConfig zkConfig = new ZkConfig(config);
 ZkClient zkClient =
   createZkClient(zkConfig.getZkConnect(), zkConfig.getZkSessionTimeoutMs(), zkConfig.getZkConnectionTimeoutMs());
 ZkUtils zkUtils = new ZkUtils(new ZkKeyBuilder(groupId), zkClient, zkConfig.getZkConnectionTimeoutMs(), zkConfig.getZkSessionTimeoutMs(),
   new NoOpMetricsRegistry());
 return new ZkCoordinationUtils(participantId, zkConfig, zkUtils);
}

代码示例来源：origin: org.apache.samza/samza-core_2.10

public CoordinationUtils getCoordinationUtils(String groupId, String participantId, Config config) {
 ZkConfig zkConfig = new ZkConfig(config);
 ZkClient zkClient =
   createZkClient(zkConfig.getZkConnect(), zkConfig.getZkSessionTimeoutMs(), zkConfig.getZkConnectionTimeoutMs());
 ZkUtils zkUtils = new ZkUtils(new ZkKeyBuilder(groupId), zkClient, zkConfig.getZkConnectionTimeoutMs(), zkConfig.getZkSessionTimeoutMs(),
   new NoOpMetricsRegistry());
 return new ZkCoordinationUtils(participantId, zkConfig, zkUtils);
}