import org.apache.hadoop.registry.client.types.yarn.YarnRegistryAttributes; //导入依赖的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.yarn.YarnRegistryAttributes; //导入依赖的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));
}
 

import org.apache.hadoop.registry.client.types.yarn.YarnRegistryAttributes; //导入依赖的package包/类
@Test
public void testPutNoParent() throws Throwable {
  ServiceRecord record = new ServiceRecord();
  record.set(YarnRegistryAttributes.YARN_ID, "testPutNoParent");
  String path = "/path/without/parent";
  try {
    operations.bind(path, record, 0);
    // didn't get a failure
    // trouble
    RegistryPathStatus stat = operations.stat(path);
    fail("Got a status " + stat);
  } catch (PathNotFoundException expected) {
    // expected
  }
}
 

import org.apache.hadoop.registry.client.types.yarn.YarnRegistryAttributes; //导入依赖的package包/类
@Test(expected = PathNotFoundException.class)
public void testPutNoParent2() throws Throwable {
  ServiceRecord record = new ServiceRecord();
  record.set(YarnRegistryAttributes.YARN_ID, "testPutNoParent");
  String path = "/path/without/parent";
  operations.bind(path, record, 0);
}
 

import org.apache.hadoop.registry.client.types.yarn.YarnRegistryAttributes; //导入依赖的package包/类
/**
 * Create a service entry with the sample endpoints
 * @param persistence persistence policy
 * @return the record
 * @throws IOException on a failure
 */
public static ServiceRecord buildExampleServiceEntry(String persistence) throws
    IOException,
    URISyntaxException {
  ServiceRecord record = new ServiceRecord();
  record.set(YarnRegistryAttributes.YARN_ID, "example-0001");
  record.set(YarnRegistryAttributes.YARN_PERSISTENCE, persistence);
  addSampleEndpoints(record, "namenode");
  return record;
}
 

import org.apache.hadoop.registry.client.types.yarn.YarnRegistryAttributes; //导入依赖的package包/类
public static ServiceRecord createRecord(String id, String persistence,
    String description) {
  ServiceRecord serviceRecord = new ServiceRecord();
  serviceRecord.set(YarnRegistryAttributes.YARN_ID, id);
  serviceRecord.description = description;
  serviceRecord.set(YarnRegistryAttributes.YARN_PERSISTENCE, persistence);
  return serviceRecord;
}
 

import org.apache.hadoop.registry.client.types.yarn.YarnRegistryAttributes; //导入依赖的package包/类
/**
 * Log a service record instance
 * @param instance record
 * @param verbose verbose logging of all external endpoints
 */
private void logInstance(ServiceRecord instance,
    boolean verbose) {
  if (!verbose) {
    log.info("{}", instance.get(YarnRegistryAttributes.YARN_ID, ""));
  } else {
    log.info("{}: ", instance.get(YarnRegistryAttributes.YARN_ID, ""));
    logEndpoints(instance);
  }
}
 

import org.apache.hadoop.registry.client.types.yarn.YarnRegistryAttributes; //导入依赖的package包/类
private void registerHBaseServiceEntry() throws IOException {

    String name = amState.getApplicationName() ;
    ServiceRecord serviceRecord = new ServiceRecord();
    // bond lifespan to the application
    serviceRecord.set(YarnRegistryAttributes.YARN_ID,
        yarnRegistry.getApplicationAttemptId()
                    .getApplicationId().toString());
    serviceRecord.set(YarnRegistryAttributes.YARN_PERSISTENCE,
        PersistencePolicies.APPLICATION);
    try {
      URL configURL = new URL(amWebAPI,
          SLIDER_PATH_PUBLISHER + "/" + HBASE_SERVICE_TYPE);

      serviceRecord.addExternalEndpoint(
          RegistryTypeUtils.restEndpoint(
              CustomRegistryConstants.PUBLISHER_CONFIGURATIONS_API,
              configURL.toURI()));
    } catch (URISyntaxException e) {
      log.warn("failed to create config URL: {}", e, e);
    }
    log.info("registering {}/{}", name, HBASE_SERVICE_TYPE);
    yarnRegistry.putService(HBASE_SERVICE_TYPE, name, serviceRecord, true);

    PublishedConfiguration publishedSite =
        new PublishedConfiguration("HBase site", siteConf);
    PublishedConfigSet configSet =
        amState.getOrCreatePublishedConfigSet(HBASE_SERVICE_TYPE);

    configSet.put(HBASE_SITE_PUBLISHED_CONFIG, publishedSite);    
  }
 

import org.apache.hadoop.registry.client.types.yarn.YarnRegistryAttributes; //导入依赖的package包/类
private ServiceRecord setupServiceRecord() {
    ServiceRecord application = new ServiceRecord();
    application.set(YarnRegistryAttributes.YARN_ID, jstormMasterContext.appAttemptID.getApplicationId().toString());
    application.description = JOYConstants.AM;
    application.set(YarnRegistryAttributes.YARN_PERSISTENCE,
            PersistencePolicies.PERMANENT);
    Map<String, String> addresses = new HashMap<String, String>();
    addresses.put(JOYConstants.HOST, jstormMasterContext.appMasterHostname);
    addresses.put(JOYConstants.PORT, String.valueOf(jstormMasterContext.appMasterThriftPort));
    Endpoint endpoint = new Endpoint(JOYConstants.HTTP, JOYConstants.HOST_PORT, JOYConstants.RPC, addresses);
    application.addExternalEndpoint(endpoint);
    return application;
}
 

import org.apache.hadoop.registry.client.types.yarn.YarnRegistryAttributes; //导入依赖的package包/类
@Test
public void testPurgeEntryCuratorCallback() throws Throwable {

  String path = "/users/example/hbase/hbase1/";
  ServiceRecord written = buildExampleServiceEntry(
      PersistencePolicies.APPLICATION_ATTEMPT);
  written.set(YarnRegistryAttributes.YARN_ID,
      "testAsyncPurgeEntry_attempt_001");

  operations.mknode(RegistryPathUtils.parentOf(path), true);
  operations.bind(path, written, 0);

  ZKPathDumper dump = registry.dumpPath(false);
  CuratorEventCatcher events = new CuratorEventCatcher();

  LOG.info("Initial state {}", dump);

  // container query
  String id = written.get(YarnRegistryAttributes.YARN_ID, "");
  int opcount = purge("/",
      id,
      PersistencePolicies.CONTAINER,
      RegistryAdminService.PurgePolicy.PurgeAll,
      events);
  assertPathExists(path);
  assertEquals(0, opcount);
  assertEquals("Event counter", 0, events.getCount());

  // now the application attempt
  opcount = purge("/",
      id,
      PersistencePolicies.APPLICATION_ATTEMPT,
      RegistryAdminService.PurgePolicy.PurgeAll,
      events);

  LOG.info("Final state {}", dump);

  assertPathNotFound(path);
  assertEquals("wrong no of delete operations in " + dump, 1, opcount);
  // and validate the callback event
  assertEquals("Event counter", 1, events.getCount());
}
 

import org.apache.hadoop.registry.client.types.yarn.YarnRegistryAttributes; //导入依赖的package包/类
/**
   * This registers the service instance and its external values
   * @param instanceName name of this instance
   * @param appid application ID
   * @throws IOException
   */
  private void registerServiceInstance(String instanceName,
      ApplicationId appid) throws IOException {
    
    
    // the registry is running, so register services
    URL amWebURI = new URL(appMasterTrackingUrl);
    URL agentOpsURI = new URL(agentOpsUrl);
    URL agentStatusURI = new URL(agentStatusUrl);

    //Give the provider restricted access to the state, registry
    setupInitialRegistryPaths();
    yarnRegistryOperations = new YarnRegistryViewForProviders(
        registryOperations,
        service_user_name,
        SliderKeys.APP_TYPE,
        instanceName,
        appAttemptID);
    providerService.bindToYarnRegistry(yarnRegistryOperations);
    sliderAMProvider.bindToYarnRegistry(yarnRegistryOperations);

    // Yarn registry
    ServiceRecord serviceRecord = new ServiceRecord();
    serviceRecord.set(YarnRegistryAttributes.YARN_ID, appid.toString());
    serviceRecord.set(YarnRegistryAttributes.YARN_PERSISTENCE,
        PersistencePolicies.APPLICATION);
    serviceRecord.description = "Slider Application Master";

/* SLIDER-531: disable this addition so things compile against versions of
the registry with/without the new record format

    serviceRecord.addExternalEndpoint(
        RegistryTypeUtils.ipcEndpoint(
            CustomRegistryConstants.AM_IPC_PROTOCOL,
            rpcServiceAddress));
            
    */
    // internal services
    sliderAMProvider.applyInitialRegistryDefinitions(amWebURI,
        agentOpsURI,
        agentStatusURI,
        serviceRecord);

    // provider service dynamic definitions.
    providerService.applyInitialRegistryDefinitions(amWebURI,
        agentOpsURI,
        agentStatusURI,
        serviceRecord);

    // register the service's entry
    log.info("Service Record \n{}", serviceRecord);
    yarnRegistryOperations.registerSelf(serviceRecord, true);
    log.info("Registered service under {}; absolute path {}",
        yarnRegistryOperations.getSelfRegistrationPath(),
        yarnRegistryOperations.getAbsoluteSelfRegistrationPath());
    
    boolean isFirstAttempt = 1 == appAttemptID.getAttemptId();
    // delete the children in case there are any and this is an AM startup.
    // just to make sure everything underneath is purged
    if (isFirstAttempt) {
      yarnRegistryOperations.deleteChildren(
          yarnRegistryOperations.getSelfRegistrationPath(),
          true);
    }
  }