| package org.apache.helix; |
| |
| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.lang.reflect.Method; |
| import java.net.ServerSocket; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.TreeMap; |
| import java.util.TreeSet; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.TimeUnit; |
| |
| import org.apache.commons.io.FileUtils; |
| import org.apache.helix.PropertyKey.Builder; |
| import org.apache.helix.controller.rebalancer.strategy.CrushEdRebalanceStrategy; |
| import org.apache.helix.controller.rebalancer.strategy.RebalanceStrategy; |
| import org.apache.helix.integration.manager.ZkTestManager; |
| import org.apache.helix.manager.zk.CallbackHandler; |
| import org.apache.helix.manager.zk.ZKHelixAdmin; |
| import org.apache.helix.manager.zk.ZKHelixDataAccessor; |
| import org.apache.helix.manager.zk.ZNRecordSerializer; |
| import org.apache.helix.manager.zk.ZkBaseDataAccessor; |
| import org.apache.helix.model.CurrentState; |
| import org.apache.helix.model.ExternalView; |
| import org.apache.helix.model.IdealState.RebalanceMode; |
| import org.apache.helix.model.Message; |
| import org.apache.helix.model.Message.MessageType; |
| import org.apache.helix.model.StateModelDefinition; |
| import org.apache.helix.model.StateModelDefinition.StateModelDefinitionProperty; |
| import org.apache.helix.store.zk.ZNode; |
| import org.apache.helix.tools.ClusterSetup; |
| import org.apache.helix.util.ZKClientPool; |
| import org.apache.helix.zookeeper.api.client.HelixZkClient; |
| import org.apache.helix.zookeeper.api.client.RealmAwareZkClient; |
| import org.apache.helix.zookeeper.datamodel.ZNRecord; |
| import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory; |
| import org.apache.helix.zookeeper.zkclient.IDefaultNameSpace; |
| import org.apache.helix.zookeeper.zkclient.IZkChildListener; |
| import org.apache.helix.zookeeper.zkclient.IZkDataListener; |
| import org.apache.helix.zookeeper.zkclient.ZkClient; |
| import org.apache.helix.zookeeper.zkclient.ZkServer; |
| import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException; |
| import org.apache.zookeeper.data.Stat; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.testng.Assert; |
| |
| |
| public class TestHelper { |
| private static final Logger LOG = LoggerFactory.getLogger(TestHelper.class); |
| public static final long WAIT_DURATION = 60 * 1000L; // 60 seconds |
| public static final int DEFAULT_REBALANCE_PROCESSING_WAIT_TIME = 1500; |
| /** |
| * Returns a unused random port. |
| */ |
| public static int getRandomPort() throws IOException { |
| ServerSocket sock = new ServerSocket(); |
| sock.bind(null); |
| int port = sock.getLocalPort(); |
| sock.close(); |
| |
| return port; |
| } |
| |
| static public ZkServer startZkServer(final String zkAddress) throws Exception { |
| List<String> empty = Collections.emptyList(); |
| return TestHelper.startZkServer(zkAddress, empty, true); |
| } |
| |
| static public ZkServer startZkServer(final String zkAddress, final String rootNamespace) |
| throws Exception { |
| List<String> rootNamespaces = new ArrayList<String>(); |
| rootNamespaces.add(rootNamespace); |
| return TestHelper.startZkServer(zkAddress, rootNamespaces, true); |
| } |
| |
| static public ZkServer startZkServer(final String zkAddress, final List<String> rootNamespaces) |
| throws Exception { |
| return startZkServer(zkAddress, rootNamespaces, true); |
| } |
| |
| static public ZkServer startZkServer(final String zkAddress, final List<String> rootNamespaces, |
| boolean overwrite) throws Exception { |
| System.out.println( |
| "Start zookeeper at " + zkAddress + " in thread " + Thread.currentThread().getName()); |
| |
| String zkDir = zkAddress.replace(':', '_'); |
| final String logDir = "/tmp/" + zkDir + "/logs"; |
| final String dataDir = "/tmp/" + zkDir + "/dataDir"; |
| if (overwrite) { |
| FileUtils.deleteDirectory(new File(dataDir)); |
| FileUtils.deleteDirectory(new File(logDir)); |
| } |
| ZKClientPool.reset(); |
| |
| IDefaultNameSpace defaultNameSpace = new IDefaultNameSpace() { |
| @Override |
| public void createDefaultNameSpace(ZkClient zkClient) { |
| if (rootNamespaces == null) { |
| return; |
| } |
| |
| for (String rootNamespace : rootNamespaces) { |
| try { |
| zkClient.deleteRecursive(rootNamespace); |
| } catch (Exception e) { |
| LOG.error("fail to deleteRecursive path:" + rootNamespace, e); |
| } |
| } |
| } |
| }; |
| |
| int port = Integer.parseInt(zkAddress.substring(zkAddress.lastIndexOf(':') + 1)); |
| ZkServer zkServer = new ZkServer(dataDir, logDir, defaultNameSpace, port); |
| zkServer.start(); |
| |
| return zkServer; |
| } |
| |
| static public void stopZkServer(ZkServer zkServer) { |
| if (zkServer != null) { |
| zkServer.shutdown(); |
| System.out.println( |
| "Shut down zookeeper at port " + zkServer.getPort() + " in thread " + Thread |
| .currentThread().getName()); |
| } |
| } |
| |
| public static void setupEmptyCluster(HelixZkClient zkClient, String clusterName) { |
| ZKHelixAdmin admin = new ZKHelixAdmin(zkClient); |
| admin.addCluster(clusterName, true); |
| } |
| |
| /** |
| * convert T[] to set<T> |
| * @param s |
| * @return |
| */ |
| public static <T> Set<T> setOf(T... s) { |
| Set<T> set = new HashSet<T>(Arrays.asList(s)); |
| return set; |
| } |
| |
| /** |
| * generic method for verification with a timeout |
| * @param verifierName |
| * @param args |
| */ |
| public static void verifyWithTimeout(String verifierName, long timeout, Object... args) { |
| final long sleepInterval = 1000; // in ms |
| final int loop = (int) (timeout / sleepInterval) + 1; |
| try { |
| boolean result = false; |
| int i = 0; |
| for (; i < loop; i++) { |
| Thread.sleep(sleepInterval); |
| // verifier should be static method |
| result = (Boolean) TestHelper.getMethod(verifierName).invoke(null, args); |
| |
| if (result == true) { |
| break; |
| } |
| } |
| |
| // debug |
| // LOG.info(verifierName + ": wait " + ((i + 1) * 1000) + "ms to verify (" |
| // + result + ")"); |
| System.err.println( |
| verifierName + ": wait " + ((i + 1) * 1000) + "ms to verify " + " (" + result + ")"); |
| LOG.debug("args:" + Arrays.toString(args)); |
| // System.err.println("args:" + Arrays.toString(args)); |
| |
| if (result == false) { |
| LOG.error(verifierName + " fails"); |
| LOG.error("args:" + Arrays.toString(args)); |
| } |
| |
| Assert.assertTrue(result); |
| } catch (Exception e) { |
| LOG.error("Exception in verify: " + verifierName, e); |
| } |
| } |
| |
| private static Method getMethod(String name) { |
| Method[] methods = TestHelper.class.getMethods(); |
| for (Method method : methods) { |
| if (name.equals(method.getName())) { |
| return method; |
| } |
| } |
| return null; |
| } |
| |
| public static boolean verifyEmptyCurStateAndExtView(String clusterName, String resourceName, |
| Set<String> instanceNames, String zkAddr) { |
| HelixZkClient zkClient = SharedZkClientFactory.getInstance() |
| .buildZkClient(new HelixZkClient.ZkConnectionConfig(zkAddr)); |
| zkClient.setZkSerializer(new ZNRecordSerializer()); |
| |
| try { |
| ZKHelixDataAccessor accessor = |
| new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(zkClient)); |
| Builder keyBuilder = accessor.keyBuilder(); |
| |
| for (String instanceName : instanceNames) { |
| List<String> sessionIds = accessor.getChildNames(keyBuilder.sessions(instanceName)); |
| |
| for (String sessionId : sessionIds) { |
| CurrentState curState = |
| accessor.getProperty(keyBuilder.currentState(instanceName, sessionId, resourceName)); |
| |
| if (curState != null && curState.getRecord().getMapFields().size() != 0) { |
| return false; |
| } |
| |
| CurrentState taskCurState = |
| accessor.getProperty(keyBuilder.taskCurrentState(instanceName, sessionId, resourceName)); |
| |
| if (taskCurState != null && taskCurState.getRecord().getMapFields().size() != 0) { |
| return false; |
| } |
| } |
| |
| ExternalView extView = accessor.getProperty(keyBuilder.externalView(resourceName)); |
| |
| if (extView != null && extView.getRecord().getMapFields().size() != 0) { |
| return false; |
| } |
| } |
| |
| return true; |
| } finally { |
| zkClient.close(); |
| } |
| } |
| |
| public static boolean verifyNotConnected(HelixManager manager) { |
| return !manager.isConnected(); |
| } |
| |
| public static void setupCluster(String clusterName, String zkAddr, int startPort, |
| String participantNamePrefix, String resourceNamePrefix, int resourceNb, int partitionNb, |
| int nodesNb, int replica, String stateModelDef, boolean doRebalance) throws Exception { |
| TestHelper |
| .setupCluster(clusterName, zkAddr, startPort, participantNamePrefix, resourceNamePrefix, |
| resourceNb, partitionNb, nodesNb, replica, stateModelDef, RebalanceMode.SEMI_AUTO, |
| doRebalance); |
| } |
| |
| public static void setupCluster(String clusterName, String zkAddr, int startPort, |
| String participantNamePrefix, String resourceNamePrefix, int resourceNb, int partitionNb, |
| int nodesNb, int replica, String stateModelDef, RebalanceMode mode, boolean doRebalance) { |
| HelixZkClient zkClient = SharedZkClientFactory.getInstance() |
| .buildZkClient(new HelixZkClient.ZkConnectionConfig(zkAddr)); |
| try { |
| if (zkClient.exists("/" + clusterName)) { |
| LOG.warn("Cluster already exists:" + clusterName + ". Deleting it"); |
| zkClient.deleteRecursively("/" + clusterName); |
| } |
| |
| ClusterSetup setupTool = new ClusterSetup(zkAddr); |
| setupTool.addCluster(clusterName, true); |
| |
| for (int i = 0; i < nodesNb; i++) { |
| int port = startPort + i; |
| setupTool.addInstanceToCluster(clusterName, participantNamePrefix + "_" + port); |
| } |
| |
| for (int i = 0; i < resourceNb; i++) { |
| String resourceName = resourceNamePrefix + i; |
| setupTool.addResourceToCluster(clusterName, resourceName, partitionNb, stateModelDef, |
| mode.name(), |
| mode == RebalanceMode.FULL_AUTO ? CrushEdRebalanceStrategy.class.getName() |
| : RebalanceStrategy.DEFAULT_REBALANCE_STRATEGY); |
| if (doRebalance) { |
| setupTool.rebalanceStorageCluster(clusterName, resourceName, replica); |
| } |
| } |
| } finally { |
| zkClient.close(); |
| } |
| } |
| |
| public static void dropCluster(String clusterName, RealmAwareZkClient zkClient) { |
| ClusterSetup setupTool = new ClusterSetup(zkClient); |
| dropCluster(clusterName, zkClient, setupTool); |
| } |
| |
| public static void dropCluster(String clusterName, RealmAwareZkClient zkClient, ClusterSetup setup) { |
| String namespace = "/" + clusterName; |
| if (zkClient.exists(namespace)) { |
| try { |
| setup.deleteCluster(clusterName); |
| } catch (Exception ex) { |
| // Failed to delete, give some more time for connections to drop |
| try { |
| Thread.sleep(3000L); |
| setup.deleteCluster(clusterName); |
| } catch (Exception ignored) { |
| // OK - just ignore |
| } |
| } |
| } |
| } |
| |
| /** |
| * @param stateMap |
| * : "ResourceName/partitionKey" -> setOf(instances) |
| * @param state |
| * : MASTER|SLAVE|ERROR... |
| */ |
| public static void verifyState(String clusterName, String zkAddr, |
| Map<String, Set<String>> stateMap, String state) { |
| HelixZkClient zkClient = SharedZkClientFactory.getInstance() |
| .buildZkClient(new HelixZkClient.ZkConnectionConfig(zkAddr)); |
| zkClient.setZkSerializer(new ZNRecordSerializer()); |
| |
| try { |
| ZKHelixDataAccessor accessor = |
| new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(zkClient)); |
| Builder keyBuilder = accessor.keyBuilder(); |
| |
| for (String resGroupPartitionKey : stateMap.keySet()) { |
| Map<String, String> retMap = getResourceAndPartitionKey(resGroupPartitionKey); |
| String resGroup = retMap.get("RESOURCE"); |
| String partitionKey = retMap.get("PARTITION"); |
| |
| ExternalView extView = accessor.getProperty(keyBuilder.externalView(resGroup)); |
| for (String instance : stateMap.get(resGroupPartitionKey)) { |
| String actualState = extView.getStateMap(partitionKey).get(instance); |
| Assert.assertNotNull(actualState, |
| "externalView doesn't contain state for " + resGroup + "/" + partitionKey + " on " |
| + instance + " (expect " + state + ")"); |
| |
| Assert.assertEquals(actualState, state, |
| "externalView for " + resGroup + "/" + partitionKey + " on " + instance + " is " |
| + actualState + " (expect " + state + ")"); |
| } |
| } |
| } finally { |
| zkClient.close(); |
| } |
| } |
| |
| /** |
| * @param resourcePartition |
| * : key is in form of "resource/partitionKey" or "resource_x" |
| * @return |
| */ |
| private static Map<String, String> getResourceAndPartitionKey(String resourcePartition) { |
| String resourceName; |
| String partitionName; |
| int idx = resourcePartition.indexOf('/'); |
| if (idx > -1) { |
| resourceName = resourcePartition.substring(0, idx); |
| partitionName = resourcePartition.substring(idx + 1); |
| } else { |
| idx = resourcePartition.lastIndexOf('_'); |
| resourceName = resourcePartition.substring(0, idx); |
| partitionName = resourcePartition; |
| } |
| |
| Map<String, String> retMap = new HashMap<String, String>(); |
| retMap.put("RESOURCE", resourceName); |
| retMap.put("PARTITION", partitionName); |
| return retMap; |
| } |
| |
| public static <T> Map<String, T> startThreadsConcurrently(final int nrThreads, |
| final Callable<T> method, final long timeout) { |
| final CountDownLatch startLatch = new CountDownLatch(1); |
| final CountDownLatch finishCounter = new CountDownLatch(nrThreads); |
| final Map<String, T> resultsMap = new ConcurrentHashMap<String, T>(); |
| final List<Thread> threadList = new ArrayList<Thread>(); |
| |
| for (int i = 0; i < nrThreads; i++) { |
| Thread thread = new Thread() { |
| @Override |
| public void run() { |
| try { |
| boolean isTimeout = !startLatch.await(timeout, TimeUnit.SECONDS); |
| if (isTimeout) { |
| LOG.error("Timeout while waiting for start latch"); |
| } |
| } catch (InterruptedException ex) { |
| LOG.error("Interrupted while waiting for start latch"); |
| } |
| |
| try { |
| T result = method.call(); |
| if (result != null) { |
| resultsMap.put("thread_" + this.getId(), result); |
| } |
| LOG.debug("result=" + result); |
| } catch (Exception e) { |
| LOG.error("Exeption in executing " + method.getClass().getName(), e); |
| } |
| |
| finishCounter.countDown(); |
| } |
| }; |
| threadList.add(thread); |
| thread.start(); |
| } |
| startLatch.countDown(); |
| |
| // wait for all thread to complete |
| try { |
| boolean isTimeout = !finishCounter.await(timeout, TimeUnit.SECONDS); |
| if (isTimeout) { |
| LOG.error("Timeout while waiting for finish latch. Interrupt all threads"); |
| for (Thread thread : threadList) { |
| thread.interrupt(); |
| } |
| } |
| } catch (InterruptedException e) { |
| LOG.error("Interrupted while waiting for finish latch", e); |
| } |
| |
| return resultsMap; |
| } |
| |
| public static Message createMessage(String msgId, String fromState, String toState, |
| String tgtName, String resourceName, String partitionName) { |
| Message msg = new Message(MessageType.STATE_TRANSITION, msgId); |
| msg.setFromState(fromState); |
| msg.setToState(toState); |
| msg.setTgtName(tgtName); |
| msg.setResourceName(resourceName); |
| msg.setPartitionName(partitionName); |
| msg.setStateModelDef("MasterSlave"); |
| |
| return msg; |
| } |
| |
| public static String getTestMethodName() { |
| StackTraceElement[] calls = Thread.currentThread().getStackTrace(); |
| return calls[2].getMethodName(); |
| } |
| |
| public static String getTestClassName() { |
| StackTraceElement[] calls = Thread.currentThread().getStackTrace(); |
| String fullClassName = calls[2].getClassName(); |
| return fullClassName.substring(fullClassName.lastIndexOf('.') + 1); |
| } |
| |
| public static <T> Map<String, T> startThreadsConcurrently(final List<Callable<T>> methods, |
| final long timeout) { |
| final int nrThreads = methods.size(); |
| final CountDownLatch startLatch = new CountDownLatch(1); |
| final CountDownLatch finishCounter = new CountDownLatch(nrThreads); |
| final Map<String, T> resultsMap = new ConcurrentHashMap<String, T>(); |
| final List<Thread> threadList = new ArrayList<Thread>(); |
| |
| for (int i = 0; i < nrThreads; i++) { |
| final Callable<T> method = methods.get(i); |
| |
| Thread thread = new Thread() { |
| @Override |
| public void run() { |
| try { |
| boolean isTimeout = !startLatch.await(timeout, TimeUnit.SECONDS); |
| if (isTimeout) { |
| LOG.error("Timeout while waiting for start latch"); |
| } |
| } catch (InterruptedException ex) { |
| LOG.error("Interrupted while waiting for start latch"); |
| } |
| |
| try { |
| T result = method.call(); |
| if (result != null) { |
| resultsMap.put("thread_" + this.getId(), result); |
| } |
| LOG.debug("result=" + result); |
| } catch (Exception e) { |
| LOG.error("Exeption in executing " + method.getClass().getName(), e); |
| } |
| |
| finishCounter.countDown(); |
| } |
| }; |
| threadList.add(thread); |
| thread.start(); |
| } |
| startLatch.countDown(); |
| |
| // wait for all thread to complete |
| try { |
| boolean isTimeout = !finishCounter.await(timeout, TimeUnit.SECONDS); |
| if (isTimeout) { |
| LOG.error("Timeout while waiting for finish latch. Interrupt all threads"); |
| for (Thread thread : threadList) { |
| thread.interrupt(); |
| } |
| } |
| } catch (InterruptedException e) { |
| LOG.error("Interrupted while waiting for finish latch", e); |
| } |
| |
| return resultsMap; |
| } |
| |
| public static void printCache(Map<String, ZNode> cache) { |
| System.out.println("START:Print cache"); |
| TreeMap<String, ZNode> map = new TreeMap<String, ZNode>(); |
| map.putAll(cache); |
| |
| for (String key : map.keySet()) { |
| ZNode node = map.get(key); |
| TreeSet<String> childSet = new TreeSet<String>(); |
| childSet.addAll(node.getChildSet()); |
| System.out.print( |
| key + "=" + node.getData() + ", " + childSet + ", " + (node.getStat() == null ? "null\n" |
| : node.getStat())); |
| } |
| System.out.println("END:Print cache"); |
| } |
| |
| public static void readZkRecursive(String path, Map<String, ZNode> map, HelixZkClient zkclient) { |
| try { |
| Stat stat = new Stat(); |
| ZNRecord record = zkclient.readData(path, stat); |
| List<String> childNames = zkclient.getChildren(path); |
| ZNode node = new ZNode(path, record, stat); |
| node.addChildren(childNames); |
| map.put(path, node); |
| |
| for (String childName : childNames) { |
| String childPath = path + "/" + childName; |
| readZkRecursive(childPath, map, zkclient); |
| } |
| } catch (ZkNoNodeException e) { |
| // OK |
| } |
| } |
| |
| public static void readZkRecursive(String path, Map<String, ZNode> map, |
| BaseDataAccessor<ZNRecord> zkAccessor) { |
| try { |
| Stat stat = new Stat(); |
| ZNRecord record = zkAccessor.get(path, stat, 0); |
| List<String> childNames = zkAccessor.getChildNames(path, 0); |
| // System.out.println("childNames: " + childNames); |
| ZNode node = new ZNode(path, record, stat); |
| node.addChildren(childNames); |
| map.put(path, node); |
| |
| if (childNames != null && !childNames.isEmpty()) { |
| for (String childName : childNames) { |
| String childPath = path + "/" + childName; |
| readZkRecursive(childPath, map, zkAccessor); |
| } |
| } |
| } catch (ZkNoNodeException e) { |
| // OK |
| } |
| } |
| |
| public static boolean verifyZkCache(List<String> paths, BaseDataAccessor<ZNRecord> zkAccessor, |
| HelixZkClient zkclient, boolean needVerifyStat) { |
| // read everything |
| Map<String, ZNode> zkMap = new HashMap<String, ZNode>(); |
| Map<String, ZNode> cache = new HashMap<String, ZNode>(); |
| for (String path : paths) { |
| readZkRecursive(path, zkMap, zkclient); |
| readZkRecursive(path, cache, zkAccessor); |
| } |
| // printCache(map); |
| |
| return verifyZkCache(paths, null, cache, zkMap, needVerifyStat); |
| } |
| |
| public static boolean verifyZkCache(List<String> paths, Map<String, ZNode> cache, |
| HelixZkClient zkclient, boolean needVerifyStat) { |
| return verifyZkCache(paths, null, cache, zkclient, needVerifyStat); |
| } |
| |
| public static boolean verifyZkCache(List<String> paths, List<String> pathsExcludeForStat, |
| Map<String, ZNode> cache, HelixZkClient zkclient, boolean needVerifyStat) { |
| // read everything on zk under paths |
| Map<String, ZNode> zkMap = new HashMap<String, ZNode>(); |
| for (String path : paths) { |
| readZkRecursive(path, zkMap, zkclient); |
| } |
| // printCache(map); |
| |
| return verifyZkCache(paths, pathsExcludeForStat, cache, zkMap, needVerifyStat); |
| } |
| |
| public static boolean verifyZkCache(List<String> paths, List<String> pathsExcludeForStat, |
| Map<String, ZNode> cache, Map<String, ZNode> zkMap, boolean needVerifyStat) { |
| // equal size |
| if (zkMap.size() != cache.size()) { |
| System.err |
| .println("size mismatch: cacheSize: " + cache.size() + ", zkMapSize: " + zkMap.size()); |
| System.out.println("cache: (" + cache.size() + ")"); |
| TestHelper.printCache(cache); |
| |
| System.out.println("zkMap: (" + zkMap.size() + ")"); |
| TestHelper.printCache(zkMap); |
| |
| return false; |
| } |
| |
| // everything in cache is also in map |
| for (String path : cache.keySet()) { |
| ZNode cacheNode = cache.get(path); |
| ZNode zkNode = zkMap.get(path); |
| |
| if (zkNode == null) { |
| // in cache but not on zk |
| System.err.println("path: " + path + " in cache but not on zk: inCacheNode: " + cacheNode); |
| return false; |
| } |
| |
| if ((zkNode.getData() == null && cacheNode.getData() != null) || (zkNode.getData() != null |
| && cacheNode.getData() == null) || (zkNode.getData() != null |
| && cacheNode.getData() != null && !zkNode.getData().equals(cacheNode.getData()))) { |
| // data not equal |
| System.err.println( |
| "data mismatch on path: " + path + ", inCache: " + cacheNode.getData() + ", onZk: " |
| + zkNode.getData()); |
| return false; |
| } |
| |
| if ((zkNode.getChildSet() == null && cacheNode.getChildSet() != null) || ( |
| zkNode.getChildSet() != null && cacheNode.getChildSet() == null) || ( |
| zkNode.getChildSet() != null && cacheNode.getChildSet() != null && !zkNode.getChildSet() |
| .equals(cacheNode.getChildSet()))) { |
| // childSet not equal |
| System.err.println( |
| "childSet mismatch on path: " + path + ", inCache: " + cacheNode.getChildSet() |
| + ", onZk: " + zkNode.getChildSet()); |
| return false; |
| } |
| |
| if (needVerifyStat && pathsExcludeForStat != null && !pathsExcludeForStat.contains(path)) { |
| if (cacheNode.getStat() == null || !zkNode.getStat().equals(cacheNode.getStat())) { |
| // stat not equal |
| System.err.println( |
| "Stat mismatch on path: " + path + ", inCache: " + cacheNode.getStat() + ", onZk: " |
| + zkNode.getStat()); |
| return false; |
| } |
| } |
| } |
| |
| return true; |
| } |
| |
| public static StateModelDefinition generateStateModelDefForBootstrap() { |
| ZNRecord record = new ZNRecord("Bootstrap"); |
| record.setSimpleField(StateModelDefinitionProperty.INITIAL_STATE.toString(), "IDLE"); |
| List<String> statePriorityList = new ArrayList<String>(); |
| statePriorityList.add("ONLINE"); |
| statePriorityList.add("BOOTSTRAP"); |
| statePriorityList.add("OFFLINE"); |
| statePriorityList.add("IDLE"); |
| statePriorityList.add("DROPPED"); |
| statePriorityList.add("ERROR"); |
| record.setListField(StateModelDefinitionProperty.STATE_PRIORITY_LIST.toString(), |
| statePriorityList); |
| for (String state : statePriorityList) { |
| String key = state + ".meta"; |
| Map<String, String> metadata = new HashMap<String, String>(); |
| if (state.equals("ONLINE")) { |
| metadata.put("count", "R"); |
| record.setMapField(key, metadata); |
| } else if (state.equals("BOOTSTRAP")) { |
| metadata.put("count", "-1"); |
| record.setMapField(key, metadata); |
| } else if (state.equals("OFFLINE")) { |
| metadata.put("count", "-1"); |
| record.setMapField(key, metadata); |
| } else if (state.equals("IDLE")) { |
| metadata.put("count", "-1"); |
| record.setMapField(key, metadata); |
| } else if (state.equals("DROPPED")) { |
| metadata.put("count", "-1"); |
| record.setMapField(key, metadata); |
| } else if (state.equals("ERROR")) { |
| metadata.put("count", "-1"); |
| record.setMapField(key, metadata); |
| } |
| } |
| |
| for (String state : statePriorityList) { |
| String key = state + ".next"; |
| if (state.equals("ONLINE")) { |
| Map<String, String> metadata = new HashMap<String, String>(); |
| metadata.put("BOOTSTRAP", "OFFLINE"); |
| metadata.put("OFFLINE", "OFFLINE"); |
| metadata.put("DROPPED", "OFFLINE"); |
| metadata.put("IDLE", "OFFLINE"); |
| record.setMapField(key, metadata); |
| } else if (state.equals("BOOTSTRAP")) { |
| Map<String, String> metadata = new HashMap<String, String>(); |
| metadata.put("ONLINE", "ONLINE"); |
| metadata.put("OFFLINE", "OFFLINE"); |
| metadata.put("DROPPED", "OFFLINE"); |
| metadata.put("IDLE", "OFFLINE"); |
| record.setMapField(key, metadata); |
| } else if (state.equals("OFFLINE")) { |
| Map<String, String> metadata = new HashMap<String, String>(); |
| metadata.put("ONLINE", "BOOTSTRAP"); |
| metadata.put("BOOTSTRAP", "BOOTSTRAP"); |
| metadata.put("DROPPED", "IDLE"); |
| metadata.put("IDLE", "IDLE"); |
| record.setMapField(key, metadata); |
| } else if (state.equals("IDLE")) { |
| Map<String, String> metadata = new HashMap<String, String>(); |
| metadata.put("ONLINE", "OFFLINE"); |
| metadata.put("BOOTSTRAP", "OFFLINE"); |
| metadata.put("OFFLINE", "OFFLINE"); |
| metadata.put("DROPPED", "DROPPED"); |
| record.setMapField(key, metadata); |
| } else if (state.equals("ERROR")) { |
| Map<String, String> metadata = new HashMap<String, String>(); |
| metadata.put("IDLE", "IDLE"); |
| record.setMapField(key, metadata); |
| } |
| } |
| List<String> stateTransitionPriorityList = new ArrayList<String>(); |
| stateTransitionPriorityList.add("ONLINE-OFFLINE"); |
| stateTransitionPriorityList.add("BOOTSTRAP-ONLINE"); |
| stateTransitionPriorityList.add("OFFLINE-BOOTSTRAP"); |
| stateTransitionPriorityList.add("BOOTSTRAP-OFFLINE"); |
| stateTransitionPriorityList.add("OFFLINE-IDLE"); |
| stateTransitionPriorityList.add("IDLE-OFFLINE"); |
| stateTransitionPriorityList.add("IDLE-DROPPED"); |
| stateTransitionPriorityList.add("ERROR-IDLE"); |
| record.setListField(StateModelDefinitionProperty.STATE_TRANSITION_PRIORITYLIST.toString(), |
| stateTransitionPriorityList); |
| return new StateModelDefinition(record); |
| } |
| |
| public static String znrecordToString(ZNRecord record) { |
| StringBuffer sb = new StringBuffer(); |
| sb.append(record.getId() + "\n"); |
| Map<String, String> simpleFields = record.getSimpleFields(); |
| if (simpleFields != null) { |
| sb.append("simpleFields\n"); |
| for (String key : simpleFields.keySet()) { |
| sb.append(" " + key + "\t: " + simpleFields.get(key) + "\n"); |
| } |
| } |
| |
| Map<String, List<String>> listFields = record.getListFields(); |
| sb.append("listFields\n"); |
| for (String key : listFields.keySet()) { |
| List<String> list = listFields.get(key); |
| sb.append(" " + key + "\t: "); |
| for (String listValue : list) { |
| sb.append(listValue + ", "); |
| } |
| sb.append("\n"); |
| } |
| |
| Map<String, Map<String, String>> mapFields = record.getMapFields(); |
| sb.append("mapFields\n"); |
| for (String key : mapFields.keySet()) { |
| Map<String, String> map = mapFields.get(key); |
| sb.append(" " + key + "\t: \n"); |
| for (String mapKey : map.keySet()) { |
| sb.append(" " + mapKey + "\t: " + map.get(mapKey) + "\n"); |
| } |
| } |
| |
| return sb.toString(); |
| } |
| |
| public interface Verifier { |
| boolean verify() throws Exception; |
| } |
| |
| public static boolean verify(Verifier verifier, long timeout) throws Exception { |
| long start = System.currentTimeMillis(); |
| do { |
| boolean result = verifier.verify(); |
| boolean isTimedout = (System.currentTimeMillis() - start) > timeout; |
| if (result || isTimedout) { |
| if (isTimedout && !result) { |
| LOG.error("verifier time out, consider try longer timeout, stack trace{}", |
| Arrays.asList(Thread.currentThread().getStackTrace())); |
| } |
| return result; |
| } |
| Thread.sleep(50); |
| } while (true); |
| } |
| |
| // debug code |
| public static String printHandlers(ZkTestManager manager) { |
| StringBuilder sb = new StringBuilder(); |
| List<CallbackHandler> handlers = manager.getHandlers(); |
| sb.append(manager.getInstanceName() + " has " + handlers.size() + " cb-handlers. ["); |
| |
| for (int i = 0; i < handlers.size(); i++) { |
| CallbackHandler handler = handlers.get(i); |
| String path = handler.getPath(); |
| sb.append( |
| path.substring(manager.getClusterName().length() + 1) + ": " + handler.getListener()); |
| if (i < (handlers.size() - 1)) { |
| sb.append(", "); |
| } |
| } |
| sb.append("]"); |
| |
| return sb.toString(); |
| } |
| |
| public static void printZkListeners(HelixZkClient client) throws Exception { |
| Map<String, Set<IZkDataListener>> datalisteners = ZkTestHelper.getZkDataListener(client); |
| Map<String, Set<IZkChildListener>> childListeners = ZkTestHelper.getZkChildListener(client); |
| |
| System.out.println("dataListeners {"); |
| for (String path : datalisteners.keySet()) { |
| System.out.println("\t" + path + ": "); |
| Set<IZkDataListener> set = datalisteners.get(path); |
| for (IZkDataListener listener : set) { |
| CallbackHandler handler = (CallbackHandler) listener; |
| System.out.println("\t\t" + handler.getListener()); |
| } |
| } |
| System.out.println("}"); |
| |
| System.out.println("childListeners {"); |
| for (String path : childListeners.keySet()) { |
| System.out.println("\t" + path + ": "); |
| Set<IZkChildListener> set = childListeners.get(path); |
| for (IZkChildListener listener : set) { |
| CallbackHandler handler = (CallbackHandler) listener; |
| System.out.println("\t\t" + handler.getListener()); |
| } |
| } |
| System.out.println("}"); |
| } |
| } |
