import org.apache.hadoop.registry.client.types.ServiceRecord; //导入依赖的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.types.ServiceRecord; //导入依赖的package包/类
@Test
public void testLsParent() throws Throwable {
  ServiceRecord written = putExampleServiceEntry(ENTRY_PATH, 0);
  RegistryPathStatus stat = operations.stat(ENTRY_PATH);

  List<String> children = operations.list(PARENT_PATH);
  assertEquals(1, children.size());
  assertEquals(NAME, children.get(0));
  Map<String, RegistryPathStatus> childStats =
      RegistryUtils.statChildren(operations, PARENT_PATH);
  assertEquals(1, childStats.size());
  assertEquals(stat, childStats.get(NAME));

  Map<String, ServiceRecord> records =
      RegistryUtils.extractServiceRecords(operations,
          PARENT_PATH,
          childStats.values());
  assertEquals(1, records.size());
  ServiceRecord record = records.get(ENTRY_PATH);
  RegistryTypeUtils.validateServiceRecord(ENTRY_PATH, record);
  assertMatches(written, record);
}
 

import org.apache.hadoop.registry.client.types.ServiceRecord; //导入依赖的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.types.ServiceRecord; //导入依赖的package包/类
/**
 * Assert the records match.
 * @param source record that was written
 * @param resolved the one that resolved.
 */
public static void assertMatches(ServiceRecord source, ServiceRecord resolved) {
  assertNotNull("Null source record ", source);
  assertNotNull("Null resolved record ", resolved);
  assertEquals(source.description, resolved.description);

  Map<String, String> srcAttrs = source.attributes();
  Map<String, String> resolvedAttrs = resolved.attributes();
  String sourceAsString = source.toString();
  String resolvedAsString = resolved.toString();
  assertEquals("Wrong count of attrs in \n" + sourceAsString
               + "\nfrom\n" + resolvedAsString,
      srcAttrs.size(),
      resolvedAttrs.size());
  for (Map.Entry<String, String> entry : srcAttrs.entrySet()) {
    String attr = entry.getKey();
    assertEquals("attribute "+ attr, entry.getValue(), resolved.get(attr));
  }
  assertEquals("wrong external endpoint count",
      source.external.size(), resolved.external.size());
  assertEquals("wrong external endpoint count",
      source.internal.size(), resolved.internal.size());
}
 

import org.apache.hadoop.registry.client.types.ServiceRecord; //导入依赖的package包/类
/**
 * Find an endpoint in a record or fail,
 * @param record record
 * @param api API
 * @param external external?
 * @param addressElements expected # of address elements?
 * @param addressTupleSize expected size of a type
 * @return the endpoint.
 */
public static Endpoint findEndpoint(ServiceRecord record,
    String api, boolean external, int addressElements, int addressTupleSize) {
  Endpoint epr = external ? record.getExternalEndpoint(api)
                          : record.getInternalEndpoint(api);
  if (epr != null) {
    assertEquals("wrong # of addresses",
        addressElements, epr.addresses.size());
    assertEquals("wrong # of elements in an address tuple",
        addressTupleSize, epr.addresses.get(0).size());
    return epr;
  }
  List<Endpoint> endpoints = external ? record.external : record.internal;
  StringBuilder builder = new StringBuilder();
  for (Endpoint endpoint : endpoints) {
    builder.append("\"").append(endpoint).append("\" ");
  }
  fail("Did not find " + api + " in endpoints " + builder);
  // never reached; here to keep the compiler happy
  return null;
}
 

import org.apache.hadoop.registry.client.types.ServiceRecord; //导入依赖的package包/类
/**
 * Add some endpoints
 * @param entry entry
 */
public static void addSampleEndpoints(ServiceRecord entry, String hostname)
    throws URISyntaxException {
  assertNotNull(hostname);
  entry.addExternalEndpoint(webEndpoint(HTTP_API,
      new URI("http", hostname + ":80", "/")));
  entry.addExternalEndpoint(
      restEndpoint(API_WEBHDFS,
          new URI("http", hostname + ":8020", "/")));

  Endpoint endpoint = ipcEndpoint(API_HDFS, null);
  endpoint.addresses.add(RegistryTypeUtils.hostnamePortPair(hostname, 8030));
  entry.addInternalEndpoint(endpoint);
  InetSocketAddress localhost = new InetSocketAddress("localhost", 8050);
  entry.addInternalEndpoint(
      inetAddrEndpoint(NNIPC, ProtocolTypes.PROTOCOL_THRIFT, "localhost",
          8050));
  entry.addInternalEndpoint(
      RegistryTypeUtils.ipcEndpoint(
          IPC2, localhost));
}
 

import org.apache.hadoop.registry.client.types.ServiceRecord; //导入依赖的package包/类
private String formatOutput(Map<String, ServiceRecord> containers) {
  String format = "  %-40s %-20s %-40s %-25s %-15s %-15s %s\n";
  StringBuilder builder = new StringBuilder().append("\n\n");
  builder.append(String.format(format, YacopConstant.CONTAINER_ID, YacopConstant.IMAGE, YacopConstant.COMMAND, YacopConstant.CREATED, YacopConstant.STATUS, YacopConstant.HOST, YacopConstant.PORT));
  Set<String> containerIds = containers.keySet();
  for (String containerId : containerIds) {
    ServiceRecord record = containers.get(containerId);
    String createdTime = record.get(YacopConstant.CREATED);
    String host = record.get(YacopConstant.HOST);
    String port = record.get(YacopConstant.PORT);
    String image = record.get(YacopConstant.IMAGE);
    String status = record.get(YacopConstant.STATUS);
    String command = record.get(YacopConstant.COMMAND);
    builder.append(String.format(format, containerId, image, command, createdTime, status, host, port));
  }
  return builder.toString();
}
 

import org.apache.hadoop.registry.client.types.ServiceRecord; //导入依赖的package包/类
/**
 * list configs available for an instance
 *
 * @param registryArgs registry Arguments
 * @throws YarnException YARN problems
 * @throws IOException Network or other problems
 * @throws FileNotFoundException if the config is not found
 */
@VisibleForTesting
public PublishedConfiguration actionRegistryGetConfig(ActionRegistryArgs registryArgs)
    throws YarnException, IOException {
  ServiceRecord instance = lookupServiceRecord(registryArgs);

  RegistryRetriever retriever = new RegistryRetriever(instance);
  boolean external = !registryArgs.internal;
  PublishedConfigSet configurations =
      retriever.getConfigurations(external);

  PublishedConfiguration published = retriever.retrieveConfiguration(configurations,
          registryArgs.getConf,
          external);
  return published;
}
 

import org.apache.hadoop.registry.client.types.ServiceRecord; //导入依赖的package包/类
/**
 * get a specific export group
 *
 * @param registryArgs registry Arguments
 *
 * @throws YarnException         YARN problems
 * @throws IOException           Network or other problems
 * @throws FileNotFoundException if the config is not found
 */
@VisibleForTesting
public PublishedExports actionRegistryGetExport(ActionRegistryArgs registryArgs)
    throws YarnException, IOException {
  ServiceRecord instance = lookupServiceRecord(registryArgs);

  RegistryRetriever retriever = new RegistryRetriever(instance);
  boolean external = !registryArgs.internal;
  PublishedExportsSet exports =
      retriever.getExports(external);

  PublishedExports published = retriever.retrieveExports(exports,
                                                         registryArgs.getExport,
                                                         external);
  return published;
}
 

import org.apache.hadoop.registry.client.types.ServiceRecord; //导入依赖的package包/类
/**
 * Handler for {@link RegisterComponentInstance action}
 * Register/re-register an ephemeral container that is already in the app state
 * @param id the component
 * @param description
 */
public boolean registerComponent(ContainerId id, String description) throws
    IOException {
  RoleInstance instance = appState.getOwnedContainer(id);
  if (instance == null) {
    return false;
  }
  // this is where component registrations  go
  log.info("Registering component {}", id);
  String cid = RegistryPathUtils.encodeYarnID(id.toString());
  ServiceRecord container = new ServiceRecord();
  container.set(YarnRegistryAttributes.YARN_ID, cid);
  container.description = description;
  container.set(YarnRegistryAttributes.YARN_PERSISTENCE,
      PersistencePolicies.CONTAINER);
  try {
    yarnRegistryOperations.putComponent(cid, container);
  } catch (IOException e) {
    log.warn("Failed to register container {}/{}: {}",
        id, description, e, e);
  }
  return true;
}
 

import org.apache.hadoop.registry.client.types.ServiceRecord; //导入依赖的package包/类
protected YarnRegistryViewForProviders createYarnRegistryViewForProviders(
    Configuration conf) throws IOException {
  conf.set(SliderXmlConfKeys.REGISTRY_PATH,
      SliderXmlConfKeys.DEFAULT_REGISTRY_PATH);

  RegistryOperations registryOperations = new MockRegistryOperations();
  registryOperations.init(conf);
  YarnRegistryViewForProviders registryViewForProviders =
      new YarnRegistryViewForProviders(registryOperations,
          "hbase",
          SliderKeys.APP_TYPE,
          "hbase1",
          new MockApplicationAttemptId(new MockApplicationId(1), 1));
  registryViewForProviders.registerSelf(new ServiceRecord(), true);
  return registryViewForProviders;
}
 

import org.apache.hadoop.registry.client.types.ServiceRecord; //导入依赖的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.types.ServiceRecord; //导入依赖的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.types.ServiceRecord; //导入依赖的package包/类
/**
 * List service records directly under a path
 * @param registryOperations registry operations instance
 * @param path path to list
 * @return a mapping of the service records that were resolved, indexed
 * by their full path
 * @throws IOException
 */
public static Map<String, ServiceRecord> listServiceRecords(
    RegistryOperations registryOperations,
    String path) throws IOException {
  Map<String, RegistryPathStatus> children =
      statChildren(registryOperations, path);
  return extractServiceRecords(registryOperations,
      path,
      children.values());
}
 

import org.apache.hadoop.registry.client.types.ServiceRecord; //导入依赖的package包/类
@Override
public ServiceRecord resolve(String path) throws IOException {
  byte[] bytes = zkRead(path);

  ServiceRecord record = serviceRecordMarshal.fromBytes(path,
      bytes, ServiceRecord.RECORD_TYPE);
  RegistryTypeUtils.validateServiceRecord(path, record);
  return record;
}
 

import org.apache.hadoop.registry.client.types.ServiceRecord; //导入依赖的package包/类
@Override
public boolean shouldSelect(String path,
    RegistryPathStatus registryPathStatus,
    ServiceRecord serviceRecord) {
  String policy =
      serviceRecord.get(YarnRegistryAttributes.YARN_PERSISTENCE, "");
  return id.equals(serviceRecord.get(YarnRegistryAttributes.YARN_ID, ""))
         && (targetPolicy.equals(policy));
}