import org.apache.hadoop.registry.client.api.BindFlags; //导入依赖的package包/类
@Test
public void testChildDeletion() throws Throwable {
  ServiceRecord app = createRecord("app1",
      PersistencePolicies.APPLICATION, "app",
      null);
  ServiceRecord container = createRecord("container1",
      PersistencePolicies.CONTAINER, "container",
      null);

  operations.bind("/app", app, BindFlags.OVERWRITE);
  operations.bind("/app/container", container, BindFlags.OVERWRITE);

  try {
    int p = purge("/",
        "app1",
        PersistencePolicies.APPLICATION,
        RegistryAdminService.PurgePolicy.FailOnChildren);
    fail("expected a failure, got a purge count of " + p);
  } catch (PathIsNotEmptyDirectoryException expected) {
    // expected
  }

}
 

import org.apache.hadoop.registry.client.api.BindFlags; //导入依赖的package包/类
@Test
public void testOverwrite() throws Throwable {
  ServiceRecord written = putExampleServiceEntry(ENTRY_PATH, 0);
  ServiceRecord resolved1 = operations.resolve(ENTRY_PATH);
  resolved1.description = "resolved1";
  try {
    operations.bind(ENTRY_PATH, resolved1, 0);
    fail("overwrite succeeded when it should have failed");
  } catch (FileAlreadyExistsException expected) {
    // expected
  }

  // verify there's no changed
  ServiceRecord resolved2 = operations.resolve(ENTRY_PATH);
  assertMatches(written, resolved2);
  operations.bind(ENTRY_PATH, resolved1, BindFlags.OVERWRITE);
  ServiceRecord resolved3 = operations.resolve(ENTRY_PATH);
  assertMatches(resolved1, resolved3);
}
 

import org.apache.hadoop.registry.client.api.BindFlags; //导入依赖的package包/类
public boolean needUpgrade() {
    String containerPath = RegistryUtils.componentPath(
            JOYConstants.APP_TYPE, this.executorMeta.getInstanceName(),
            this.executorMeta.getApplicationId(), this.executorMeta.getRunningContainer());

    try {
        if (registryOperations.exists(containerPath)) {
            ServiceRecord sr = registryOperations.resolve(containerPath);
            if (sr.get(JOYConstants.NEED_UPGRADE) != null && sr.get(JOYConstants.NEED_UPGRADE).equals(JOYConstants.TRUE)) {
                sr.set(JOYConstants.NEED_UPGRADE, JOYConstants.FALSE);
                registryOperations.bind(containerPath, sr, BindFlags.OVERWRITE);
                LOG.info(JOYConstants.NEED_UPGRADE);
                return true;
            }
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
    return false;
}
 

import org.apache.hadoop.registry.client.api.BindFlags; //导入依赖的package包/类
@Override
public void bind(String path,
    ServiceRecord record,
    int flags) throws IOException {
  Preconditions.checkArgument(record != null, "null record");
  validatePath(path);
  // validate the record before putting it
  RegistryTypeUtils.validateServiceRecord(path, record);
  LOG.info("Bound at {} : {}", path, record);

  CreateMode mode = CreateMode.PERSISTENT;
  byte[] bytes = serviceRecordMarshal.toBytes(record);
  zkSet(path, mode, bytes, getClientAcls(),
      ((flags & BindFlags.OVERWRITE) != 0));
}
 

import org.apache.hadoop.registry.client.api.BindFlags; //导入依赖的package包/类
@Test
public void testPutGetContainerPersistenceServiceEntry() throws Throwable {

  String path = ENTRY_PATH;
  ServiceRecord written = buildExampleServiceEntry(
      PersistencePolicies.CONTAINER);

  operations.mknode(RegistryPathUtils.parentOf(path), true);
  operations.bind(path, written, BindFlags.CREATE);
  ServiceRecord resolved = operations.resolve(path);
  validateEntry(resolved);
  assertMatches(written, resolved);
}
 

import org.apache.hadoop.registry.client.api.BindFlags; //导入依赖的package包/类
@Test
public void testPutMinimalRecord() throws Throwable {
  String path = "/path/with/minimal";
  operations.mknode(path, true);
  ServiceRecord record = new ServiceRecord();
  operations.bind(path, record, BindFlags.OVERWRITE);
  ServiceRecord resolve = operations.resolve(path);
  assertMatches(record, resolve);

}
 

import org.apache.hadoop.registry.client.api.BindFlags; //导入依赖的package包/类
/**
 * Add a component 
 * @param serviceClass service class to use under ~user
 * @param componentName component name
 * @param record record to put
 * @throws IOException
 */
public void putComponent(String serviceClass,
    String serviceName,
    String componentName,
    ServiceRecord record) throws IOException {
  String path = RegistryUtils.componentPath(
      user, serviceClass, serviceName, componentName);
  registryOperations.mknode(RegistryPathUtils.parentOf(path), true);
  registryOperations.bind(path, record, BindFlags.OVERWRITE);
}
 

import org.apache.hadoop.registry.client.api.BindFlags; //导入依赖的package包/类
/**
 * Add a service under a path, optionally purging any history
 * @param username user
 * @param serviceClass service class to use under ~user
 * @param serviceName name of the service
 * @param record service record
 * @param deleteTreeFirst perform recursive delete of the path first.
 * @return the path the service was created at
 * @throws IOException
 */
public String putService(String username,
    String serviceClass,
    String serviceName,
    ServiceRecord record,
    boolean deleteTreeFirst) throws IOException {
  String path = RegistryUtils.servicePath(
      username, serviceClass, serviceName);
  if (deleteTreeFirst) {
    registryOperations.delete(path, true);
  }
  registryOperations.mknode(RegistryPathUtils.parentOf(path), true);
  registryOperations.bind(path, record, BindFlags.OVERWRITE);
  return path;
}
 

import org.apache.hadoop.registry.client.api.BindFlags; //导入依赖的package包/类
/**
 * see if anyone is updating host's port list, if not start , update this host itself
 * timeout is 45 seconds
 *
 * @param hostPath
 * @throws InterruptedException
 * @throws IOException
 */
private void tryHostLock(String hostPath) throws Exception {

    //if path has created 60 seconds ago, then delete
    if (registryOperations.exists(hostPath)) {
        try {
            ServiceRecord host = registryOperations.resolve(hostPath);
            Long cTime = Long.parseLong(host.get(JOYConstants.CTIME, JOYConstants.DEFAULT_CTIME));
            Date now = new Date();
            if (now.getTime() - cTime > JOYConstants.HOST_LOCK_TIMEOUT || cTime > now.getTime())
                registryOperations.delete(hostPath, true);
        } catch (Exception ex) {
            LOG.error(ex);
        }
    }

    int failedCount = JOYConstants.RETRY_TIMES;
    while (!registryOperations.mknode(hostPath, true)) {
        Thread.sleep(JOYConstants.SLEEP_INTERVAL);
        failedCount--;
        if (failedCount <= 0)
            break;
    }

    if (failedCount > 0) {
        ServiceRecord sr = new ServiceRecord();
        Date date = new Date();
        date.getTime();
        sr.set(JOYConstants.CTIME, String.valueOf(date.getTime()));
        registryOperations.bind(hostPath, sr, BindFlags.OVERWRITE);
        return;
    }
    throw new Exception("can't get host lock");
}
 

import org.apache.hadoop.registry.client.api.BindFlags; //导入依赖的package包/类
/**
     * see if anyone is updating host's port list, if not start , update this host itself
     * timeout is 45 seconds
     *
     * @throws InterruptedException
     * @throws IOException
     */
    private void tryHostLock() throws Exception {

        String hostPath = getHostPath();
        //if path has created 60 seconds ago, then delete
        if (registryOperations.exists(hostPath)) {
            try {
                ServiceRecord host = registryOperations.resolve(hostPath);
                Long cTime = Long.parseLong(host.get("cTime", "0"));
                Date now = new Date();
                if (now.getTime() - cTime > 60 * 1000 || cTime > now.getTime())
                    registryOperations.delete(hostPath, true);
            } catch (Exception ex) {
                LOG.error(ex);
//                registryOperations.delete(hostPath, true);
            }
        }

        int failedCount = 45;
        while (!registryOperations.mknode(hostPath, true)) {
            Thread.sleep(1000);
            failedCount--;
            if (failedCount <= 0)
                break;
        }

        if (failedCount > 0) {
            ServiceRecord sr = new ServiceRecord();
            Date date = new Date();
            date.getTime();
            sr.set("cTime", String.valueOf(date.getTime()));
            registryOperations.bind(hostPath, sr, BindFlags.OVERWRITE);
            return;
        }


        throw new Exception("can't get host lock");
    }
 

import org.apache.hadoop.registry.client.api.BindFlags; //导入依赖的package包/类
private void putService() throws IOException {
  String path = RegistryUtils.servicePath(user, serviceClass, appIdAsName);
  registryOperations.mknode(RegistryPathUtils.parentOf(path), true);
  registryOperations.bind(path, appRecord, BindFlags.OVERWRITE);
}
 

import org.apache.hadoop.registry.client.api.BindFlags; //导入依赖的package包/类
private void putComponent(String containerIdAsName) throws IOException {
  String path = RegistryUtils.componentPath(user, serviceClass, appIdAsName, containerIdAsName);
  registryOperations.mknode(RegistryPathUtils.parentOf(path), true);
  registryOperations.bind(path, containerRecords.get(containerIdAsName), BindFlags.OVERWRITE);
}