Java 类org.apache.zookeeper.CreateMode 实例源码

@Test
public void testCreate()
        throws IOException, KeeperException, InterruptedException {
    Stat stat = new Stat();
    zk.create("/foo", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_WITH_TTL, stat, 100);
    Assert.assertEquals(0, stat.getEphemeralOwner());

    final AtomicLong fakeElapsed = new AtomicLong(0);
    ContainerManager containerManager = newContainerManager(fakeElapsed);
    containerManager.checkContainers();
    Assert.assertNotNull("Ttl node should not have been deleted yet", zk.exists("/foo", false));

    fakeElapsed.set(1000);
    containerManager.checkContainers();
    Assert.assertNull("Ttl node should have been deleted", zk.exists("/foo", false));
}

@Test
public void testChRootCreateDelete() throws Exception {
    // creating the subtree for chRoot clients.
    String chRoot = createNameSpace();
    // Creating child using chRoot client.
    zk_chroot = createClient(this.hostPort + chRoot);
    Op createChild = Op.create("/myid", new byte[0],
            Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    multi(zk_chroot, Arrays.asList(createChild));

    Assert.assertNotNull("zNode is not created under chroot:" + chRoot, zk
            .exists(chRoot + "/myid", false));
    Assert.assertNotNull("zNode is not created under chroot:" + chRoot,
            zk_chroot.exists("/myid", false));
    Assert.assertNull("zNode is created directly under '/', ignored configured chroot",
            zk.exists("/myid", false));

    // Deleting child using chRoot client.
    Op deleteChild = Op.delete("/myid", 0);
    multi(zk_chroot, Arrays.asList(deleteChild));
    Assert.assertNull("zNode exists under chroot:" + chRoot, zk.exists(
            chRoot + "/myid", false));
    Assert.assertNull("zNode exists under chroot:" + chRoot, zk_chroot
            .exists("/myid", false));
}

@Test
public void testTtls() {
    long ttls[] = {100, 1, EphemeralType.MAX_TTL};
    for (long ttl : ttls) {
        long ephemeralOwner = EphemeralType.ttlToEphemeralOwner(ttl);
        Assert.assertEquals(EphemeralType.TTL, EphemeralType.get(ephemeralOwner));
        Assert.assertEquals(ttl, EphemeralType.getTTL(ephemeralOwner));
    }

    EphemeralType.validateTTL(CreateMode.PERSISTENT_WITH_TTL, 100);
    EphemeralType.validateTTL(CreateMode.PERSISTENT_SEQUENTIAL_WITH_TTL, 100);

    try {
        EphemeralType.validateTTL(CreateMode.EPHEMERAL, 100);
        Assert.fail("Should have thrown IllegalArgumentException");
    } catch (IllegalArgumentException dummy) {
        // expected
    }
}

/**
 * Test to verify that server is able to start with valid credentials
 */
@Test(timeout = 120000)
public void testValidCredentials() throws Exception {
    String serverPrincipal = hostServerPrincipal.substring(0, hostServerPrincipal.lastIndexOf("@"));
    Map<String, String> authConfigs = new HashMap<String, String>();
    authConfigs.put(QuorumAuth.QUORUM_SASL_AUTH_ENABLED, "true");
    authConfigs.put(QuorumAuth.QUORUM_SERVER_SASL_AUTH_REQUIRED, "true");
    authConfigs.put(QuorumAuth.QUORUM_LEARNER_SASL_AUTH_REQUIRED, "true");
    authConfigs.put(QuorumAuth.QUORUM_KERBEROS_SERVICE_PRINCIPAL, serverPrincipal);
    String connectStr = startQuorum(3, authConfigs, 3, true);
    CountdownWatcher watcher = new CountdownWatcher();
    ZooKeeper zk = new ZooKeeper(connectStr, ClientBase.CONNECTION_TIMEOUT, watcher);
    watcher.waitForConnected(ClientBase.CONNECTION_TIMEOUT);
    for (int i = 0; i < 10; i++) {
        zk.create("/" + i, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }
    zk.close();
}

@Test
public void testChRootCreateDelete() throws Exception {
    // creating the subtree for chRoot clients.
    String chRoot = createNameSpace();
    // Creating child using chRoot client.
    zk_chroot = createClient(this.hostPort + chRoot);
    Op createChild = Op.create("/myid", new byte[0],
            Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    multi(zk_chroot, Arrays.asList(createChild));

    Assert.assertNotNull("zNode is not created under chroot:" + chRoot, zk
            .exists(chRoot + "/myid", false));
    Assert.assertNotNull("zNode is not created under chroot:" + chRoot,
            zk_chroot.exists("/myid", false));
    Assert.assertNull("zNode is created directly under '/', ignored configured chroot",
            zk.exists("/myid", false));

    // Deleting child using chRoot client.
    Op deleteChild = Op.delete("/myid", 0);
    multi(zk_chroot, Arrays.asList(deleteChild));
    Assert.assertNull("zNode exists under chroot:" + chRoot, zk.exists(
            chRoot + "/myid", false));
    Assert.assertNull("zNode exists under chroot:" + chRoot, zk_chroot
            .exists("/myid", false));
}

public void addPartner(String key, Partner partner) {
    String partnerPath = ZKPaths.makePath(partnerStorePath, key);
    RetryRunner.create().onFinalError(e -> {
        LOGGER.error("addPartner.error", e);
        ReporterHolder.incException(e);
        throw new ServiceErrorException(ErrorCode.SYSTEM_ERROR);
    }).run((Callable<Void>) () -> {
        if (client.checkExists().creatingParentsIfNeeded().forPath(partnerPath) != null) {
            client.setData()
                  .forPath(partnerPath, JSONObject.toJSONBytes(partner));
        } else {
            client.create()
                  .creatingParentsIfNeeded()
                  .withMode(CreateMode.PERSISTENT)
                  .forPath(partnerPath, JSONObject.toJSONBytes(partner));
        }
        return null;
    });
}

/**
 * 创建任务项，注意其中的 CurrentSever和RequestServer不会起作用
 * @param taskItems
 * @throws Exception
 */
public void createScheduleTaskItem(ScheduleTaskItem[] taskItems) throws Exception {
    for (ScheduleTaskItem taskItem : taskItems){
       String zkPath = this.PATH_BaseTaskType + "/" + taskItem.getBaseTaskType() + "/" + taskItem.getTaskType() +"/" + this.PATH_TaskItem;
       if(this.getZooKeeper().exists(zkPath, false)== null){
           ZKTools.createPath(this.getZooKeeper(), zkPath, CreateMode.PERSISTENT, this.zkManager.getAcl());
       }
       String zkTaskItemPath = zkPath + "/" + taskItem.getTaskItem();
       this.getZooKeeper().create(zkTaskItemPath,null, this.zkManager.getAcl(),CreateMode.PERSISTENT);
       this.getZooKeeper().create(zkTaskItemPath + "/cur_server",null, this.zkManager.getAcl(),CreateMode.PERSISTENT);
       this.getZooKeeper().create(zkTaskItemPath + "/req_server",null, this.zkManager.getAcl(),CreateMode.PERSISTENT);
       this.getZooKeeper().create(zkTaskItemPath + "/sts",taskItem.getSts().toString().getBytes(), this.zkManager.getAcl(),CreateMode.PERSISTENT);
       this.getZooKeeper().create(zkTaskItemPath + "/parameter",taskItem.getDealParameter().getBytes(), this.zkManager.getAcl(),CreateMode.PERSISTENT);
       this.getZooKeeper().create(zkTaskItemPath + "/deal_desc",taskItem.getDealDesc().getBytes(), this.zkManager.getAcl(),CreateMode.PERSISTENT);
    }
}

/**
 * Tests a situation when client firstly connects to a read-only server and
 * then connects to a majority server. Transition should be transparent for
 * the user.
 */
@Test(timeout = 90000)
public void testSessionEstablishment() throws Exception {
    qu.shutdown(2);

    CountdownWatcher watcher = new CountdownWatcher();
    ZooKeeper zk = new ZooKeeper(qu.getConnString(), CONNECTION_TIMEOUT,
            watcher, true);
    watcher.waitForConnected(CONNECTION_TIMEOUT);
    Assert.assertSame("should be in r/o mode", States.CONNECTEDREADONLY, zk
            .getState());
    long fakeId = zk.getSessionId();

    watcher.reset();
    qu.start(2);
    Assert.assertTrue("waiting for server up", ClientBase.waitForServerUp(
            "127.0.0.1:" + qu.getPeer(2).clientPort, CONNECTION_TIMEOUT));
    watcher.waitForConnected(CONNECTION_TIMEOUT);
    zk.create("/test", "test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT);
    Assert.assertFalse("fake session and real session have same id", zk
            .getSessionId() == fakeId);

    zk.close();
}

/**
 * Write the "ActiveBreadCrumb" node, indicating that this node may need
 * to be fenced on failover.
 * @param oldBreadcrumbStat 
 */
private void writeBreadCrumbNode(Stat oldBreadcrumbStat)
    throws KeeperException, InterruptedException {
  Preconditions.checkState(appData != null, "no appdata");

  LOG.info("Writing znode " + zkBreadCrumbPath +
      " to indicate that the local node is the most recent active...");
  if (oldBreadcrumbStat == null) {
    // No previous active, just create the node
    createWithRetries(zkBreadCrumbPath, appData, zkAcl,
      CreateMode.PERSISTENT);
  } else {
    // There was a previous active, update the node
    setDataWithRetries(zkBreadCrumbPath, appData, oldBreadcrumbStat.getVersion());
  }
}

/**
 * Create a node.
 * 
 * @param path
 * @param data
 * @param mode
 * @return create node's path
 * @throws ZkInterruptedException if operation was interrupted, or a required reconnection got interrupted
 * @throws IllegalArgumentException if called from anything except the ZooKeeper event thread
 * @throws ZkException if any ZooKeeper exception occurred
 * @throws RuntimeException if any other exception occurs
 */
public String create(final String path, Object data, final CreateMode mode) throws ZkInterruptedException,
                                                                           IllegalArgumentException, ZkException,
                                                                           RuntimeException {
    if (path == null) {
        throw new NullPointerException("path must not be null.");
    }
    final byte[] bytes = data == null ? null : serialize(data);

    return retryUntilConnected(new Callable<String>() {

        @Override
        public String call() throws Exception {
            return _connection.create(path, bytes, mode);
        }
    });
}

@Override
public void set_data(String path,byte[] data) throws Exception
{
    if (data.length > Utils.SIZE_1_K * 800)
    {
        throw new Exception("Writing 800k+ data into ZK is not allowed!, data size is " + data.length);
    }
    if (zkobj.exists(zk,path,false))
    {
        zkobj.setData(zk,path,data);
    }
    else
    {
        zkobj.mkdirs(zk, PathUtils.parent_path(path));
        zkobj.createNode(zk,path,data,CreateMode.PERSISTENT);
    }
    if (zkCache != null)
        zkCache.put(path,data);
}

/**
 * verify that if create znode results in nodeexists and that znode is deleted
 * before exists() watch is set then the return of the exists() method results
 * in attempt to re-create the znode and become active
 */
@Test
public void testCreateNodeResultRetryNoNode() {
  elector.joinElection(data);

  elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
      ZK_LOCK_NAME);
  elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
      ZK_LOCK_NAME);
  elector.processResult(Code.NODEEXISTS.intValue(), ZK_LOCK_NAME, mockZK,
      ZK_LOCK_NAME);
  verifyExistCall(1);

  elector.processResult(Code.NONODE.intValue(), ZK_LOCK_NAME, mockZK,
      (Stat) null);
  Mockito.verify(mockApp, Mockito.times(1)).enterNeutralMode();
  Mockito.verify(mockZK, Mockito.times(4)).create(ZK_LOCK_NAME, data,
      Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
}

/**
 * Create the znodes, this may fail if the lower 32 roll over, if so
 * wait for the clients to be re-connected after the re-election
 */
private int createNodes(ZooKeeper zk, int start, int count) throws Exception {
    LOG.info("Creating nodes " + start + " thru " + (start + count));
    int j = 0;
    try {
        for (int i = start; i < start + count; i++) {
            zk.create("/foo" + i, new byte[0], Ids.READ_ACL_UNSAFE,
                    CreateMode.EPHEMERAL);
            j++;
        }
    } catch (ConnectionLossException e) {
        // this is ok - the leader has dropped leadership
        waitForClientsConnected();
    }
    return j;
}

/**
 * Write the "ActiveBreadCrumb" node, indicating that this node may need
 * to be fenced on failover.
 * @param oldBreadcrumbStat 
 */
private void writeBreadCrumbNode(Stat oldBreadcrumbStat)
    throws KeeperException, InterruptedException {
  Preconditions.checkState(appData != null, "no appdata");

  LOG.info("Writing znode " + zkBreadCrumbPath +
      " to indicate that the local node is the most recent active...");
  if (oldBreadcrumbStat == null) {
    // No previous active, just create the node
    createWithRetries(zkBreadCrumbPath, appData, zkAcl,
      CreateMode.PERSISTENT);
  } else {
    // There was a previous active, update the node
    setDataWithRetries(zkBreadCrumbPath, appData, oldBreadcrumbStat.getVersion());
  }
}

public ZkAbstractStore(CuratorFramework framework, PStoreConfig<V> config)
    throws IOException {
  this.parent = "/" + config.getName();
  this.prefix = parent + "/";
  this.framework = framework;
  this.config = config;

  // make sure the parent node exists.
  try {
    if (framework.checkExists().forPath(parent) == null) {
      framework.create().withMode(CreateMode.PERSISTENT).forPath(parent);
    }

    this.childrenCache = new PathChildrenCache(framework, parent, true);
    this.childrenCache.start(StartMode.BUILD_INITIAL_CACHE);

  } catch (Exception e) {
    throw new RuntimeException("Failure while accessing Zookeeper for PStore: " + e.getMessage(), e);
  }

}

@Override
public void setUp() throws Exception {
    String hp = hostPort;
    hostPort = hostPort + "/chrootclienttest";

    System.out.println(hostPort);
    super.setUp();

    LOG.info("STARTING " + getTestName());

    ZooKeeper zk = createClient(hp);
    try {
        zk.create("/chrootclienttest", null, Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);
    } finally {
        zk.close();
    }
}

/**
 * <p>
 * NONSEQUENTIAL create is idempotent operation.
 * Retry before throwing exceptions.
 * But this function will not throw the NodeExist exception back to the
 * application.
 * </p>
 * <p>
 * But SEQUENTIAL is NOT idempotent operation. It is necessary to add
 * identifier to the path to verify, whether the previous one is successful
 * or not.
 * </p>
 *
 * @return Path
 */
public String create(String path, byte[] data, List<ACL> acl,
    CreateMode createMode)
throws KeeperException, InterruptedException {
  TraceScope traceScope = null;
  try {
    traceScope = Trace.startSpan("RecoverableZookeeper.create");
    byte[] newData = appendMetaData(data);
    switch (createMode) {
      case EPHEMERAL:
      case PERSISTENT:
        return createNonSequential(path, newData, acl, createMode);

      case EPHEMERAL_SEQUENTIAL:
      case PERSISTENT_SEQUENTIAL:
        return createSequential(path, newData, acl, createMode);

      default:
        throw new IllegalArgumentException("Unrecognized CreateMode: " +
            createMode);
    }
  } finally {
    if (traceScope != null) traceScope.close();
  }
}

public void processResult(int rc, String path, Object ctx, byte[] data,
        Stat stat) {
    if (rc != 0) {
        zk.getData("/simpleCase", true, this, null);
    } else {
        this.data = data;
        String content = new String(data);
        if (content.equals("die")) {
            this.stop();
            return;
        }
        if (!createdEphemeral) {
            zk.create(myPath, data, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, this, null);
            createdEphemeral = true;
        } else {
            zk.setData(myPath, data, -1, this, null);
        }
    }            
}

/**
 * Test verifies the multi.create with invalid createModeFlag
 */
@Test(timeout = 90000)
public void testInvalidCreateModeFlag() throws Exception {
    List<Integer> expectedResultCodes = new ArrayList<Integer>();
    expectedResultCodes.add(KeeperException.Code.RUNTIMEINCONSISTENCY
                            .intValue());
    expectedResultCodes.add(KeeperException.Code.BADARGUMENTS.intValue());
    expectedResultCodes.add(KeeperException.Code.RUNTIMEINCONSISTENCY
                            .intValue());

    int createModeFlag = 6789;
    List<Op> opList = Arrays.asList(Op.create("/multi0", new byte[0],
                                              Ids.OPEN_ACL_UNSAFE,
                                              CreateMode.PERSISTENT),
                                    Op.create("/multi1", new byte[0],
                                              Ids.OPEN_ACL_UNSAFE,
                                              createModeFlag),
                                    Op.create("/multi2", new byte[0],
                                              Ids.OPEN_ACL_UNSAFE,
                                              CreateMode.PERSISTENT));
    String expectedErr = KeeperException.Code.BADARGUMENTS.name();
    multiHavingErrors(zk, opList, expectedResultCodes, expectedErr);
}

public boolean createNode(String parent, String nodeName) {
    if (connected) {
        try {
            String[] nodeElements = nodeName.split("/");
            for (String nodeElement : nodeElements) {
                String node = parent + "/" + nodeElement;
                Stat s = zooKeeper.exists(node, false);
                if (s == null) {
                    zooKeeper.create(node, this.encryptionManager
                            .encryptData(null), Ids.OPEN_ACL_UNSAFE,
                            CreateMode.PERSISTENT);
                    parent = node;
                }
            }
            return true;
        } catch (Exception e) {
            LoggerFactory.getLogger().error(
                    "Error occurred creating node: " + parent + "/"
                            + nodeName, e);
        }
    }
    return false;
}

/**
 * @param path
 * @return
 */
private static Stat createWithParent(String path){
    if (null == path || path.trim().length()==0) {
        return null;
    }

    try {
        Stat stat = getInstance().exists(path, true);
        if(null == stat) {
            if(path.lastIndexOf("/") > 0) {
                String parentPath = path.substring(0,path.lastIndexOf("/"));
                Stat parentStat = getInstance().exists(parentPath, true);
                if (null == parentStat) {
                    createWithParent(parentPath);
                }
            }
            //添加节点
            zooKeeper.create(path, new byte[]{}, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        return getInstance().exists(path, true);
    } catch (Exception e) {
        e.printStackTrace();
    }
    return null;
}

/**
 * 更改恢复次数
 *
 * @param id              事务id
 * @param retry           恢复次数
 * @param applicationName 应用名称
 * @return true 成功
 */
@Override
public Boolean updateRetry(String id, Integer retry, String applicationName) {
    if (StringUtils.isBlank(id) || StringUtils.isBlank(applicationName) || Objects.isNull(retry)) {
        return Boolean.FALSE;
    }
    final String rootPathPrefix = RepositoryPathUtils.buildZookeeperPathPrefix(applicationName);
    final String path = RepositoryPathUtils.buildZookeeperRootPath(rootPathPrefix, id);
    try {
        byte[] content = zooKeeper.getData(path,
                false, new Stat());
        final CoordinatorRepositoryAdapter adapter = objectSerializer.deSerialize(content, CoordinatorRepositoryAdapter.class);
        adapter.setLastTime(DateUtils.getDateYYYY());
        adapter.setRetriedCount(retry);
        zooKeeper.create(path,
                objectSerializer.serialize(adapter),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        return Boolean.TRUE;
    } catch (Exception e) {
        e.printStackTrace();
    }

    return Boolean.FALSE;
}

@Test
public void testNodeChildrenChanged() throws Exception {
    QuorumUtil qu = new QuorumUtil(1);
    qu.startAll();

    EventsWatcher watcher = new EventsWatcher();
    ZooKeeper zk1 = createClient(qu, 1, watcher);
    ZooKeeper zk2 = createClient(qu, 2);

    String path = "/test-children-changed";

    zk1.create(path, new byte[1], ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT);
    zk1.getChildren(path, watcher);
    qu.shutdown(1);
    zk2.create(path + "/children-1", new byte[2],
            ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    qu.start(1);
    watcher.waitForConnected(TIMEOUT * 1000L);
    watcher.assertEvent(TIMEOUT, EventType.NodeChildrenChanged);

    qu.shutdownAll();
}

@Test
public void testChRootSetData() throws Exception {
    // creating the subtree for chRoot clients.
    String chRoot = createNameSpace();
    // setData using chRoot client.
    zk_chroot = createClient(this.hostPort + chRoot);
    String[] names = {"/multi0", "/multi1", "/multi2"};
    List<Op> ops = new ArrayList<Op>();

    for (int i = 0; i < names.length; i++) {
        ops.add(Op.create(names[i], new byte[0], Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT));
        ops.add(Op.setData(names[i], names[i].getBytes(), 0));
    }

    multi(zk_chroot, ops) ;

    for (int i = 0; i < names.length; i++) {
        Assert.assertArrayEquals("zNode data not matching", names[i]
                .getBytes(), zk_chroot.getData(names[i], false, null));
    }
}

@Ignore("DLR is broken by HBASE-12751") @Test(timeout=60000)
public void testGetPreviousRecoveryMode() throws Exception {
  LOG.info("testGetPreviousRecoveryMode");
  SplitLogCounters.resetCounters();
  // Not actually enabling DLR for the cluster, just for the ZkCoordinatedStateManager to use.
  // The test is just manipulating ZK manually anyways.
  conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);

  zkw.getRecoverableZooKeeper().create(ZKSplitLog.getEncodedNodeName(zkw, "testRecovery"),
    new SplitLogTask.Unassigned(
      ServerName.valueOf("mgr,1,1"), RecoveryMode.LOG_SPLITTING).toByteArray(),
      Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

  slm = new SplitLogManager(ds, conf, stopper, master, DUMMY_MASTER);
  LOG.info("Mode1=" + slm.getRecoveryMode());
  assertTrue(slm.isLogSplitting());
  zkw.getRecoverableZooKeeper().delete(ZKSplitLog.getEncodedNodeName(zkw, "testRecovery"), -1);
  LOG.info("Mode2=" + slm.getRecoveryMode());
  slm.setRecoveryMode(false);
  LOG.info("Mode3=" + slm.getRecoveryMode());
  assertTrue("Mode4=" + slm.getRecoveryMode(), slm.isLogReplaying());
}

@Override
public void unRegisterScheduleServer(ScheduleServer server) {
    List<String> serverList = this.loadScheduleServerNames();
    try {
        if (server.isRegister() && this.isLeader(server.getUuid(), serverList)) {
            //delete task
            String zkPath = this.pathTask;
            String serverPath = this.pathServer;
            if (this.zkManager.getZooKeeper().exists(zkPath, false) == null) {
                this.zkManager.getZooKeeper().create(zkPath, null, this.zkManager.getAcl(), CreateMode.PERSISTENT);
            }
            //get all task
            List<String> children = this.zkManager.getZooKeeper().getChildren(zkPath, false);
            if (null != children && children.size() > 0) {
                for (String taskName : children) {
                    String taskPath = zkPath + "/" + taskName;
                    if (this.zkManager.getZooKeeper().exists(taskPath, false) != null) {
                        ZKTools.deleteTree(this.zkManager.getZooKeeper(), taskPath + "/" + server.getUuid());
                    }
                }
            }
            //删除
            if (this.zkManager.getZooKeeper().exists(this.pathServer, false) == null) {
                ZKTools.deleteTree(this.zkManager.getZooKeeper(), serverPath + serverPath + "/" + server.getUuid());
            }
            server.setRegister(false);
        }
    } catch (Exception e) {
        LOG.error("unRegisterScheduleServer failed", e);
    }
}

/**
 * Create an instance
 */
protected void createCuratorService() throws IOException {
  curatorService = new CuratorService("curatorService");
  curatorService.init(createRegistryConfiguration());
  curatorService.start();
  rootACL = RegistrySecurity.WorldReadWriteACL;
  curatorService.maybeCreate("", CreateMode.PERSISTENT, rootACL, true);
}

@Override
public void createNodeInZK(String key, V value) {
  try {
    framework.create().withMode(CreateMode.PERSISTENT).forPath(p(key), config.getSerializer().serialize(value));
  } catch (Exception e) {
    throw new RuntimeException("Failure while accessing Zookeeper", e);
  }
}

@Test
public void testMaybeCreate() throws Throwable {
  assertTrue(curatorService.maybeCreate("/p3", CreateMode.PERSISTENT,
      RegistrySecurity.WorldReadWriteACL, false));
  assertFalse(curatorService.maybeCreate("/p3", CreateMode.PERSISTENT,
      RegistrySecurity.WorldReadWriteACL, false));
}

public void updateConsumerOffset(String consumerGroup, String topic, Integer queueId, long offset) {
    String path = zkConf.getZKBasePath() + "/consumers/" + consumerGroup + "/offsets/" + topic + "/" + queueId;
    int maxTryCount = 2;
    int currentTryCount = 0;
    while (currentTryCount++ < maxTryCount) {
        try {
            // 更新zk
            byte[] dataBytes = String.valueOf(offset).getBytes();
            zkClient.setData().forPath(path, dataBytes);
            LOG.info("updateConsumerOffset success, consumerGroup={}, topic={}, queue={}, offset={}",
                    consumerGroup, topic, queueId, offset);

            // 更新本地内存
            metadata.getConsumerOffsetLock().lock();
            try {
                metadata.getConsumerOffsetMap().put(queueId, offset);
                LOG.info("new consumer offset map={}",
                        JsonUtil.toJson(metadata.getConsumerOffsetMap()));
            } finally {
                metadata.getConsumerOffsetLock().unlock();
            }
            break;
        } catch (KeeperException.NoNodeException ex1) {
            createPath(path, CreateMode.PERSISTENT);
            continue;
        } catch (Exception ex2) {
            LOG.warn("updateConsumerOffset exception:", ex2);
            break;
        }
    }
}

protected static String createBaseZNode() throws Exception {
    ZooKeeper zk = new ZooKeeper(ZKHOSTPORT, 30000, new MyWatcher());

    String baseZnode = zk.create("/test-", null, Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT_SEQUENTIAL);
    zk.close();

    return baseZnode;
}

private void verifyCreateFails(String path, ZooKeeper zk) throws Exception {
    try {
        zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    } catch (IllegalArgumentException e) {
        // this is good
        return;
    }
    Assert.fail("bad path \"" + path + "\" not caught");
}

/**
 * When we create ephemeral node, we need to check against global
 * session, so the leader never accept request from an expired session
 * (that we no longer track)
 *
 * This is not the same as SessionInvalidationTest since session
 * is not in closing state
 */
public void testCreateEphemeral(boolean localSessionEnabled) throws Exception {
    if (localSessionEnabled) {
        qu.enableLocalSession(true);
    }
    qu.startAll();
    QuorumPeer leader = qu.getLeaderQuorumPeer();

    ZooKeeper zk = ClientBase.createZKClient(qu.getConnectString(leader));

    CreateRequest createRequest = new CreateRequest("/impossible",
            new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL.toFlag());
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
    createRequest.serialize(boa, "request");
    ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());

    // Mimic sessionId generated by follower's local session tracker
    long sid = qu.getFollowerQuorumPeers().get(0).getActiveServer()
            .getServerId();
    long fakeSessionId = (sid << 56) + 1;

    LOG.info("Fake session Id: " + Long.toHexString(fakeSessionId));

    Request request = new Request(null, fakeSessionId, 0, OpCode.create,
            bb, new ArrayList<Id>());

    // Submit request directly to leader
    leader.getActiveServer().submitRequest(request);

    // Make sure that previous request is finished
    zk.create("/ok", new byte[0], Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT);

    Stat stat = zk.exists("/impossible", null);
    Assert.assertEquals("Node from fake session get created", null, stat);

}

/**
 * Test write operations using multi request.
 */
@Test(timeout = 90000)
public void testMultiTransaction() throws Exception {
    CountdownWatcher watcher = new CountdownWatcher();
    ZooKeeper zk = new ZooKeeper(qu.getConnString(), CONNECTION_TIMEOUT,
            watcher, true);
    watcher.waitForConnected(CONNECTION_TIMEOUT); // ensure zk got connected

    final String data = "Data to be read in RO mode";
    final String node1 = "/tnode1";
    final String node2 = "/tnode2";
    zk.create(node1, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT);

    watcher.reset();
    qu.shutdown(2);
    watcher.waitForConnected(CONNECTION_TIMEOUT);
    Assert.assertEquals("Should be in r-o mode", States.CONNECTEDREADONLY,
            zk.getState());

    // read operation during r/o mode
    String remoteData = new String(zk.getData(node1, false, null));
    Assert.assertEquals("Failed to read data in r-o mode", data, remoteData);

    try {
        Transaction transaction = zk.transaction();
        transaction.setData(node1, "no way".getBytes(), -1);
        transaction.create(node2, data.getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        transaction.commit();
        Assert.fail("Write operation using multi-transaction"
                + " api has succeeded during RO mode");
    } catch (NotReadOnlyException e) {
        // ok
    }

    Assert.assertNull("Should have created the znode:" + node2,
            zk.exists(node2, false));
}

/**
 * test that ZK can write as itself
 * @throws Throwable
 */
@Test
public void testZookeeperCanWrite() throws Throwable {

  System.setProperty("curator-log-events", "true");
  startSecureZK();
  CuratorService curator = null;
  LoginContext login = login(ZOOKEEPER_LOCALHOST,
      ZOOKEEPER_CLIENT_CONTEXT,
      keytab_zk);
  try {
    logLoginDetails(ZOOKEEPER, login);
    RegistrySecurity.setZKSaslClientProperties(ZOOKEEPER,
                                              ZOOKEEPER_CLIENT_CONTEXT);
    curator = startCuratorServiceInstance("ZK", true);
    LOG.info(curator.toString());

    addToTeardown(curator);
    curator.zkMkPath("/", CreateMode.PERSISTENT, false,
        RegistrySecurity.WorldReadWriteACL);
    curator.zkList("/");
    curator.zkMkPath("/zookeeper", CreateMode.PERSISTENT, false,
        RegistrySecurity.WorldReadWriteACL);
  } finally {
    logout(login);
    ServiceOperations.stop(curator);
  }
}

private void addOrUpdateToken(TokenIdent ident,
    DelegationTokenInformation info, boolean isUpdate) throws Exception {
  String nodeCreatePath =
      getNodePath(ZK_DTSM_TOKENS_ROOT, DELEGATION_TOKEN_PREFIX
          + ident.getSequenceNumber());
  ByteArrayOutputStream tokenOs = new ByteArrayOutputStream();
  DataOutputStream tokenOut = new DataOutputStream(tokenOs);
  ByteArrayOutputStream seqOs = new ByteArrayOutputStream();

  try {
    ident.write(tokenOut);
    tokenOut.writeLong(info.getRenewDate());
    tokenOut.writeInt(info.getPassword().length);
    tokenOut.write(info.getPassword());
    if (LOG.isDebugEnabled()) {
      LOG.debug((isUpdate ? "Updating " : "Storing ")
          + "ZKDTSMDelegationToken_" +
          ident.getSequenceNumber());
    }
    if (isUpdate) {
      zkClient.setData().forPath(nodeCreatePath, tokenOs.toByteArray())
          .setVersion(-1);
    } else {
      zkClient.create().withMode(CreateMode.PERSISTENT)
          .forPath(nodeCreatePath, tokenOs.toByteArray());
    }
  } finally {
    seqOs.close();
  }
}

/**
 * Utility function to ensure that the configured base znode exists.
 * This recursively creates the znode as well as all of its parents.
 */
public synchronized void ensureParentZNode()
    throws IOException, InterruptedException {
  Preconditions.checkState(!wantToBeInElection,
      "ensureParentZNode() may not be called while in the election");

  String pathParts[] = znodeWorkingDir.split("/");
  Preconditions.checkArgument(pathParts.length >= 1 &&
      pathParts[0].isEmpty(),
      "Invalid path: %s", znodeWorkingDir);

  StringBuilder sb = new StringBuilder();
  for (int i = 1; i < pathParts.length; i++) {
    sb.append("/").append(pathParts[i]);
    String prefixPath = sb.toString();
    LOG.debug("Ensuring existence of " + prefixPath);
    try {
      createWithRetries(prefixPath, new byte[]{}, zkAcl, CreateMode.PERSISTENT);
    } catch (KeeperException e) {
      if (isNodeExists(e.code())) {
        // This is OK - just ensuring existence.
        continue;
      } else {
        throw new IOException("Couldn't create " + prefixPath, e);
      }
    }
  }

  LOG.info("Successfully created " + znodeWorkingDir + " in ZK.");
}

/**
 * verify that successful znode create result becomes active and monitoring is
 * started
 */
@Test
public void testCreateNodeResultBecomeActive() throws Exception {
  mockNoPriorActive();

  elector.joinElection(data);
  elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK,
      ZK_LOCK_NAME);
  Mockito.verify(mockApp, Mockito.times(1)).becomeActive();
  verifyExistCall(1);

  // monitor callback verifies the leader is ephemeral owner of lock but does
  // not call becomeActive since its already active
  Stat stat = new Stat();
  stat.setEphemeralOwner(1L);
  Mockito.when(mockZK.getSessionId()).thenReturn(1L);
  elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK, stat);
  // should not call neutral mode/standby/active
  Mockito.verify(mockApp, Mockito.times(0)).enterNeutralMode();
  Mockito.verify(mockApp, Mockito.times(0)).becomeStandby();
  Mockito.verify(mockApp, Mockito.times(1)).becomeActive();
  // another joinElection not called.
  Mockito.verify(mockZK, Mockito.times(1)).create(ZK_LOCK_NAME, data,
      Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
  // no new monitor called
  verifyExistCall(1);
}

/**
 * Verify that, when the callback fails to enter active state, after
 * a ZK disconnect (i.e from the StatCallback), that the elector rejoins
 * the election after sleeping for a short period.
 */
@Test
public void testFailToBecomeActiveAfterZKDisconnect() throws Exception {
  mockNoPriorActive();
  elector.joinElection(data);
  Assert.assertEquals(0, elector.sleptFor);

  elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
      ZK_LOCK_NAME);
  Mockito.verify(mockZK, Mockito.times(2)).create(ZK_LOCK_NAME, data,
      Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);

  elector.processResult(Code.NODEEXISTS.intValue(), ZK_LOCK_NAME, mockZK,
      ZK_LOCK_NAME);
  verifyExistCall(1);

  Stat stat = new Stat();
  stat.setEphemeralOwner(1L);
  Mockito.when(mockZK.getSessionId()).thenReturn(1L);

  // Fake failure to become active from within the stat callback
  Mockito.doThrow(new ServiceFailedException("fail to become active"))
      .when(mockApp).becomeActive();
  elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK, stat);
  Mockito.verify(mockApp, Mockito.times(1)).becomeActive();

  // should re-join
  Mockito.verify(mockZK, Mockito.times(3)).create(ZK_LOCK_NAME, data,
      Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
  Assert.assertEquals(2, count);
  Assert.assertTrue(elector.sleptFor > 0);
}

@Test
public void testChild()
    throws IOException, KeeperException, InterruptedException
{
    String name = "/foo";
    zk.create(name, name.getBytes(), Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT);

    String childname = name + "/bar";
    zk.create(childname, childname.getBytes(), Ids.OPEN_ACL_UNSAFE,
            CreateMode.EPHEMERAL);

    Stat stat = new Stat();
    List<String> s = zk.getChildren(name, false, stat);

    Assert.assertEquals(stat.getCzxid(), stat.getMzxid());
    Assert.assertEquals(stat.getCzxid() + 1, stat.getPzxid());
    Assert.assertEquals(stat.getCtime(), stat.getMtime());
    Assert.assertEquals(1, stat.getCversion());
    Assert.assertEquals(0, stat.getVersion());
    Assert.assertEquals(0, stat.getAversion());
    Assert.assertEquals(0, stat.getEphemeralOwner());
    Assert.assertEquals(name.length(), stat.getDataLength());
    Assert.assertEquals(1, stat.getNumChildren());
    Assert.assertEquals(s.size(), stat.getNumChildren());

    s = zk.getChildren(childname, false, stat);

    Assert.assertEquals(stat.getCzxid(), stat.getMzxid());
    Assert.assertEquals(stat.getCzxid(), stat.getPzxid());
    Assert.assertEquals(stat.getCtime(), stat.getMtime());
    Assert.assertEquals(0, stat.getCversion());
    Assert.assertEquals(0, stat.getVersion());
    Assert.assertEquals(0, stat.getAversion());
    Assert.assertEquals(zk.getSessionId(), stat.getEphemeralOwner());
    Assert.assertEquals(childname.length(), stat.getDataLength());
    Assert.assertEquals(0, stat.getNumChildren());
    Assert.assertEquals(s.size(), stat.getNumChildren());
}