import org.apache.hadoop.ha.HAServiceTarget; //导入依赖的package包/类
@Override
protected HAServiceTarget resolveTarget(String rmId) {
  Collection<String> rmIds = HAUtil.getRMHAIds(getConf());
  if (!rmIds.contains(rmId)) {
    StringBuilder msg = new StringBuilder();
    msg.append(rmId + " is not a valid serviceId. It should be one of ");
    for (String id : rmIds) {
      msg.append(id + " ");
    }
    throw new IllegalArgumentException(msg.toString());
  }
  try {
    YarnConfiguration conf = new YarnConfiguration(getConf());
    conf.set(YarnConfiguration.RM_HA_ID, rmId);
    return new RMHAServiceTarget(conf);
  } catch (IllegalArgumentException iae) {
    throw new YarnRuntimeException("Could not connect to " + rmId +
        "; the configuration for it might be missing");
  } catch (IOException ioe) {
    throw new YarnRuntimeException(
        "Could not connect to RM HA Admin for node " + rmId);
  }
}
 

import org.apache.hadoop.ha.HAServiceTarget; //导入依赖的package包/类
@Override
protected HAServiceTarget dataToTarget(byte[] data) {
  ActiveNodeInfo proto;
  try {
    proto = ActiveNodeInfo.parseFrom(data);
  } catch (InvalidProtocolBufferException e) {
    throw new RuntimeException("Invalid data in ZK: " +
        StringUtils.byteToHexString(data));
  }
  NNHAServiceTarget ret = new NNHAServiceTarget(
      conf, proto.getNameserviceId(), proto.getNamenodeId());
  InetSocketAddress addressFromProtobuf = new InetSocketAddress(
      proto.getHostname(), proto.getPort());
  
  if (!addressFromProtobuf.equals(ret.getAddress())) {
    throw new RuntimeException("Mismatched address stored in ZK for " +
        ret + ": Stored protobuf was " + proto + ", address from our own " +
        "configuration for this NameNode was " + ret.getAddress());
  }
  
  ret.setZkfcPort(proto.getZkfcPort());
  return ret;
}
 

import org.apache.hadoop.ha.HAServiceTarget; //导入依赖的package包/类
@Before
public void setup() throws IOException {
  mockProtocol = MockitoUtil.mockProtocol(HAServiceProtocol.class);
  mockZkfcProtocol = MockitoUtil.mockProtocol(ZKFCProtocol.class);
  tool = new DFSHAAdmin() {

    @Override
    protected HAServiceTarget resolveTarget(String nnId) {
      HAServiceTarget target = super.resolveTarget(nnId);
      HAServiceTarget spy = Mockito.spy(target);
      // OVerride the target to return our mock protocol
      try {
        Mockito.doReturn(mockProtocol).when(spy).getProxy(
            Mockito.<Configuration>any(), Mockito.anyInt());
        Mockito.doReturn(mockZkfcProtocol).when(spy).getZKFCProxy(
            Mockito.<Configuration>any(), Mockito.anyInt());
      } catch (IOException e) {
        throw new AssertionError(e); // mock setup doesn't really throw
      }
      return spy;
    }
  };
  tool.setConf(getHAConf());
  tool.setErrOut(new PrintStream(errOutBytes));
  tool.setOut(new PrintStream(outBytes));
}
 

import org.apache.hadoop.ha.HAServiceTarget; //导入依赖的package包/类
@Override
protected HAServiceTarget dataToTarget(byte[] data) {
  ActiveNodeInfo proto;
  try {
    proto = ActiveNodeInfo.parseFrom(data);
  } catch (InvalidProtocolBufferException e) {
    throw new RuntimeException("Invalid data in ZK: " +
        StringUtils.byteToHexString(data));
  }
  JobTrackerHAServiceTarget ret = new JobTrackerHAServiceTarget(
      conf, proto.getNamenodeId());
  InetSocketAddress addressFromProtobuf = new InetSocketAddress(
      proto.getHostname(), proto.getPort());
  
  if (!addressFromProtobuf.equals(ret.getAddress())) {
    throw new RuntimeException("Mismatched address stored in ZK for " +
        ret + ": Stored protobuf was " + addressFromProtobuf + ", address from our own " +
        "configuration for this JobTracker was " + ret.getAddress());
  }
  
  ret.setZkfcPort(proto.getZkfcPort());
  return ret;
}
 

import org.apache.hadoop.ha.HAServiceTarget; //导入依赖的package包/类
@Override
protected byte[] targetToData(HAServiceTarget target) {
  InetSocketAddress addr = target.getAddress();

  return ActiveNodeInfo.newBuilder()
    .setHostname(addr.getHostName())
    .setPort(addr.getPort())
    .setZkfcPort(target.getZKFCAddress().getPort())
    .setNameserviceId(localNNTarget.getNameServiceId())
    .setNamenodeId(localNNTarget.getNameNodeId())
    .build()
    .toByteArray();
}
 

import org.apache.hadoop.ha.HAServiceTarget; //导入依赖的package包/类
@Override
public List<HAServiceTarget> getAllOtherNodes() {
  String nsId = DFSUtil.getNamenodeNameServiceId(conf);
  List<String> otherNn = HAUtil.getNameNodeIdOfOtherNodes(conf, nsId);

  List<HAServiceTarget> targets = new ArrayList<HAServiceTarget>(otherNn.size());
  for (String nnId : otherNn) {
    targets.add(new NNHAServiceTarget(conf, nsId, nnId));
  }
  return targets;
}
 

import org.apache.hadoop.ha.HAServiceTarget; //导入依赖的package包/类
@Override
protected byte[] targetToData(HAServiceTarget target) {
  InetSocketAddress addr = target.getAddress();

  return ActiveNodeInfo.newBuilder()
    .setHostname(addr.getHostName())
    .setPort(addr.getPort())
    .setZkfcPort(target.getZKFCAddress().getPort())
    .setNameserviceId(localJTTarget.getLogicalName())
    .setNamenodeId(localJTTarget.getJobTrackerId())
    .build()
    .toByteArray();
}
 

import org.apache.hadoop.ha.HAServiceTarget; //导入依赖的package包/类
/**
 * Try to map the given namenode ID to its service address.
 */
@Override
protected HAServiceTarget resolveTarget(String nnId) {
  HdfsConfiguration conf = (HdfsConfiguration)getConf();
  return new NNHAServiceTarget(conf, nameserviceId, nnId);
}
 

import org.apache.hadoop.ha.HAServiceTarget; //导入依赖的package包/类
@SuppressWarnings("static-access")
@Before
public void configure() throws IOException, YarnException {
  remoteAdminServiceAccessed = false;
  admin = mock(ResourceManagerAdministrationProtocol.class);
  when(admin.addToClusterNodeLabels(any(AddToClusterNodeLabelsRequest.class)))
      .thenAnswer(new Answer<AddToClusterNodeLabelsResponse>() {

        @Override
        public AddToClusterNodeLabelsResponse answer(
            InvocationOnMock invocation) throws Throwable {
          remoteAdminServiceAccessed = true;
          return AddToClusterNodeLabelsResponse.newInstance();
        }
      });

  haadmin = mock(HAServiceProtocol.class);
  when(haadmin.getServiceStatus()).thenReturn(new HAServiceStatus(
      HAServiceProtocol.HAServiceState.INITIALIZING));

  final HAServiceTarget haServiceTarget = mock(HAServiceTarget.class);
  when(haServiceTarget.getProxy(any(Configuration.class), anyInt()))
      .thenReturn(haadmin);
  rmAdminCLI = new RMAdminCLI(new Configuration()) {
    @Override
    protected ResourceManagerAdministrationProtocol createAdminProtocol()
        throws IOException {
      return admin;
    }

    @Override
    protected HAServiceTarget resolveTarget(String rmId) {
      return haServiceTarget;
    }
  };
  initDummyNodeLabelsManager();
  rmAdminCLI.localNodeLabelsManager = dummyNodeLabelsManager;

  YarnConfiguration conf = new YarnConfiguration();
  conf.setBoolean(YarnConfiguration.RM_HA_ENABLED, true);
  conf.set(YarnConfiguration.RM_HA_IDS, "rm1,rm2");
  rmAdminCLIWithHAEnabled = new RMAdminCLI(conf) {

    @Override
    protected ResourceManagerAdministrationProtocol createAdminProtocol()
        throws IOException {
      return admin;
    }

    @Override
    protected HAServiceTarget resolveTarget(String rmId) {
      return haServiceTarget;
    }
  };
}