import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
@BeforeSuite
public void beforeSuite() throws Exception {
  // TODO: use logging.properties file to config java.util.logging.Logger levels
  java.util.logging.Logger topJavaLogger = java.util.logging.Logger.getLogger("");
  topJavaLogger.setLevel(Level.WARNING);

  // Due to ZOOKEEPER-2693 fix, we need to specify whitelist for execute zk commends
  System.setProperty("zookeeper.4lw.commands.whitelist", "*");

  _zkServer = TestHelper.startZkServer(ZK_ADDR);
  AssertJUnit.assertTrue(_zkServer != null);
  ZKClientPool.reset();

  _gZkClient = new ZkClient(ZK_ADDR);
  _gZkClient.setZkSerializer(new ZNRecordSerializer());
  _gSetupTool = new ClusterSetup(_gZkClient);
  _baseAccessor = new ZkBaseDataAccessor<>(_gZkClient);
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
@Override
public boolean verify() {
  HelixDataAccessor accessor =
      new ZKHelixDataAccessor(_clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
  Builder keyBuilder = accessor.keyBuilder();
  int numberOfPartitions =
      accessor.getProperty(keyBuilder.idealStates(_resourceName)).getRecord().getListFields()
          .size();
  ClusterDataCache cache = new ClusterDataCache();
  cache.refresh(accessor);
  String masterValue =
      cache.getStateModelDef(cache.getIdealState(_resourceName).getStateModelDefRef())
          .getStatesPriorityList().get(0);
  int replicas = Integer.parseInt(cache.getIdealState(_resourceName).getReplicas());
  return verifyBalanceExternalView(accessor.getProperty(keyBuilder.externalView(_resourceName))
      .getRecord(), numberOfPartitions, masterValue, replicas, cache.getLiveInstances().size(),
      cache.getIdealState(_resourceName).getMaxPartitionsPerInstance());
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
protected void setupStateModel(String clusterName) {
  ZKHelixDataAccessor accessor =
      new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
  Builder keyBuilder = accessor.keyBuilder();

  StateModelDefinition masterSlave =
      new StateModelDefinition(StateModelConfigGenerator.generateConfigForMasterSlave());
  accessor.setProperty(keyBuilder.stateModelDef(masterSlave.getId()), masterSlave);

  StateModelDefinition leaderStandby =
      new StateModelDefinition(StateModelConfigGenerator.generateConfigForLeaderStandby());
  accessor.setProperty(keyBuilder.stateModelDef(leaderStandby.getId()), leaderStandby);

  StateModelDefinition onlineOffline =
      new StateModelDefinition(StateModelConfigGenerator.generateConfigForOnlineOffline());
  accessor.setProperty(keyBuilder.stateModelDef(onlineOffline.getId()), onlineOffline);

}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
@BeforeSuite
public void beforeSuite() throws Exception {
  if (!_init) {
    // TODO: use logging.properties file to config java.util.logging.Logger levels
    java.util.logging.Logger topJavaLogger = java.util.logging.Logger.getLogger("");
    topJavaLogger.setLevel(Level.WARNING);

    _gZkClient = new ZkClient(ZK_ADDR, ZkClient.DEFAULT_CONNECTION_TIMEOUT,
        ZkClient.DEFAULT_SESSION_TIMEOUT, new ZNRecordSerializer());
    _gZkClientTestNS = new ZkClient(_zkAddrTestNS, ZkClient.DEFAULT_CONNECTION_TIMEOUT, ZkClient.DEFAULT_SESSION_TIMEOUT,
        new ZNRecordSerializer());
    _gSetupTool = new ClusterSetup(_gZkClient);
    _configAccessor = new ConfigAccessor(_gZkClient);
    _baseAccessor = new ZkBaseDataAccessor<>(_gZkClient);
    _baseAccessorTestNS = new ZkBaseDataAccessor<>(_gZkClientTestNS);

    // wait for the web service to start
    Thread.sleep(100);

    setup();
    _init = true;
  }
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
public static Map<String, String> readZkChildrenAsBytesMap(ZkClient zkclient, PropertyKey propertyKey) {
  BaseDataAccessor<byte[]> baseAccessor = new ZkBaseDataAccessor<byte[]>(zkclient);
  String parentPath = propertyKey.getPath();
  List<String> childNames = baseAccessor.getChildNames(parentPath, 0);
  if (childNames == null) {
    return null;
  }
  List<String> paths = new ArrayList<String>();
  for (String childName : childNames) {
    paths.add(parentPath + "/" + childName);
  }
  List<byte[]> values = baseAccessor.get(paths, null, 0);
  Map<String, String> ret = new HashMap<String, String>();
  for (int i = 0; i < childNames.size(); i++) {
    ret.put(childNames.get(i), new String(values.get(i)));
  }
  return ret;
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
/**
 * A constructor that gets a {@link HelixNotifier} based on {@link HelixPropertyStoreConfig}.
 * @param storeConfig A {@link HelixPropertyStore} used to instantiate a {@link HelixNotifier}. Cannot be {@code null}.
 */
public HelixNotifier(HelixPropertyStoreConfig storeConfig) {
  if (storeConfig == null) {
    throw new IllegalArgumentException("storeConfig cannot be null");
  }
  long startTimeMs = System.currentTimeMillis();
  logger.info("Starting a HelixNotifier");
  ZkClient zkClient = new ZkClient(storeConfig.zkClientConnectString, storeConfig.zkClientSessionTimeoutMs,
      storeConfig.zkClientConnectionTimeoutMs, new ZNRecordSerializer());
  List<String> subscribedPaths = Collections.singletonList(storeConfig.rootPath + HelixNotifier.TOPIC_PATH);
  HelixPropertyStore<ZNRecord> helixStore =
      new ZkHelixPropertyStore<>(new ZkBaseDataAccessor<>(zkClient), storeConfig.rootPath, subscribedPaths);
  logger.info("HelixPropertyStore started with zkClientConnectString={}, zkClientSessionTimeoutMs={}, "
          + "zkClientConnectionTimeoutMs={}, rootPath={}, subscribedPaths={}", storeConfig.zkClientConnectString,
      storeConfig.zkClientSessionTimeoutMs, storeConfig.zkClientConnectionTimeoutMs, storeConfig.rootPath,
      subscribedPaths);
  this.helixStore = helixStore;
  long startUpTimeInMs = System.currentTimeMillis() - startTimeMs;
  logger.info("HelixNotifier started, took {} ms", startUpTimeInMs);
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
/**
 * A constructor that gets a {@link HelixStoreOperator} based on the {@link HelixPropertyStoreConfig}.
 * @param storeConfig A {@link HelixPropertyStore} used to instantiate a {@link HelixStoreOperator}.
 */
public HelixStoreOperator(HelixPropertyStoreConfig storeConfig) {
  if (storeConfig == null) {
    throw new IllegalArgumentException("storeConfig cannot be null");
  }
  long startTimeMs = System.currentTimeMillis();
  logger.info("Starting a HelixStoreOperator");
  ZkClient zkClient = new ZkClient(storeConfig.zkClientConnectString, storeConfig.zkClientSessionTimeoutMs,
      storeConfig.zkClientConnectionTimeoutMs, new ZNRecordSerializer());
  List<String> subscribedPaths = Collections.singletonList(storeConfig.rootPath);
  HelixPropertyStore<ZNRecord> helixStore =
      new ZkHelixPropertyStore<>(new ZkBaseDataAccessor<>(zkClient), storeConfig.rootPath, subscribedPaths);
  logger.info("HelixPropertyStore started with zkClientConnectString={}, zkClientSessionTimeoutMs={}, "
          + "zkClientConnectionTimeoutMs={}, rootPath={}, subscribedPaths={}", storeConfig.zkClientConnectString,
      storeConfig.zkClientSessionTimeoutMs, storeConfig.zkClientConnectionTimeoutMs, storeConfig.rootPath,
      subscribedPaths);
  this.helixStore = helixStore;
  logger.info("HelixStoreOperator started, took {}ms", System.currentTimeMillis() - startTimeMs);
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
@Override
public AccountService getAccountService() {
  try {
    long startTimeMs = System.currentTimeMillis();
    logger.info("Starting a HelixAccountService");
    ZkClient zkClient = new ZkClient(storeConfig.zkClientConnectString, storeConfig.zkClientSessionTimeoutMs,
        storeConfig.zkClientConnectionTimeoutMs, new ZNRecordSerializer());
    HelixPropertyStore<ZNRecord> helixStore =
        new ZkHelixPropertyStore<>(new ZkBaseDataAccessor<>(zkClient), storeConfig.rootPath, null);
    logger.info("HelixPropertyStore started with zkClientConnectString={}, zkClientSessionTimeoutMs={}, "
            + "zkClientConnectionTimeoutMs={}, rootPath={}", storeConfig.zkClientConnectString,
        storeConfig.zkClientSessionTimeoutMs, storeConfig.zkClientConnectionTimeoutMs, storeConfig.rootPath);
    ScheduledExecutorService scheduler =
        accountServiceConfig.updaterPollingIntervalMs > 0 ? Utils.newScheduler(1, HELIX_ACCOUNT_UPDATER_PREFIX, false)
            : null;
    HelixAccountService helixAccountService =
        new HelixAccountService(helixStore, accountServiceMetrics, notifier, scheduler, accountServiceConfig);
    long spentTimeMs = System.currentTimeMillis() - startTimeMs;
    logger.info("HelixAccountService started, took {} ms", spentTimeMs);
    accountServiceMetrics.startupTimeInMs.update(spentTimeMs);
    return helixAccountService;
  } catch (Exception e) {
    throw new IllegalStateException("Could not instantiate HelixAccountService", e);
  }
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
public static ZkHelixPropertyStore<ZNRecord> getZkPropertyStore(HelixManager helixManager,
    String clusterName) {
  ZkBaseDataAccessor<ZNRecord> baseAccessor =
      (ZkBaseDataAccessor<ZNRecord>) helixManager.getHelixDataAccessor().getBaseDataAccessor();
  String propertyStorePath = PropertyPathConfig.getPath(PropertyType.PROPERTYSTORE, clusterName);

  ZkHelixPropertyStore<ZNRecord> propertyStore = new ZkHelixPropertyStore<ZNRecord>(baseAccessor,
      propertyStorePath, Arrays.asList(propertyStorePath));

  return propertyStore;
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
@BeforeTest
public void beforeTest() {
  ZkStarter.startLocalZkServer();

  _zkClient =
      new ZkClient(StringUtil.join("/", StringUtils.chomp(ZkStarter.DEFAULT_ZK_STR, "/")),
          ZkClient.DEFAULT_SESSION_TIMEOUT, ZkClient.DEFAULT_CONNECTION_TIMEOUT, new ZNRecordSerializer());
  String helixClusterName = "TestTimeBoundaryService";
  _zkClient.deleteRecursive("/" + helixClusterName + "/PROPERTYSTORE");
  _zkClient.createPersistent("/" + helixClusterName + "/PROPERTYSTORE", true);
  _propertyStore =
      new ZkHelixPropertyStore<ZNRecord>(new ZkBaseDataAccessor<ZNRecord>(_zkClient), "/" + helixClusterName
          + "/PROPERTYSTORE", null);

}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
/**
 * Returns a handle to the propertyStore, setting up a watch on all nodes under PROPERTYSTORE/clusterName, if
 * setWatch is specified as true.
 *
 * @param helixManager * A valid helix manager bound to a cluster.
 * @param clusterName The pathname under propertyStore that we want a handle for
 * @param setWatch Must be set to true if it is desired to set up a watch on all nodes under the PROPERTYSTORE node.
 * @return A handle to the propertyStore.
 *
 * @note: Setting up a watch can lead to connection problems with zookeeper during a re-connect from the client. The
 * client ends up sending all the path names in the session request, and that can exceed the max size of a session
 * request if there are too many nodes, causing the connection to be dropped and retried (with the same parameters).
 */
public static ZkHelixPropertyStore<ZNRecord> getZkPropertyStore(HelixManager helixManager, String clusterName, boolean setWatch) {
  ZkBaseDataAccessor<ZNRecord> baseAccessor =
      (ZkBaseDataAccessor<ZNRecord>) helixManager.getHelixDataAccessor().getBaseDataAccessor();
  String propertyStorePath = PropertyPathConfig.getPath(PropertyType.PROPERTYSTORE, clusterName);

  List<String> watchList = null;
  if (setWatch) {
    watchList = Arrays.asList(propertyStorePath);
  }

  ZkHelixPropertyStore<ZNRecord> propertyStore = new ZkHelixPropertyStore<ZNRecord>(baseAccessor, propertyStorePath, watchList);

  return propertyStore;
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
@Override
public boolean verify() {
  try {
    HelixDataAccessor accessor =
        new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(zkClient));

    return verifyBestPossAndExtView(accessor, errStates, clusterName, resources);
  } catch (Exception e) {
    LOG.error("exception in verification", e);
  }
  return false;
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
public ZkHelixClusterVerifier(ZkClient zkClient, String clusterName) {
  if (zkClient == null || clusterName == null) {
    throw new IllegalArgumentException("requires zkClient|clusterName");
  }
  _zkClient = zkClient;
  _clusterName = clusterName;
  _accessor = new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_zkClient));
  _keyBuilder = _accessor.keyBuilder();
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
public AutoFallbackPropertyStore(ZkBaseDataAccessor<T> accessor, String root, String fallbackRoot) {
  super(accessor, root, null);

  if (accessor.exists(fallbackRoot, 0)) {
    _fallbackStore = new ZkHelixPropertyStore<T>(accessor, fallbackRoot, null);
  } else {
    LOG.info("fallbackRoot: " + fallbackRoot
        + " doesn't exist, skip creating fallback property store");
    _fallbackStore = null;
  }

}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
@Test(dependsOnMethods = {"testUpdateOnNotification"})
public void testSelectiveUpdates() throws Exception {
  MockZkHelixDataAccessor accessor =
      new MockZkHelixDataAccessor(CLUSTER_NAME, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));

  ClusterDataCache cache =
      new ClusterDataCache("CLUSTER_" + TestHelper.getTestClassName());
  cache.refresh(accessor);

  Assert.assertEquals(accessor.getReadCount(PropertyType.IDEALSTATES), 1);
  Assert.assertEquals(accessor.getReadCount(PropertyType.LIVEINSTANCES), NODE_NR);
  Assert.assertEquals(accessor.getReadCount(PropertyType.CURRENTSTATES), NODE_NR);
  Assert.assertEquals(accessor.getReadCount(PropertyType.CONFIGS), NODE_NR + 1);

  accessor.clearReadCounters();

  // refresh again should read nothing
  cache.refresh(accessor);
  Assert.assertEquals(accessor.getReadCount(PropertyType.IDEALSTATES), 0);
  Assert.assertEquals(accessor.getReadCount(PropertyType.LIVEINSTANCES), 0);
  Assert.assertEquals(accessor.getReadCount(PropertyType.CURRENTSTATES), 0);
  Assert.assertEquals(accessor.getReadCount(PropertyType.CONFIGS), 1);

  // add a new resource
  _setupTool.addResourceToCluster(CLUSTER_NAME, "TestDB_1", _PARTITIONS, STATE_MODEL);
  _setupTool.rebalanceStorageCluster(CLUSTER_NAME, "TestDB_1", _replica);

  Thread.sleep(100);
  HelixClusterVerifier _clusterVerifier =
      new BestPossibleExternalViewVerifier.Builder(CLUSTER_NAME).setZkAddr(ZK_ADDR).build();
  Assert.assertTrue(_clusterVerifier.verify());

  accessor.clearReadCounters();

  // refresh again should read only new current states
  cache.refresh(accessor);
  Assert.assertEquals(accessor.getReadCount(PropertyType.CURRENTSTATES), NODE_NR);
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
/**
 * Assert externalView and currentState for each participant are empty
 * @param clusterName
 * @param db
 * @param participants
 */
private void assertEmptyCSandEV(String clusterName, String db, MockParticipantManager[] participants) {
  HelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
  PropertyKey.Builder keyBuilder = accessor.keyBuilder();
  Assert.assertNull(accessor.getProperty(keyBuilder.externalView(db)));

  for (MockParticipantManager participant : participants) {
    String instanceName = participant.getInstanceName();
    String sessionId = participant.getSessionId();
    Assert.assertNull(accessor.getProperty(keyBuilder.currentState(instanceName, sessionId, db)));
  }
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
@Test()
public void testStandAloneCMMain() throws Exception {
  logger.info("RUN testStandAloneCMMain() at " + new Date(System.currentTimeMillis()));
  ClusterControllerManager newController = null;
  for (int i = 1; i <= 2; i++) {
    String controllerName = "controller_" + i;
    newController =
        new ClusterControllerManager(ZK_ADDR, CLUSTER_NAME, controllerName);
    newController.syncStart();
  }

  // stopCurrentLeader(_zkClient, CLUSTER_NAME, _startCMResultMap);
  _controller.syncStop();

  final HelixDataAccessor accessor =
      new ZKHelixDataAccessor(CLUSTER_NAME, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
  final PropertyKey.Builder keyBuilder = accessor.keyBuilder();
  final String newControllerName = newController.getInstanceName();
  TestHelper.verify(new Verifier() {

    @Override
    public boolean verify() throws Exception {
      LiveInstance leader = accessor.getProperty(keyBuilder.controllerLeader());
      if (leader == null) {
        return false;
      }
      return leader.getInstanceName().equals(newControllerName);

    }
  }, 30 * 1000);

  boolean result =
      ClusterStateVerifier.verifyByPolling(new ClusterStateVerifier.BestPossAndExtViewZkVerifier(
          ZK_ADDR, CLUSTER_NAME));
  Assert.assertTrue(result);

  logger.info("STOP testStandAloneCMMain() at " + new Date(System.currentTimeMillis()));
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
@Test
public void testDisableExternalView() throws InterruptedException {
  ZKHelixDataAccessor accessor =
      new ZKHelixDataAccessor(CLUSTER_NAME, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
  PropertyKey.Builder keyBuilder = accessor.keyBuilder();

  // verify external view for TEST_DB1 does not exist
  ExternalView externalView = null;
  externalView = accessor.getProperty(keyBuilder.externalView(TEST_DB1));
  Assert.assertNull(externalView,
      "There should be no external-view for " + TEST_DB1 + ", but is: " + externalView);

  // verify external view for TEST_DB2 exists
  externalView = accessor.getProperty(keyBuilder.externalView(TEST_DB2));
  Assert.assertNotNull(externalView,
      "Could not find external-view for " + TEST_DB2);

  // disable external view in IS
  IdealState idealState = _admin.getResourceIdealState(CLUSTER_NAME, TEST_DB2);
  idealState.setDisableExternalView(true);
  _admin.setResourceIdealState(CLUSTER_NAME, TEST_DB2, idealState);

  // touch liveinstance to trigger externalview compute stage
  String instance = PARTICIPANT_PREFIX + "_" + START_PORT;
  HelixProperty liveInstance = accessor.getProperty(keyBuilder.liveInstance(instance));
  accessor.setProperty(keyBuilder.liveInstance(instance), liveInstance);

  // verify the external view for the db got removed
  for (int i = 0; i < 10; i++) {
    Thread.sleep(100);
    externalView = accessor.getProperty(keyBuilder.externalView(TEST_DB2));
    if (externalView == null) {
      break;
    }
  }

  Assert.assertNull(externalView, "external-view for " + TEST_DB2 + " should be removed, but was: "
      + externalView);
}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
private void clearStatusUpdate(String clusterName, String instance, String resource,
    String partition) {
  // clear status update for error partition so verify() will not fail on old
  // errors
  ZKHelixDataAccessor accessor =
      new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
  Builder keyBuilder = accessor.keyBuilder();

  LiveInstance liveInstance = accessor.getProperty(keyBuilder.liveInstance(instance));
  accessor.removeProperty(keyBuilder.stateTransitionStatus(instance, liveInstance.getSessionId(),
      resource, partition));

}
 

import org.apache.helix.manager.zk.ZkBaseDataAccessor; //导入依赖的package包/类
/**
 * Check all partitions are in OFFLINE state
 * @param clusterName
 * @throws Exception
 */
private void checkExternalView(String clusterName) throws Exception {
  BaseDataAccessor<ZNRecord> baseAccessor = new ZkBaseDataAccessor<ZNRecord>(_gZkClient);
  final HelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, baseAccessor);

  // verify that states of TestDB0 are all OFFLINE
  boolean result = TestHelper.verify(new TestHelper.Verifier() {

    @Override
    public boolean verify() throws Exception {
      PropertyKey.Builder keyBuilder = accessor.keyBuilder();

      ExternalView extView = accessor.getProperty(keyBuilder.externalView("TestDB0"));
      if (extView == null) {
        return false;
      }
      Set<String> partitionSet = extView.getPartitionSet();
      if (partitionSet == null || partitionSet.size() != PARTITION_NUM) {
        return false;
      }
      for (String partition : partitionSet) {
        Map<String, String> instanceStates = extView.getStateMap(partition);
        for (String state : instanceStates.values()) {
          if (!"OFFLINE".equals(state)) {
            return false;
          }
        }
      }
      return true;
    }
  }, 10 * 1000);
  Assert.assertTrue(result);
}