### Question:
ZKClientPool { public static ZkClient getZkClient(String zkServer) { if (_zkClientMap.containsKey(zkServer)) { ZkClient zkClient = _zkClientMap.get(zkServer); if (zkClient.getConnection().getZookeeperState() == States.CONNECTED) { return zkClient; } } synchronized (_zkClientMap) { if (_zkClientMap.containsKey(zkServer)) { ZkClient zkClient = _zkClientMap.get(zkServer); if (zkClient.getConnection().getZookeeperState() != States.CONNECTED) { _zkClientMap.remove(zkServer); } } if (!_zkClientMap.containsKey(zkServer)) { ZkClient zkClient = new ZkClient(zkServer, DEFAULT_SESSION_TIMEOUT, ZkClient.DEFAULT_CONNECTION_TIMEOUT, new ZNRecordSerializer()); _zkClientMap.put(zkServer, zkClient); } return _zkClientMap.get(zkServer); } }  static ZkClient getZkClient(String zkServer); static void reset(); static void main(String[] args);  }### Answer:
@Test public void test() throws Exception { String testName = "TestZKClientPool"; System.out.println("START " + testName + " at " + new Date(System.currentTimeMillis())); String zkAddr = "localhost:21891"; ZkServer zkServer = TestHelper.startZkServer(zkAddr); ZkClient zkClient = ZKClientPool.getZkClient(zkAddr); zkClient.createPersistent("/" + testName, new ZNRecord(testName)); ZNRecord record = zkClient.readData("/" + testName); Assert.assertEquals(record.getId(), testName); TestHelper.stopZkServer(zkServer); zkServer = TestHelper.startZkServer(zkAddr); try { zkClient = ZKClientPool.getZkClient(zkAddr); record = zkClient.readData("/" + testName); Assert.fail("should fail on zk no node exception"); } catch (ZkNoNodeException e) { } catch (Exception e) { Assert.fail("should not fail on exception other than ZkNoNodeException"); } zkClient.createPersistent("/" + testName, new ZNRecord(testName)); record = zkClient.readData("/" + testName); Assert.assertEquals(record.getId(), testName); zkClient.close(); TestHelper.stopZkServer(zkServer); System.out.println("END " + testName + " at " + new Date(System.currentTimeMillis())); }