public static final int DEFAULT_SEG_DATA_BLOCKS_COUNT = 1 << 5; public static final int DEFAULT_TREE_ID = 1; public static final int ROOT_NODE = 0; public static final SegIdProviderTest SEG_ID_PROVIDER = new SegIdProviderTest(); public static final SimpleTreeIdProvider TREE_ID_PROVIDER = new SimpleTreeIdProvider(); public static HTreeComponents createHashTree(int noOfSegDataBlocks,
		boolean enabledNonBlockingCalls,
		final HashTreesIdProvider treeIdProv,
		final SegmentIdProvider segIdPro, final HashTreesStore hTStore) {
	SimpleMemStore store = new SimpleMemStore();
	HashTreesImpl hTree = new HashTreesImpl.Builder(store, treeIdProv,
			hTStore).setNoOfSegments(noOfSegDataBlocks)
			.setEnabledNonBlockingCalls(enabledNonBlockingCalls)
			.setSegmentIdProvider(segIdPro).build();
	store.registerHashTrees(hTree);
	return new HTreeComponents(hTStore, store, hTree);
} public static HashTreesStore[] generateInMemoryAndPersistentStores()
		throws IOException {
	HashTreesStore[] stores = new HashTreesStore[2];
	stores[0] = generateInMemoryStore();
	stores[1] = generatePersistentStore();
	return stores;
} public static HashTreesMemStore generateInMemoryStore() {
	return new HashTreesMemStore();
} public static HashTreesPersistentStore generatePersistentStore()
		throws IOException {
	return new HashTreesPersistentStore(randomDirName());
} public static ByteBuffer randomByteBuffer() {
	byte[] random = new byte[8];
	RANDOM.nextBytes(random);
	return ByteBuffer.wrap(random);
} public static byte[] randomBytes() {
	byte[] emptyBuffer = new byte[8];
	RANDOM.nextBytes(emptyBuffer);
	return emptyBuffer;
} public interface HashTrees {

	/**
	 * Adds the (key,value) pair to the store. Intended to be used while synch
	 * operation.
	 * 
	 * @param keyValuePairs
	 */
	void sPut(List<KeyValue> keyValuePairs) throws IOException;

	/**
	 * Deletes the keys from the store. Intended to be used while synch
	 * operation.
	 * 
	 * @param keys
	 * 
	 */
	void sRemove(List<ByteBuffer> keys) throws IOException;

	/**
	 * Hash tree internal nodes store the hash of their children nodes. Given a
	 * set of internal node ids, this returns the hashes that are stored on the
	 * internal node.
	 * 
	 * @param treeId
	 * @param nodeIds
	 *            , internal tree node ids.
	 * @return
	 * 
	 */
	List<SegmentHash> getSegmentHashes(long treeId, List<Integer> nodeIds)
			throws IOException;

	/**
	 * Returns the segment hash that is stored on the tree.
	 * 
	 * @param treeId
	 *            , hash tree id.
	 * @param nodeId
	 *            , node id
	 * @return
	 * 
	 */
	SegmentHash getSegmentHash(long treeId, int nodeId) throws IOException;

	/**
	 * Hash tree data is stored on the leaf blocks. Given a segment id this
	 * method is supposed to return (key,hash) pairs.
	 * 
	 * @param treeId
	 * @param segId
	 *            , id of the segment block.
	 * @return
	 * 
	 */
	List<SegmentData> getSegment(long treeId, int segId) throws IOException;

	/**
	 * Returns the (key,digest) for the given key in the given segment.
	 * 
	 * 
	 * @param treeId
	 * @param segId
	 * @param key
	 */
	SegmentData getSegmentData(long treeId, int segId, ByteBuffer key)
			throws IOException;

	/**
	 * Deletes the given tree node.
	 * 
	 * @param treeId
	 * @param nodeId
	 * @throws IOException
	 */
	void deleteTreeNode(long treeId, int nodeId) throws IOException;

	/**
	 * Adds the key, and digest of value to the segment block in HashTree.
	 * 
	 * @param key
	 * @param value
	 * @throws IOException
	 */
	void hPut(ByteBuffer key, ByteBuffer value) throws IOException;

	/**
	 * Deletes the key from the hash tree.
	 * 
	 * @param key
	 * @throws IOException
	 */
	void hRemove(ByteBuffer key) throws IOException;

	/**
	 * Updates the other HTree based on the differences with local objects. This
	 * function should be running on primary to synch with other replicas, and
	 * not the other way.
	 * 
	 * @param remoteTree
	 * @param syncType
	 * @return, gives stats about total number of differences between local node
	 *          and remote node.
	 */
	SyncDiffResult synch(long treeId, HashTrees remoteTree, SyncType syncType)
			throws IOException;

	/**
	 * Same as {@link #synch(long, HashTrees, SyncType)} except that, syncType
	 * is passed as SyncType.UPDATE.
	 * 
	 * @param treeId
	 * @param remoteTree
	 * @return
	 */
	SyncDiffResult synch(long treeId, HashTrees remoteTree) throws IOException;

	/**
	 * Updates segment hashes based on the dirty entries.
	 * 
	 * @param treeId
	 * @param fullRebuildPeriod
	 *            , indicates a time interval in milliseconds, and if a tree is
	 *            not fully rebuilt for more than this value, a full rebuild
	 *            will be triggered. Otherwise only dirty segments will be
	 *            updated. If there is no information about when was the last
	 *            time the tree was fully rebuilt, then a full rebuild will be
	 *            triggered. A negative value indicates disabling full rebuild.
	 * @return, returns how many dirty segments are marked before the rebuild.
	 */
	int rebuildHashTree(long treeId, long fullRebuildPeriod) throws IOException;

	/**
	 * Similar to {@link #rebuildHashTree(long, long)}, except that allows to
	 * specify whether a full rebuild should be triggered or not.
	 * 
	 * @param treeId
	 * @param fullRebuild
	 * @return, returns how many dirty segments are marked before the rebuild.
	 * @throws IOException
	 */
	int rebuildHashTree(long treeId, boolean fullRebuild) throws IOException;

	void addObserver(HashTreesObserver observer) throws IOException;

	void removeObserver(HashTreesObserver observer) throws IOException;
} @ThreadSafe
public class HashTreesImpl implements HashTrees, Service {

	/**
	 * Specifies how much element can be queued when hash tree is backed by a
	 * non blocking queue. When the queue is full, the new puts or removes are
	 * rejected.
	 */
	public final static int DEFAULT_NB_QUE_SIZE = Integer.MAX_VALUE;
	/**
	 * Specifies at most how many segments can be used by a single HashTree.
	 */
	public final static int MAX_NO_OF_SEGMENTS = 1 << 30;

	private final static Logger LOGGER = LoggerFactory
			.getLogger(HashTreesImpl.class.getName());
	private final static char COMMA_DELIMETER = ',';
	private final static char NEW_LINE_DELIMETER = '\n';
	private final static int ROOT_NODE = 0;
	private final static int BINARY_TREE = 2;

	private final int noOfChildren;
	private final int internalNodesCount;
	private final int segmentsCount;
	private final int height;

	private final Store store;
	private final HashTreesStore htStore;
	private final HashTreesIdProvider treeIdProvider;
	private final SegmentIdProvider segIdProvider;

	private final boolean enabledNonBlockingCalls;
	private final int nonBlockingQueueSize;
	private final Object nonBlockingCallsLock = new Object();
	@LockedBy("nonBlockingCallsLock")
	private volatile NonBlockingHTDataUpdater bgDataUpdater;
	private final HashTreesObserverNotifier notifier = new HashTreesObserverNotifier();
	private final LockProvider lockProvider;

	public HashTreesImpl(int noOfSegments, boolean enabledNonBlockingCalls,
			int nonBlockingQueueSize, final HashTreesIdProvider treeIdProvider,
			final SegmentIdProvider segIdProvider,
			final HashTreesStore htStore, final Store store,
			final LockProvider lockProvider) {
		this.noOfChildren = BINARY_TREE;
		this.segmentsCount = getValidSegmentsCount(noOfSegments);
		this.enabledNonBlockingCalls = enabledNonBlockingCalls;
		this.nonBlockingQueueSize = nonBlockingQueueSize;
		this.height = height(this.segmentsCount, noOfChildren);
		this.internalNodesCount = getNoOfNodes((height - 1), noOfChildren);
		this.treeIdProvider = treeIdProvider;
		this.segIdProvider = segIdProvider;
		this.htStore = htStore;
		this.store = store;
		this.lockProvider = lockProvider;
	}

	@Override
	public void hPut(final ByteBuffer key, final ByteBuffer value)
			throws IOException {
		hPutInternal(HTOperation.PUT, key, value);
	}

	private void hPutInternal(HTOperation operation, final ByteBuffer key,
			final ByteBuffer value) throws IOException {
		if (enabledNonBlockingCalls) {
			List<ByteBuffer> input = new ArrayList<ByteBuffer>();
			input.add(key);
			input.add(value);
			bgDataUpdater.enque(Pair.create(operation, input));
		} else
			hPutInternal(key, value);
	}

	private void hPutInternal(final ByteBuffer key, final ByteBuffer value)
			throws IOException {
		notifier.preHPut(key, value);
		long treeId = treeIdProvider.getTreeId(key.array());
		int segId = segIdProvider.getSegmentId(key.array());
		ByteBuffer digest = ByteBuffer.wrap(sha1(value.array()));
		htStore.setDirtySegment(treeId, segId);
		htStore.putSegmentData(treeId, segId, key, digest);
		notifier.postHPut(key, value);
	}

	@Override
	public void hRemove(final ByteBuffer key) throws IOException {
		hRemoveInternal(HTOperation.REMOVE, key);
	}

	private void hRemoveInternal(HTOperation operation, final ByteBuffer key)
			throws IOException {
		if (enabledNonBlockingCalls) {
			List<ByteBuffer> input = new ArrayList<ByteBuffer>();
			input.add(key);
			bgDataUpdater.enque(Pair.create(operation, input));
		} else {
			hRemoveInternal(key);
		}
	}

	private void hRemoveInternal(final ByteBuffer key) throws IOException {
		notifier.preHRemove(key);
		long treeId = treeIdProvider.getTreeId(key.array());
		int segId = segIdProvider.getSegmentId(key.array());
		htStore.setDirtySegment(treeId, segId);
		htStore.deleteSegmentData(treeId, segId, key);
		notifier.postHRemove(key);
	}

	@Override
	public SyncDiffResult synch(long treeId, final HashTrees remoteTree)
			throws IOException {
		return synch(treeId, remoteTree, SyncType.UPDATE);
	}

	@Override
	public SyncDiffResult synch(long treeId, final HashTrees remoteTree,
			SyncType syncType) throws IOException {
		if (lockProvider.acquireLock(treeId)) {
			try {
				boolean doUpdate = (syncType == SyncType.UPDATE) ? true : false;

				PeekingIterator<SegmentHash> localItr = null, remoteItr = null;
				SegmentHash local, remote;

				List<Integer> pQueue = new ArrayList<Integer>();
				pQueue.add(ROOT_NODE);

				int totKeyDifferences = 0, totExtrinsicSegments = 0;

				while (!pQueue.isEmpty()) {

					localItr = Iterators.peekingIterator(getSegmentHashes(
							treeId, pQueue).iterator());
					remoteItr = Iterators.peekingIterator(remoteTree
							.getSegmentHashes(treeId, pQueue).iterator());
					pQueue = new ArrayList<Integer>();

					while (localItr.hasNext() || remoteItr.hasNext()) {
						local = localItr.hasNext() ? localItr.peek() : null;
						remote = remoteItr.hasNext() ? remoteItr.peek() : null;

						int compareRes = compareSegNodeIds(local, remote);

						if (compareRes == 0) {
							if (!Arrays.equals(local.getHash(),
									remote.getHash())) {
								if (isLeafNode(local.getNodeId())) {
									totKeyDifferences += syncSegment(treeId,
											getSegmentIdFromLeafId(local
													.getNodeId()), remoteTree,
											doUpdate);
								} else
									pQueue.addAll(getImmediateChildren(
											local.getNodeId(), noOfChildren));

							}
							localItr.next();
							remoteItr.next();
						} else if (compareRes < 0) {
							totKeyDifferences += updateRemoteTreeWithMissingSegment(
									treeId, local.getNodeId(), remoteTree,
									doUpdate);
							localItr.next();
						} else {
							if (doUpdate)
								remoteTree.deleteTreeNode(treeId,
										remote.getNodeId());
							remoteItr.next();
							totExtrinsicSegments += 1;
						}
					}
				}
				return new SyncDiffResult(totKeyDifferences,
						totExtrinsicSegments);
			} finally {
				lockProvider.releaseLock(treeId);
			}
		}
		return new SyncDiffResult(0, 0);
	}

	private int syncSegment(long treeId, int segId, HashTrees remoteTree,
			boolean doUpdate) throws IOException {
		PeekingIterator<SegmentData> localDataItr = Iterators
				.peekingIterator(getSegment(treeId, segId).iterator());
		PeekingIterator<SegmentData> remoteDataItr = Iterators
				.peekingIterator(remoteTree.getSegment(treeId, segId)
						.iterator());

		List<KeyValue> kvsForAddition = new ArrayList<KeyValue>();
		List<ByteBuffer> keysForRemoval = new ArrayList<ByteBuffer>();

		SegmentData local, remote;
		while (localDataItr.hasNext() || remoteDataItr.hasNext()) {
			local = localDataItr.hasNext() ? localDataItr.peek() : null;
			remote = remoteDataItr.hasNext() ? remoteDataItr.peek() : null;

			int compRes = compareSegmentKeys(local, remote);
			if (compRes == 0) {
				if (!Arrays.equals(local.getDigest(), remote.getDigest())) {
					ByteBuffer key = ByteBuffer.wrap(local.getKey());
					byte[] value = store.get(local.getKey());
					if (value != null)
						kvsForAddition.add(new KeyValue(key, ByteBuffer
								.wrap(value)));
				}
				localDataItr.next();
				remoteDataItr.next();
			} else if (compRes < 0) {
				ByteBuffer key = ByteBuffer.wrap(local.getKey());
				byte[] value = store.get(local.getKey());
				if (value != null)
					kvsForAddition
							.add(new KeyValue(key, ByteBuffer.wrap(value)));
				localDataItr.next();
			} else {
				keysForRemoval.add(ByteBuffer.wrap(remote.getKey()));
				remoteDataItr.next();
			}
		}

		if (doUpdate) {
			if (kvsForAddition.size() > 0)
				remoteTree.sPut(kvsForAddition);
			if (keysForRemoval.size() > 0)
				remoteTree.sRemove(keysForRemoval);
		}

		return kvsForAddition.size() + keysForRemoval.size();
	}

	private int updateRemoteTreeWithMissingSegment(long treeId,
			int rootMissingNodeId, HashTrees remoteTree, boolean doUpdate)
			throws IOException {
		final List<KeyValue> keyValuePairs = new ArrayList<>();
		int maxSizeToTransfer = 5000;
		int totKeyUpdates = 0;
		int leftMostNodeId = getSegmentIdFromLeafId(getLeftMostChildNode(
				rootMissingNodeId, noOfChildren, height));
		int rightMostNodeId = getSegmentIdFromLeafId(getRightMostChildNode(
				rootMissingNodeId, noOfChildren, height));
		Iterator<SegmentData> sdItr = htStore.getSegmentDataIterator(treeId,
				leftMostNodeId, rightMostNodeId);
		while (sdItr.hasNext()) {
			SegmentData sd = sdItr.next();
			byte[] value = store.get(sd.getKey());
			if (value != null)
				keyValuePairs.add(new KeyValue(ByteBuffer.wrap(sd.getKey()),
						ByteBuffer.wrap(value)));
			if (keyValuePairs.size() > maxSizeToTransfer) {
				if (doUpdate)
					remoteTree.sPut(keyValuePairs);
				totKeyUpdates += keyValuePairs.size();
				keyValuePairs.clear();
			}
		}
		if (keyValuePairs.size() > 0) {
			if (doUpdate)
				remoteTree.sPut(keyValuePairs);
			totKeyUpdates += keyValuePairs.size();
			keyValuePairs.clear();
		}
		return totKeyUpdates;
	}

	@Override
	public SegmentHash getSegmentHash(long treeId, int nodeId)
			throws IOException {
		return htStore.getSegmentHash(treeId, nodeId);
	}

	@Override
	public List<SegmentHash> getSegmentHashes(long treeId,
			final List<Integer> nodeIds) throws IOException {
		return htStore.getSegmentHashes(treeId, nodeIds);
	}

	@Override
	public SegmentData getSegmentData(long treeId, int segId, ByteBuffer key)
			throws IOException {
		return htStore.getSegmentData(treeId, segId, key);
	}

	@Override
	public List<SegmentData> getSegment(long treeId, int segId)
			throws IOException {
		return htStore.getSegment(treeId, segId);
	}

	@Override
	public int rebuildHashTree(long treeId, long fullRebuildPeriod)
			throws IOException {
		long lastFullRebuiltTime = htStore.getCompleteRebuiltTimestamp(treeId);
		boolean fullRebuild = (lastFullRebuiltTime == 0) ? true
				: ((fullRebuildPeriod < 0) ? false
						: (System.currentTimeMillis() - lastFullRebuiltTime) > fullRebuildPeriod);
		return rebuildHashTree(treeId, fullRebuild);
	}

	@Override
	public int rebuildHashTree(long treeId, boolean fullRebuild)
			throws IOException {
		List<Integer> dirtySegments = null;
		if (lockProvider.acquireLock(treeId)) {
			try {
				notifier.preRebuild(treeId, fullRebuild);
				long buildBeginTS = System.currentTimeMillis();
				if (fullRebuild)
					rebuildCompleteTree(treeId);
				dirtySegments = htStore.getDirtySegments(treeId);
				htStore.markSegments(treeId, dirtySegments);
				List<Integer> dirtyNodes = rebuildLeaves(treeId, dirtySegments);
				rebuildInternalNodes(treeId, dirtyNodes);
				htStore.unmarkSegments(treeId, dirtySegments);
				if (fullRebuild)
					htStore.setCompleteRebuiltTimestamp(treeId, buildBeginTS);
				notifier.postRebuild(treeId, fullRebuild);
				return dirtySegments.size();
			} catch (HashTreesCustomRuntimeException e) {
				if (dirtySegments != null)
					htStore.markSegments(treeId, dirtySegments);
			} finally {
				lockProvider.releaseLock(treeId);
			}
		}
		return 0;
	}

	/**
	 * This reads all the entries from the {@link Store}, and updates
	 * {@link HashTreesStore} with those (key,value) pairs. Also reads all the
	 * existing entries from {@link HashTreesStore} and if they don't exist on
	 * the {@link Store}, removes from {@link HashTreesStore}.
	 * 
	 * @param treeId
	 * @throws IOException
	 */
	private void rebuildCompleteTree(long treeId) throws IOException {
		Iterator<Map.Entry<byte[], byte[]>> itr = store.iterator(treeId);
		while (itr.hasNext()) {
			Map.Entry<byte[], byte[]> pair = itr.next();
			hPutInternal(HTOperation.PUT_IF_ABSENT,
					ByteBuffer.wrap(pair.getKey()),
					ByteBuffer.wrap(pair.getValue()));
		}
		Iterator<SegmentData> segDataItr = htStore
				.getSegmentDataIterator(treeId);
		while (segDataItr.hasNext()) {
			SegmentData sd = segDataItr.next();
			if (!store.contains(sd.getKey())) {
				hRemoveInternal(HTOperation.REMOVE_IF_ABSENT, sd.key);
			}
		}
	}

	/**
	 * Rebuilds the dirty segments, and updates the segment hashes of the
	 * leaves.
	 * 
	 * @param treeId
	 * @param dirtySegments
	 * @return corresponding nodeIds of the segments.
	 * @throws IOException
	 */
	private List<Integer> rebuildLeaves(long treeId,
			final List<Integer> dirtySegments) throws IOException {
		List<Integer> nodeIds = new ArrayList<>();
		for (int dirtySegId : dirtySegments) {
			if (htStore.clearDirtySegment(treeId, dirtySegId)) {
				ByteBuffer digest = digestSegmentData(treeId, dirtySegId);
				int nodeId = getLeafIdFromSegmentId(dirtySegId);
				htStore.putSegmentHash(treeId, nodeId, digest);
				nodeIds.add(nodeId);
			}
		}
		return nodeIds;
	}

	private ByteBuffer digestSegmentData(long treeId, int segId)
			throws IOException {
		List<SegmentData> dirtySegmentData = htStore.getSegment(treeId, segId);
		List<String> hexStrings = new ArrayList<String>();
		for (SegmentData sd : dirtySegmentData)
			hexStrings.add(getHexString(sd.key, sd.digest));
		return digestHexStrings(hexStrings);
	}

	/**
	 * Updates the segment hashes iteratively for each level on the tree.
	 * 
	 * @param dirtyNodeIds
	 * @throws IOException
	 */
	private void rebuildInternalNodes(long treeId,
			final List<Integer> dirtyNodeIds) throws IOException {
		Set<Integer> parentNodeIds = new TreeSet<Integer>();
		Set<Integer> nodeIds = new TreeSet<Integer>(dirtyNodeIds);
		while (!nodeIds.isEmpty()) {
			for (int nodeId : nodeIds)
				parentNodeIds.add(getParent(nodeId, noOfChildren));
			rebuildParentNodes(treeId, parentNodeIds);
			nodeIds.clear();
			nodeIds.addAll(parentNodeIds);
			parentNodeIds.clear();
			if (nodeIds.contains(ROOT_NODE))
				break;
		}
	}

	/**
	 * For each parent id, gets all the child hashes, and updates the parent
	 * hash.
	 * 
	 * @param parentIds
	 * @throws IOException
	 */
	private void rebuildParentNodes(long treeId, final Set<Integer> parentIds)
			throws IOException {
		List<Integer> children;
		List<ByteBuffer> segHashes = new ArrayList<>(noOfChildren);
		ByteBuffer segHashBB;
		SegmentHash segHash;

		for (int parentId : parentIds) {
			children = getImmediateChildren(parentId, noOfChildren);
			for (int child : children) {
				segHash = htStore.getSegmentHash(treeId, child);
				segHashBB = (segHash == null) ? null : segHash.hash.duplicate();
				if (segHashBB != null)
					segHashes.add(segHashBB);
			}
			ByteBuffer digest = digestByteBuffers(segHashes);
			htStore.putSegmentHash(treeId, parentId, digest);
			segHashes.clear();
		}
	}

	@Override
	public void sPut(final List<KeyValue> keyValuePairs) throws IOException {
		notifier.preSPut(keyValuePairs);
		for (KeyValue keyValuePair : keyValuePairs)
			store.put(keyValuePair.getKey(), keyValuePair.getValue());
		notifier.postSPut(keyValuePairs);
	}

	@Override
	public void sRemove(final List<ByteBuffer> keys) throws IOException {
		notifier.preSRemove(keys);
		for (ByteBuffer key : keys)
			store.delete(key.array());
		notifier.postSRemove(keys);
	}

	@Override
	public void deleteTreeNode(long treeId, int nodeId) throws IOException {
		int leftMostNodeId = getSegmentIdFromLeafId(getLeftMostChildNode(
				nodeId, noOfChildren, height));
		int rightMostNodeId = getSegmentIdFromLeafId(getRightMostChildNode(
				nodeId, noOfChildren, height));
		Iterator<SegmentData> sdItr = htStore.getSegmentDataIterator(treeId,
				leftMostNodeId, rightMostNodeId);
		while (sdItr.hasNext()) {
			SegmentData sd = sdItr.next();
			store.delete(sd.getKey());
		}
	}

	// Should be always called before using this instance.
	@Override
	public void start() {
		enableNonBlockingOperations();
	}

	private void enableNonBlockingOperations() {
		synchronized (nonBlockingCallsLock) {
			if (enabledNonBlockingCalls) {
				if (bgDataUpdater == null)
					bgDataUpdater = new NonBlockingHTDataUpdater(this,
							nonBlockingQueueSize);
				new Thread(bgDataUpdater).start();
				LOGGER.info("Non blocking calls are enabled.");
			}
		}
	}

	private void disableNonblockingOperations() {
		synchronized (nonBlockingCallsLock) {
			if (enabledNonBlockingCalls && bgDataUpdater != null) {
				CountDownLatch countDownLatch = new CountDownLatch(1);
				bgDataUpdater.stopAsync(countDownLatch);
				try {
					countDownLatch.await();
				} catch (InterruptedException e) {
					LOGGER.warn(
							"Exception occurred while waiting data updater to stop",
							e);
				}
				bgDataUpdater = null;
				LOGGER.info("Non blocking calls are disabled.");
			}
		}
	}

	@Override
	public void stop() {
		disableNonblockingOperations();
	}

	@Override
	public void addObserver(HashTreesObserver observer) {
		notifier.addObserver(observer);
	}

	@Override
	public void removeObserver(HashTreesObserver observer) {
		notifier.removeObserver(observer);
	}

	private static int compareSegNodeIds(SegmentHash left, SegmentHash right) {
		if (left == null && right == null)
			return 0;
		if (left == null)
			return 1;
		if (right == null)
			return -1;
		return left.getNodeId() - right.getNodeId();
	}

	private static int compareSegmentKeys(SegmentData left, SegmentData right) {
		if (left == null && right == null)
			return 0;
		if (left == null)
			return 1;
		if (right == null)
			return -1;
		return ByteUtils.compareTo(left.getKey(), right.getKey());
	}

	private static int getValidSegmentsCount(int noOfSegments) {
		return ((noOfSegments > MAX_NO_OF_SEGMENTS) || (noOfSegments < 0)) ? MAX_NO_OF_SEGMENTS
				: roundUpToPowerOf2(noOfSegments);
	}

	/**
	 * Concatenates the given ByteBuffer values by first converting them to the
	 * equivalent hex strings, and then concatenates by adding the comma
	 * delimiter.
	 * 
	 * @param values
	 * @return
	 */
	public static String getHexString(ByteBuffer... values) {
		StringBuffer sb = new StringBuffer();
		for (int i = 0; i < values.length - 1; i++)
			sb.append(Hex.encodeHexString(values[i].array()) + COMMA_DELIMETER);
		if (values.length > 0)
			sb.append(Hex.encodeHexString(values[values.length - 1].array()));
		return sb.toString();
	}

	public static ByteBuffer digestByteBuffers(List<ByteBuffer> bbList) {
		List<String> hexStrings = new ArrayList<String>();
		for (ByteBuffer bb : bbList)
			hexStrings.add(Hex.encodeHexString(bb.array()));
		return digestHexStrings(hexStrings);
	}

	public static ByteBuffer digestHexStrings(List<String> hexStrings) {
		StringBuilder sb = new StringBuilder();
		for (String hexString : hexStrings)
			sb.append(hexString + NEW_LINE_DELIMETER);
		return ByteBuffer.wrap(sha1(sb.toString().getBytes()));
	}

	/**
	 * Segment block id starts with 0. Each leaf node corresponds to a segment
	 * block. This function does the mapping from segment block id to leaf node
	 * id.
	 * 
	 * @param segId
	 * @return
	 */
	private int getLeafIdFromSegmentId(int segId) {
		return internalNodesCount + segId;
	}

	private int getSegmentIdFromLeafId(int leafNodeId) {
		return leafNodeId - internalNodesCount;
	}

	private boolean isLeafNode(int nodeId) {
		return nodeId >= internalNodesCount;
	}

	/**
	 * Used to tag type of operation when the input is fed into the non blocking
	 * version of {@link HashTreesImpl} hPut and hRemove methods.
	 * 
	 */
	private static enum HTOperation {
		PUT, REMOVE, PUT_IF_ABSENT, REMOVE_IF_ABSENT
	}

	/**
	 * A task to enable non blocking calls on all
	 * {@link HashTreesImpl#hPut(ByteArray, ByteArray)} and
	 * {@link HashTreesImpl#hRemove(ByteArray)} operation.
	 * 
	 */
	@ThreadSafe
	private static class NonBlockingHTDataUpdater extends
			NonBlockingQueuingTask<Pair<HTOperation, List<ByteBuffer>>> {

		private static final Pair<HTOperation, List<ByteBuffer>> STOP_MARKER = new Pair<HTOperation, List<ByteBuffer>>(
				HTOperation.PUT, null);
		private final ConcurrentSkipListSet<ByteBuffer> keysOnQueue = new ConcurrentSkipListSet<>();
		private final HashTreesImpl hTree;

		public NonBlockingHTDataUpdater(final HashTreesImpl hTree,
				int maxElementsToQue) {
			super(STOP_MARKER, maxElementsToQue);
			this.hTree = hTree;
		}

		@Override
		public void enque(Pair<HTOperation, List<ByteBuffer>> pair) {
			if (pair != STOP_MARKER) {
				ByteBuffer key = pair.getSecond().get(0);
				boolean isAbsent = keysOnQueue.add(key);
				switch (pair.getFirst()) {
				case PUT:
				case REMOVE:
					super.enque(pair);
					break;
				case PUT_IF_ABSENT:
				case REMOVE_IF_ABSENT:
					if (isAbsent)
						super.enque(pair);
					break;
				}
			}
		}

		@Override
		public void handleElement(Pair<HTOperation, List<ByteBuffer>> pair) {
			ByteBuffer key = pair.getSecond().get(0);
			try {
				switch (pair.getFirst()) {
				case PUT:
				case PUT_IF_ABSENT:
					hTree.hPutInternal(key, pair.getSecond().get(1));
					break;
				case REMOVE:
				case REMOVE_IF_ABSENT:
					hTree.hRemoveInternal(key);
					break;
				}
			} catch (IOException e) {
				throw new RuntimeException(e);
			} finally {
				keysOnQueue.remove(key);
			}
		}

	}

	@NotThreadSafe
	public static class Builder {

		private final Store store;
		private final HashTreesStore htStore;
		private final HashTreesIdProvider treeIdProvider;

		private SegmentIdProvider segIdProvider;
		private int noOfSegments = 1 << 17,
				nonBlockingQueueSize = DEFAULT_NB_QUE_SIZE;
		private boolean enabledNonBlockingCalls = true;
		private LockProvider lockProvider;

		public Builder(Store store, HashTreesIdProvider treeIdProvider,
				HashTreesStore htStore) {
			this.store = store;
			this.htStore = htStore;
			this.treeIdProvider = treeIdProvider;
		}

		/**
		 * Creates a in memory based hashtrees storage, and stores the digests
		 * over there. Look at {@link HashTreesMemStore} for more information.
		 * 
		 * @param store
		 * @param treeIdProvider
		 */
		public Builder(Store store, HashTreesIdProvider treeIdProvider) {
			this(store, treeIdProvider, new HashTreesMemStore());
		}

		/**
		 * Creates a LevelDB based store for storing hash tree contents. LevelDB
		 * files are stored in htStoreDirName. Look at
		 * {@link HashTreesPersistentStore}.
		 * 
		 * @param store
		 * @param treeIdProvider
		 * @param htStoreDirName
		 * @throws IOException
		 */
		public Builder(Store store, HashTreesIdProvider treeIdProvider,
				String htStoreDirName) throws IOException {
			this(store, treeIdProvider, new HashTreesPersistentStore(
					htStoreDirName));
		}

		/**
		 * By default {@link ModuloSegIdProvider} is used.
		 * 
		 * @param segIdProvider
		 * @return
		 */
		public Builder setSegmentIdProvider(SegmentIdProvider segIdProvider) {
			this.segIdProvider = segIdProvider;
			return this;
		}

		/**
		 * Depends upon data size, this should be set. Data size and
		 * noOfSegments should be directly proportional. With higher dataSize,
		 * and lesser noOfSegments means each segment will get more amount of
		 * data. When a segment is marked as dirty, the rebuild process has to
		 * read huge data unnecessarily.
		 * 
		 * Default value is 1 << 17. (More than 1 million).
		 * 
		 * @param noOfSegments
		 *            , value should be a power of 2, otherwise it will be
		 *            converted to the equivalent.
		 * @return
		 */
		public Builder setNoOfSegments(int noOfSegments) {
			this.noOfSegments = getValidSegmentsCount(noOfSegments);
			return this;
		}

		/**
		 * Enable/Disable non blocking calls. By default non blocking calls are
		 * enabled.
		 * 
		 * @param enabledNonBlockingCalls
		 *            , true enables non blocking calls, false disables non
		 *            blocking calls.
		 * @return
		 */
		public Builder setEnabledNonBlockingCalls(
				boolean enabledNonBlockingCalls) {
			this.enabledNonBlockingCalls = enabledNonBlockingCalls;
			return this;
		}

		/**
		 * Sets queue size about how many elements can be queued in memory.
		 * Default value is {@link Integer#MAX_VALUE}.
		 * 
		 * @param nonBlockingQueueSize
		 * @return
		 */
		public Builder setNonBlockingQueueSize(int nonBlockingQueueSize) {
			assert (nonBlockingQueueSize > 0);
			this.nonBlockingQueueSize = nonBlockingQueueSize;
			return this;
		}

		/**
		 * Sets a lock provider to use. By default this uses
		 * {@link HTReentrantLockProvider}.
		 * 
		 * @param lockProvider
		 * @return
		 */
		public Builder setLockProvider(LockProvider lockProvider) {
			assert (lockProvider != null);
			this.lockProvider = lockProvider;
			return this;
		}

		public HashTreesImpl build() {
			if (segIdProvider == null)
				segIdProvider = new ModuloSegIdProvider(noOfSegments);
			if (lockProvider == null)
				lockProvider = new HTReentrantLockProvider();
			return new HashTreesImpl(noOfSegments, enabledNonBlockingCalls,
					nonBlockingQueueSize, treeIdProvider, segIdProvider,
					htStore, store, lockProvider);
		}
	}
} public interface HashTreesObserver {

	void preSPut(List<KeyValue> keyValuePairs);

	void postSPut(List<KeyValue> keyValuePairs);

	void preSRemove(List<ByteBuffer> keys);

	void postSRemove(List<ByteBuffer> keys);

	void preHPut(ByteBuffer key, ByteBuffer value);

	void postHPut(ByteBuffer key, ByteBuffer value);

	void preHRemove(ByteBuffer key);

	void postHRemove(ByteBuffer key);

	void preRebuild(long treeId, boolean isFullRebuild);

	void postRebuild(long treeId, boolean isFullRebuild);

} public class SimpleTreeIdProvider implements HashTreesIdProvider {

	public static final long TREE_ID = 1;
	private static final List<Long> TREE_IDS;

	static {
		List<Long> treeIds = new ArrayList<Long>();
		treeIds.add(TREE_ID);
		TREE_IDS = Collections.unmodifiableList(treeIds);
	}

	@Override
	public long getTreeId(byte[] key) {
		return TREE_ID;
	}

	@Override
	public Iterator<Long> getAllPrimaryTreeIds() {
		return TREE_IDS.iterator();
	}
} public class SyncDiffResult {
	public final int totKeyDifferences, totExtrinsicSegments;

	public SyncDiffResult(int totKeyDifferences, int totExtrinsicSegments) {
		this.totKeyDifferences = totKeyDifferences;
		this.totExtrinsicSegments = totExtrinsicSegments;
	}

	public boolean isAnyUpdatesMade() {
		return (totKeyDifferences > 0) || (totExtrinsicSegments > 0);
	}

	@Override
	public String toString() {
		return MoreObjects.toStringHelper(this)
				.add("totKeyDifferences", totKeyDifferences)
				.add("totExtrinsicSegments", totExtrinsicSegments).toString();
	}
} public class EmptySyncListProvider implements HashTreesSynchListProvider {

	@Override
	public List<ServerName> getServerNameListFor(long treeId) {
		return Collections.emptyList();
	}

} public class HashTreesManager extends StoppableTask implements
		HashTreesSyncCallsObserver, Service {

	private final static Logger LOG = LoggerFactory
			.getLogger(HashTreesManager.class);
	private final static String HT_MGR_TPOOL = "HTMgrWorkerThreadPool";
	private final static String HT_MGR_SCHED_THREAD = "HTMgrSchedulerThread";
	private final static String HT_THRIFT_SERVER_THREAD = "HTThriftServerThread";
	private final static long MAX_UNSYNCED_TIME = 10 * 60 * 1000; // in
																	// milliseconds

	private final int noOfThreads;
	private final long fullRebuildPeriod, period;
	private final boolean synchEnabled, rebuildEnabled;
	private final ServerName localServer;
	private final HashTrees hashTrees;
	private final HashTreesIdProvider treeIdProvider;
	private final HashTreesSynchListProvider syncListProvider;
	private final HashTreesSynchAuthenticator authenticator;
	private final SyncType syncType;

	private final HashTreesManagerObserverNotifier notifier = new HashTreesManagerObserverNotifier();
	private final ConcurrentSkipListMap<ServerName, HashTreesRemoteClient> servers = new ConcurrentSkipListMap<>();
	private final ConcurrentMap<Pair<ServerName, Long>, Pair<Long, Boolean>> remoteTreeAndLastBuildReqTS = new ConcurrentHashMap<>();
	private final ConcurrentMap<Pair<ServerName, Long>, Long> remoteTreeAndLastSyncedTS = new ConcurrentHashMap<>();

	private final AtomicBoolean initialized = new AtomicBoolean(false);
	private final AtomicBoolean stopped = new AtomicBoolean(false);

	private volatile ExecutorService threadPool;
	private volatile ScheduledExecutorService scheduledExecutor;
	private volatile HashTreesThriftServerTask htThriftServer;

	public HashTreesManager(int noOfThreads, long period,
			long fullRebuildPeriod, boolean rebuildEnabled,
			boolean synchEnabled, ServerName localServer, HashTrees hashTrees,
			HashTreesIdProvider treeIdProvider,
			HashTreesSynchListProvider syncListProvider,
			HashTreesSynchAuthenticator authenticator, SyncType syncType) {
		this.noOfThreads = noOfThreads;
		this.period = period;
		this.fullRebuildPeriod = fullRebuildPeriod;
		this.synchEnabled = synchEnabled;
		this.rebuildEnabled = rebuildEnabled;
		this.localServer = localServer;
		this.syncListProvider = syncListProvider;
		this.hashTrees = hashTrees;
		this.treeIdProvider = treeIdProvider;
		this.authenticator = authenticator;
		this.syncType = syncType;
	}

	@Override
	public void onRebuildHashTreeResponse(RebuildHashTreeResponse response) {
		LOG.info("Rebuild response arrived : {}.", response);
		Pair<ServerName, Long> snAndTid = Pair.create(response.responder,
				response.treeId);
		Pair<Long, Boolean> tsAndResponse = remoteTreeAndLastBuildReqTS
				.get(snAndTid);
		if (tsAndResponse != null
				&& tsAndResponse.getFirst().equals(response.tokenNo)) {
			Pair<Long, Boolean> updatedResponse = Pair.create(response.tokenNo,
					true);
			remoteTreeAndLastBuildReqTS.replace(snAndTid, tsAndResponse,
					updatedResponse);
		}
		LOG.info("Rebuild response : {} - processed.", response);
	}

	@Override
	public void onRebuildHashTreeRequest(RebuildHashTreeRequest request)
			throws IOException {
		LOG.info("Rebuild request arrived : {} .", request);
		try {
			rebuildHashTree(request.treeId, request.expFullRebuildTimeInt);
			HashTreesRemoteClient client = getHashTreeSyncClient(request.requester);
			RebuildHashTreeResponse response = new RebuildHashTreeResponse(
					localServer, request.treeId, request.tokenNo);
			client.submitRebuildResponse(response);
		} catch (TException e) {
			throw new IOException(e);
		}
		LOG.info("Rebuild request : {} - processed", request);
	}

	private void rebuildAllLocalTrees() {
		Iterator<Callable<Void>> rebuildTasks = Iterators.transform(
				treeIdProvider.getAllPrimaryTreeIds(),
				new Function<Long, Callable<Void>>() {

					@Override
					public Callable<Void> apply(final Long treeId) {
						return new Callable<Void>() {

							@Override
							public Void call() throws IOException {
								sendRebuildRequestToRemoteTrees(treeId);
								rebuildHashTree(treeId, fullRebuildPeriod);
								return null;
							}
						};
					}
				});

		LOG.info("Building locally managed trees.");
		TaskQueue<Void> taskQueue = new TaskQueue<Void>(threadPool,
				rebuildTasks, noOfThreads);
		while (taskQueue.hasNext()) {
			try {
				taskQueue.next().get();
			} catch (ExecutionException | InterruptedException e) {
				LOG.info("Failure occurred in build task.", e);
			}
			if (hasStopRequested()) {
				taskQueue.stopAsync();
			}
		}
		LOG.info("No of successful/failed rebuild tasks : "
				+ taskQueue.getPasseTasksCount() + "/"
				+ taskQueue.getFailedTasksCount());
		LOG.info("Building locally managed trees - Done");
	}

	private void sendRebuildRequestToRemoteTrees(long treeId) {
		Iterator<ServerName> serverItr = syncListProvider.getServerNameListFor(
				treeId).iterator();
		while (serverItr.hasNext()) {
			ServerName sn = serverItr.next();
			Pair<ServerName, Long> serverNameWTreeId = Pair.create(sn, treeId);
			try {
				long buildReqTS = System.currentTimeMillis();
				HashTreesRemoteClient client = getHashTreeSyncClient(sn);
				remoteTreeAndLastSyncedTS.putIfAbsent(serverNameWTreeId,
						buildReqTS);
				remoteTreeAndLastBuildReqTS.put(serverNameWTreeId,
						Pair.create(buildReqTS, false));
				RebuildHashTreeRequest request = new RebuildHashTreeRequest(
						localServer, treeId, buildReqTS, fullRebuildPeriod);
				client.submitRebuildRequest(request);
			} catch (TException | IOException e) {
				LOG.error("Unable to send rebuild notification to {} - {}",
						serverNameWTreeId, e.getMessage(), e);
			}
		}
	}

	private void synchAllRemoteTrees() {
		final Iterator<Long> treeIds = treeIdProvider.getAllPrimaryTreeIds();
		Iterator<Pair<ServerName, Long>> remoteTreeIterator = new Iterator<Pair<ServerName, Long>>() {

			private final ConcurrentLinkedQueue<Pair<ServerName, Long>> intQue = new ConcurrentLinkedQueue<>();

			@Override
			public Pair<ServerName, Long> next() {
				if (!hasNext())
					throw new NoSuchElementException(
							"No more elements exist to return.");
				return intQue.remove();
			}

			@Override
			public boolean hasNext() {
				loadNextElement();
				return !intQue.isEmpty();
			}

			private void loadNextElement() {
				if (intQue.isEmpty()) {
					while (treeIds.hasNext()) {
						long treeId = treeIds.next();
						for (ServerName sn : syncListProvider
								.getServerNameListFor(treeId)) {
							Pair<ServerName, Long> serverNameATreeId = Pair
									.create(sn, treeId);
							Pair<Long, Boolean> lastBuildReqTSAndResponse = remoteTreeAndLastBuildReqTS
									.remove(serverNameATreeId);
							Long unsyncedTime = remoteTreeAndLastSyncedTS
									.get(serverNameATreeId);

							if (unsyncedTime == null
									|| lastBuildReqTSAndResponse == null) {
								LOG.info(
										"Unsynced info entry is not available. Synch should be followed by rebuild. Skipping syncing {}.",
										serverNameATreeId);
								continue;
							}

							if ((lastBuildReqTSAndResponse.getSecond())
									|| ((System.currentTimeMillis() - unsyncedTime) > MAX_UNSYNCED_TIME)) {
								intQue.add(serverNameATreeId);
								remoteTreeAndLastSyncedTS
										.remove(serverNameATreeId);
							} else {
								LOG.info(
										"Did not receive confirmation from {} for the rebuilding. Not syncing the remote node.",
										serverNameATreeId);
							}
						}
						if (!intQue.isEmpty())
							break;
					}
				}
			}

			@Override
			public void remove() {
				throw new UnsupportedOperationException();
			}
		};

		Iterator<Callable<Void>> syncTasks = Iterators.transform(
				remoteTreeIterator,
				new Function<Pair<ServerName, Long>, Callable<Void>>() {

					@Override
					public Callable<Void> apply(
							final Pair<ServerName, Long> serverNameAndTreeId) {
						return new Callable<Void>() {

							@Override
							public Void call() throws IOException,
									SynchNotAllowedException {
								synch(serverNameAndTreeId.getFirst(),
										serverNameAndTreeId.getSecond(), false,
										syncType);
								return null;
							}
						};
					}
				});

		LOG.info("Synching remote hash trees.");
		TaskQueue<Void> taskQueue = new TaskQueue<>(threadPool, syncTasks,
				noOfThreads);
		while (taskQueue.hasNext()) {
			try {
				taskQueue.next().get();
			} catch (InterruptedException | ExecutionException e) {
				LOG.error("Exception occurred in synch task.", e);
			}
			if (hasStopRequested()) {
				taskQueue.stopAsync();
			}
		}
		LOG.info("No of successful/failed synch tasks : {} / {}",
				taskQueue.getPasseTasksCount(), taskQueue.getFailedTasksCount());
		LOG.info("Synching remote hash trees - Done");
	}

	public void synch(ServerName sn, long treeId) throws IOException,
			SynchNotAllowedException {
		synch(sn, treeId, true, SyncType.UPDATE);
	}

	private void rebuildHashTree(final long treeId, long fullRebuildPeriod)
			throws IOException {
		Stopwatch watch = Stopwatch.createStarted();
		int dirtySegsCount = hashTrees.rebuildHashTree(treeId,
				fullRebuildPeriod);
		watch.stop();
		LOG.info("Total no of dirty segments : {} ", dirtySegsCount);
		LOG.info("Time taken for rebuilding (treeId: {}) (in ms) : {}", treeId,
				watch.elapsed(TimeUnit.MILLISECONDS));
	}

	public void synch(final ServerName sn, final long treeId,
			boolean doAuthenticate, SyncType syncType) throws IOException,
			SynchNotAllowedException {
		boolean synchAllowed = doAuthenticate ? authenticator.canSynch(
				localServer, sn) : true;
		Pair<ServerName, Long> hostNameAndTreeId = Pair.create(sn, treeId);
		if (synchAllowed) {
			boolean synced = false;
			SyncDiffResult result = null;
			notifier.preSync(treeId, sn);
			try {
				LOG.info("Syncing {}.", hostNameAndTreeId);
				Stopwatch watch = Stopwatch.createStarted();
				HashTreesRemoteClient remoteSyncClient = getHashTreeSyncClient(sn);
				result = hashTrees.synch(treeId, remoteSyncClient, syncType);
				LOG.info("Synch result for {} - {}", hostNameAndTreeId, result);
				watch.stop();
				LOG.info("Time taken for syncing ({}) (in ms) : {}",
						hostNameAndTreeId, watch.elapsed(TimeUnit.MILLISECONDS));
				LOG.info("Syncing {} complete.", hostNameAndTreeId);
				synced = true;
			} catch (TException e) {
				LOG.error("Unable to synch remote hash tree server {} : {}",
						hostNameAndTreeId, e);
			} finally {
				notifier.postSync(treeId, sn, result, synced);
			}
		} else {
			LOG.error("Synch is not allowed between {} and {}", localServer, sn);
			throw new SynchNotAllowedException(localServer, sn);
		}
	}

	private HashTreesRemoteClient getHashTreeSyncClient(ServerName sn)
			throws TTransportException {
		HashTreesRemoteClient client = servers.get(sn);
		if (client == null) {
			servers.putIfAbsent(sn, new HashTreesRemoteClient(sn));
			client = servers.get(sn);
		}
		return client;
	}

	public void addObserver(HashTreesManagerObserver observer) {
		notifier.addObserver(observer);
	}

	public void removeObserver(HashTreesManagerObserver observer) {
		notifier.removeObserver(observer);
	}

	@Override
	public void start() {
		if (initialized.compareAndSet(false, true)) {
			LOG.info("Hash tree sync manager operations starting.");
			String hostNameAndPortNo = localServer.toString();
			String threadPoolName = HT_MGR_TPOOL + "," + hostNameAndPortNo;
			String executorThreadName = HT_MGR_SCHED_THREAD + ","
					+ hostNameAndPortNo;
			threadPool = Executors.newFixedThreadPool(noOfThreads,
					new CustomThreadFactory(threadPoolName));
			scheduledExecutor = Executors.newScheduledThreadPool(1,
					new CustomThreadFactory(executorThreadName));
			CountDownLatch initializedLatch = new CountDownLatch(1);
			htThriftServer = new HashTreesThriftServerTask(hashTrees, this,
					syncListProvider, localServer.getPortNo(), initializedLatch);
			new Thread(htThriftServer, HT_THRIFT_SERVER_THREAD).start();
			try {
				initializedLatch.await();
			} catch (InterruptedException e) {
				LOG.error(
						"Exception occurred while waiting for the server to start",
						e);
			}
			scheduledExecutor.scheduleWithFixedDelay(this, 0, period,
					TimeUnit.MILLISECONDS);
		} else {
			LOG.info("HashTreeSyncManager initialized already.");
			return;
		}
	}

	@Override
	public void runImpl() {
		LOG.info("Executing rebuild/synch operations.");
		if (rebuildEnabled)
			rebuildAllLocalTrees();
		if (synchEnabled)
			synchAllRemoteTrees();
		LOG.info("Executing rebuild/synch operations - Done.");
	}

	@Override
	public void stop() {
		if (stopped.compareAndSet(false, true)) {
			LOG.info("Stopping hash trees manager operations.");
			CountDownLatch localLatch = new CountDownLatch(1);
			super.stopAsync(localLatch);
			try {
				localLatch.await();
			} catch (InterruptedException e) {
				LOG.error("Exception occurred while stopping the operations.",
						e);
			}
			if (scheduledExecutor != null)
				scheduledExecutor.shutdown();
			if (htThriftServer != null)
				htThriftServer.stopAsync();
			if (threadPool != null)
				threadPool.shutdown();
			LOG.info("Stopping hash trees manager operations - Done.");
		} else
			LOG.info("Hash trees manager operations stopped already. No actions were taken.");
	}

	@NotThreadSafe
	public static class Builder {

		public final static int DEF_NO_OF_THREADS = 10;
		// Default scheduling time interval
		public final static long DEF_SCHEDULE_PERIOD = 5 * 60 * 1000;

		private final ServerName localServer;
		private final HashTrees hashTrees;
		private final HashTreesIdProvider treeIdProvider;
		private final HashTreesSynchListProvider syncListProvider;

		private long period = DEF_SCHEDULE_PERIOD, fullRebuildPeriod = -1;
		private int noOfThreads = DEF_NO_OF_THREADS;
		private boolean rebuildEnabled = true, syncEnabled = true;
		private HashTreesSynchAuthenticator authenticator;
		private SyncType syncType;

		public Builder(String serverName, int portNo, HashTrees hashTrees,
				HashTreesIdProvider treeIdProvider,
				HashTreesSynchListProvider syncListProvider) {
			this.localServer = new ServerName(serverName, portNo);
			this.hashTrees = hashTrees;
			this.treeIdProvider = treeIdProvider;
			this.syncListProvider = syncListProvider;
		}

		/**
		 * Schedules rebuild and synch operation periodically. First execution
		 * will begin after calling {@link HashTreesManager#start()}. The
		 * following execution will begin after 'period' time interval from the
		 * first task's completion.
		 * 
		 * Default value is 5 minutes.
		 * 
		 * @param period
		 *            , in milliseconds.
		 * @return
		 */
		public Builder schedule(long period) {
			this.period = period;
			return this;
		}

		/**
		 * Allows to execute full rebuild on hash trees.This will be triggered
		 * if a tree is not fully rebuilt for more than fullRebuildPeriod. By
		 * default fullRebuild is never called on {@link HashTrees}.
		 * 
		 * @param fullRebuildPeriod
		 * @return
		 */
		public Builder setFullRebuildPeriod(long fullRebuildPeriod) {
			this.fullRebuildPeriod = fullRebuildPeriod;
			return this;
		}

		/**
		 * Sets no of threads to be created by thread pool. Thread pool is used
		 * for rebuild/synch operations. By default 10 threads are used.
		 * 
		 * @param noOfThreads
		 * @return
		 */
		public Builder setNoOfThreads(int noOfThreads) {
			this.noOfThreads = noOfThreads;
			return this;
		}

		/**
		 * Disables rebuild operation. By default this is enabled.
		 * 
		 * @return
		 */
		public Builder disableRebuild() {
			this.rebuildEnabled = false;
			return this;
		}

		/**
		 * Disables synch operation. By default this is enabled.
		 * 
		 * @param enable
		 * @return
		 */
		public Builder disableSync() {
			this.syncEnabled = false;
			return this;
		}

		public Builder setSyncType(SyncType syncType) {
			this.syncType = syncType;
			return this;
		}

		public HashTreesManager build() {
			if (authenticator == null)
				authenticator = new AllowAllSynchAuthenticator();
			if (syncType == null)
				syncType = SyncType.UPDATE;
			return new HashTreesManager(noOfThreads, period, fullRebuildPeriod,
					rebuildEnabled, syncEnabled, localServer, hashTrees,
					treeIdProvider, syncListProvider, authenticator, syncType);
		}
	}
} public class HashTreesRemoteClient implements HashTrees {

	private final HashTreesClientPool clientPool;

	public HashTreesRemoteClient(final ServerName sn) {
		this.clientPool = HashTreesClientPool.getThriftClientPool(sn);
	}

	public String ping() throws IOException {
		HashTreesSyncInterface.Client remoteTree = null;
		try {
			remoteTree = clientPool.borrowObject();
			return remoteTree.ping();
		} catch (Exception e) {
			throw new IOException(e);
		} finally {
			if (remoteTree != null)
				clientPool.returnObject(remoteTree);
		}
	}

	@Override
	public void sPut(List<KeyValue> keyValuePairs) throws IOException {
		HashTreesSyncInterface.Client remoteTree = null;
		try {
			remoteTree = clientPool.borrowObject();
			remoteTree.sPut(keyValuePairs);
		} catch (Exception e) {
			throw new IOException(e);
		} finally {
			if (remoteTree != null)
				clientPool.returnObject(remoteTree);
		}
	}

	@Override
	public void sRemove(List<ByteBuffer> keys) throws IOException {
		HashTreesSyncInterface.Client remoteTree = null;
		try {
			remoteTree = clientPool.borrowObject();
			remoteTree.sRemove(keys);
		} catch (Exception e) {
			throw new IOException(e);
		} finally {
			if (remoteTree != null)
				clientPool.returnObject(remoteTree);
		}
	}

	@Override
	public List<SegmentHash> getSegmentHashes(long treeId, List<Integer> nodeIds)
			throws IOException {
		HashTreesSyncInterface.Client remoteTree = null;
		try {
			remoteTree = clientPool.borrowObject();
			return remoteTree.getSegmentHashes(treeId, nodeIds);
		} catch (Exception e) {
			throw new IOException(e);
		} finally {
			if (remoteTree != null)
				clientPool.returnObject(remoteTree);
		}
	}

	@Override
	public SegmentHash getSegmentHash(long treeId, int nodeId)
			throws IOException {
		HashTreesSyncInterface.Client remoteTree = null;
		try {
			remoteTree = clientPool.borrowObject();
			return remoteTree.getSegmentHash(treeId, nodeId);
		} catch (Exception e) {
			throw new IOException(e);
		} finally {
			if (remoteTree != null)
				clientPool.returnObject(remoteTree);
		}
	}

	@Override
	public List<SegmentData> getSegment(long treeId, int segId)
			throws IOException {
		HashTreesSyncInterface.Client remoteTree = null;
		try {
			remoteTree = clientPool.borrowObject();
			return remoteTree.getSegment(treeId, segId);
		} catch (Exception e) {
			throw new IOException(e);
		} finally {
			if (remoteTree != null)
				clientPool.returnObject(remoteTree);
		}
	}

	@Override
	public SegmentData getSegmentData(long treeId, int segId, ByteBuffer key)
			throws IOException {
		HashTreesSyncInterface.Client remoteTree = null;
		try {
			remoteTree = clientPool.borrowObject();
			return remoteTree.getSegmentData(treeId, segId, key);
		} catch (Exception e) {
			throw new IOException(e);
		} finally {
			if (remoteTree != null)
				clientPool.returnObject(remoteTree);
		}
	}

	@Override
	public void deleteTreeNode(long treeId, int nodeId) throws IOException {
		HashTreesSyncInterface.Client remoteTree = null;
		try {
			remoteTree = clientPool.borrowObject();
			remoteTree.deleteTreeNode(treeId, nodeId);
		} catch (Exception e) {
			throw new IOException(e);
		} finally {
			if (remoteTree != null)
				clientPool.returnObject(remoteTree);
		}
	}

	public void submitRebuildResponse(RebuildHashTreeResponse response)
			throws IOException {
		HashTreesSyncInterface.Client remoteTree = null;
		try {
			remoteTree = clientPool.borrowObject();
			remoteTree.submitRebuildResponse(response);
		} catch (Exception e) {
			throw new IOException(e);
		} finally {
			if (remoteTree != null)
				clientPool.returnObject(remoteTree);
		}
	}

	public void submitRebuildRequest(RebuildHashTreeRequest request)
			throws IOException {
		HashTreesSyncInterface.Client remoteTree = null;
		try {
			remoteTree = clientPool.borrowObject();
			remoteTree.submitRebuildRequest(request);
		} catch (Exception e) {
			throw new IOException(e);
		} finally {
			if (remoteTree != null)
				clientPool.returnObject(remoteTree);
		}
	}

	@Override
	public void hPut(ByteBuffer key, ByteBuffer value) throws IOException {
		throw new IOException("Remote tree does not support this operation.");
	}

	@Override
	public void hRemove(ByteBuffer key) throws IOException {
		throw new IOException("Remote tree does not support this operation.");
	}

	@Override
	public int rebuildHashTree(long treeId, long fullRebuildPeriod)
			throws IOException {
		throw new IOException("Remote tree does not support this operation.");
	}

	@Override
	public int rebuildHashTree(long treeId, boolean fullRebuild)
			throws IOException {
		throw new IOException("Remote tree does not support this operation.");
	}

	@Override
	public SyncDiffResult synch(long treeId, HashTrees remoteTree,
			SyncType syncType) throws IOException {
		throw new IOException("Remote tree does not support this operation.");
	}

	@Override
	public SyncDiffResult synch(long treeId, HashTrees remoteTree)
			throws IOException {
		throw new IOException("Remote tree does not support this operation.");
	}

	@Override
	public void addObserver(HashTreesObserver observer) throws IOException {
		throw new IOException("Remote tree does not support this operation.");
	}

	@Override
	public void removeObserver(HashTreesObserver observer) throws IOException {
		throw new IOException("Remote tree does not support this operation.");
	}
} public interface HashTreesSynchListProvider {

	public List<ServerName> getServerNameListFor(long treeId);
} @ThreadSafe
public class HashTreesMemStore extends HashTreesBaseStore {

	private final ConcurrentMap<Long, HashTreeMemStore> treeIdAndIndHashTree = new ConcurrentHashMap<>();

	private static class HashTreeMemStore {
		private final ConcurrentMap<Integer, ByteBuffer> segmentHashes = new ConcurrentSkipListMap<Integer, ByteBuffer>();
		private final ConcurrentSkipListMap<Integer, ConcurrentSkipListMap<ByteBuffer, ByteBuffer>> segDataBlocks = new ConcurrentSkipListMap<Integer, ConcurrentSkipListMap<ByteBuffer, ByteBuffer>>();
		private final AtomicLong lastRebuiltTS = new AtomicLong(0);
		private final AtomicBitSet markedSegments = new AtomicBitSet();
	}

	private HashTreeMemStore getIndHTree(long treeId) {
		if (!treeIdAndIndHashTree.containsKey(treeId))
			treeIdAndIndHashTree.putIfAbsent(treeId, new HashTreeMemStore());
		return treeIdAndIndHashTree.get(treeId);
	}

	@Override
	public SegmentData getSegmentData(long treeId, int segId, ByteBuffer key) {
		ConcurrentSkipListMap<ByteBuffer, ByteBuffer> segDataBlock = getIndHTree(treeId).segDataBlocks
				.get(segId);
		if (segDataBlock != null) {
			ByteBuffer value = segDataBlock.get(key);
			if (value != null) {
				ByteBuffer intKey = ByteBuffer.wrap(key.array());
				return new SegmentData(segId, intKey, value);
			}
		}
		return null;
	}

	@Override
	public void putSegmentData(long treeId, int segId, ByteBuffer key,
			ByteBuffer digest) {
		HashTreeMemStore hTreeStore = getIndHTree(treeId);
		if (!hTreeStore.segDataBlocks.containsKey(segId))
			hTreeStore.segDataBlocks.putIfAbsent(segId,
					new ConcurrentSkipListMap<ByteBuffer, ByteBuffer>());
		hTreeStore.segDataBlocks.get(segId).put(key.duplicate(),
				digest.duplicate());
	}

	@Override
	public void deleteSegmentData(long treeId, int segId, ByteBuffer key) {
		HashTreeMemStore indPartition = getIndHTree(treeId);
		Map<ByteBuffer, ByteBuffer> segDataBlock = indPartition.segDataBlocks
				.get(segId);
		if (segDataBlock != null)
			segDataBlock.remove(key.duplicate());
	}

	@Override
	public List<SegmentData> getSegment(long treeId, int segId) {
		HashTreeMemStore indPartition = getIndHTree(treeId);
		ConcurrentMap<ByteBuffer, ByteBuffer> segDataBlock = indPartition.segDataBlocks
				.get(segId);
		if (segDataBlock == null)
			return Collections.emptyList();
		List<SegmentData> result = new ArrayList<SegmentData>();
		for (Map.Entry<ByteBuffer, ByteBuffer> entry : segDataBlock.entrySet())
			result.add(new SegmentData(segId, entry.getKey(), entry.getValue()));
		return result;
	}

	/**
	 * Iterator returned by this method is not thread safe.
	 */
	@Override
	public Iterator<SegmentData> getSegmentDataIterator(long treeId) {
		final HashTreeMemStore memStore = getIndHTree(treeId);
		final Iterator<Map.Entry<Integer, ConcurrentSkipListMap<ByteBuffer, ByteBuffer>>> dataBlocksItr = memStore.segDataBlocks
				.entrySet().iterator();
		return convertMapEntriesToSegmentData(dataBlocksItr);
	}

	@Override
	public Iterator<SegmentData> getSegmentDataIterator(long treeId,
			int fromSegId, int toSegId) throws IOException {
		final HashTreeMemStore memStore = getIndHTree(treeId);
		final Iterator<Map.Entry<Integer, ConcurrentSkipListMap<ByteBuffer, ByteBuffer>>> dataBlocksItr = memStore.segDataBlocks
				.subMap(fromSegId, true, toSegId, true).entrySet().iterator();
		return convertMapEntriesToSegmentData(dataBlocksItr);
	}

	private static Iterator<SegmentData> convertMapEntriesToSegmentData(
			final Iterator<Map.Entry<Integer, ConcurrentSkipListMap<ByteBuffer, ByteBuffer>>> dataBlocksItr) {
		return new Iterator<SegmentData>() {

			volatile int segId;
			volatile Iterator<Map.Entry<ByteBuffer, ByteBuffer>> itr = null;

			@Override
			public boolean hasNext() {
				if (itr == null || !itr.hasNext()) {
					while (dataBlocksItr.hasNext()) {
						Map.Entry<Integer, ConcurrentSkipListMap<ByteBuffer, ByteBuffer>> entry = dataBlocksItr
								.next();
						segId = entry.getKey();
						itr = entry.getValue().entrySet().iterator();
						if (itr.hasNext())
							break;
					}
				}
				return (itr != null) && itr.hasNext();
			}

			@Override
			public SegmentData next() {
				if (itr == null || !itr.hasNext())
					throw new NoSuchElementException(
							"No more elements exist to return.");
				Map.Entry<ByteBuffer, ByteBuffer> entry = itr.next();
				return new SegmentData(segId, entry.getKey(), entry.getValue());
			}

			@Override
			public void remove() {
				throw new UnsupportedOperationException();
			}
		};
	}

	@Override
	public void putSegmentHash(long treeId, int nodeId, ByteBuffer digest) {
		HashTreeMemStore indPartition = getIndHTree(treeId);
		indPartition.segmentHashes.put(nodeId, digest.duplicate());
	}

	@Override
	public SegmentHash getSegmentHash(long treeId, int nodeId) {
		HashTreeMemStore indPartition = getIndHTree(treeId);
		ByteBuffer hash = indPartition.segmentHashes.get(nodeId);
		if (hash == null)
			return null;
		return new SegmentHash(nodeId, hash);
	}

	@Override
	public List<SegmentHash> getSegmentHashes(long treeId,
			Collection<Integer> nodeIds) {
		List<SegmentHash> result = new ArrayList<SegmentHash>();
		for (int nodeId : nodeIds) {
			ByteBuffer hash = getIndHTree(treeId).segmentHashes.get(nodeId);
			if (hash != null)
				result.add(new SegmentHash(nodeId, hash));
		}
		return result;
	}

	@Override
	public void setCompleteRebuiltTimestamp(long treeId, long timestamp) {
		HashTreeMemStore indTree = getIndHTree(treeId);
		indTree.lastRebuiltTS.set(timestamp);
	}

	@Override
	public long getCompleteRebuiltTimestamp(long treeId) {
		long value = getIndHTree(treeId).lastRebuiltTS.get();
		return value;
	}

	@Override
	public void deleteTree(long treeId) {
		treeIdAndIndHashTree.remove(treeId);
	}

	@Override
	public void markSegments(long treeId, List<Integer> segIds) {
		for (int segId : segIds)
			getIndHTree(treeId).markedSegments.set(segId);
	}

	@Override
	public void unmarkSegments(long treeId, List<Integer> segIds) {
		getIndHTree(treeId).markedSegments.clearBits(segIds);
	}

	@Override
	public List<Integer> getMarkedSegments(long treeId) {
		return getIndHTree(treeId).markedSegments.getAllSetBits();
	}

	@Override
	public void stop() {
		// Nothing to stop.
	}

	@Override
	public void start() {
		// Nothing to initialize
	}

	@Override
	protected void setDirtySegmentInternal(long treeId, int segId) {
		// Nothing to do.
	}

	@Override
	protected void clearDirtySegmentInternal(long treeId, int segId) {
		// Nothing to do.
	}

	@Override
	protected List<Integer> getDirtySegmentsInternal(long treeId) {
		return Collections.emptyList();
	}
} public class HashTreesPersistentStore extends HashTreesBaseStore {

	private static final Logger LOG = LoggerFactory
			.getLogger(HashTreesPersistentStore.class);
	private static final byte[] EMPTY_VALUE = new byte[0];

	private final String dbDir;
	private final DB dbObj;

	public HashTreesPersistentStore(String dbDir) throws IOException {
		this.dbDir = dbDir;
		this.dbObj = initDB(dbDir);
	}

	private static boolean createDir(String dirName) {
		File file = new File(dirName);
		if (file.exists())
			return true;
		return file.mkdirs();
	}

	private static DB initDB(String dbDir) throws IOException {
		createDir(dbDir);
		Options options = new Options();
		options.createIfMissing(true);
		return new JniDBFactory().open(new File(dbDir), options);
	}

	public String getDbDir() {
		return dbDir;
	}

	@Override
	protected void setDirtySegmentInternal(long treeId, int segId) {
		dbObj.put(generateDirtySegmentKey(treeId, segId), EMPTY_VALUE);
	}

	@Override
	protected void clearDirtySegmentInternal(long treeId, int segId) {
		byte[] key = generateDirtySegmentKey(treeId, segId);
		dbObj.delete(key);
	}

	@Override
	protected List<Integer> getDirtySegmentsInternal(long treeId) {
		DBIterator itr = dbObj.iterator();
		byte[] prefixKey = generateBaseKey(BaseKey.DIRTY_SEG, treeId);
		itr.seek(prefixKey);
		Iterator<Integer> dirtySegmentsItr = new DataFilterableIterator<>(
				prefixKey, true, KVBYTES_TO_SEGID_CONVERTER, itr);
		List<Integer> dirtySegments = new ArrayList<>();
		while (dirtySegmentsItr.hasNext()) {
			Integer treeIdAndDirtySeg = dirtySegmentsItr.next();
			dirtySegments.add(treeIdAndDirtySeg);
		}
		return dirtySegments;
	}

	@Override
	public void putSegmentHash(long treeId, int nodeId, ByteBuffer digest) {
		dbObj.put(generateSegmentHashKey(treeId, nodeId), digest.array());
	}

	@Override
	public SegmentHash getSegmentHash(long treeId, int nodeId) {
		byte[] value = dbObj.get(generateSegmentHashKey(treeId, nodeId));
		if (value != null)
			return new SegmentHash(nodeId, ByteBuffer.wrap(value));
		return null;
	}

	@Override
	public List<SegmentHash> getSegmentHashes(long treeId,
			Collection<Integer> nodeIds) {
		List<SegmentHash> result = new ArrayList<SegmentHash>();
		SegmentHash temp;
		for (int nodeId : nodeIds) {
			temp = getSegmentHash(treeId, nodeId);
			if (temp != null)
				result.add(temp);
		}
		return result;
	}

	@Override
	public void setCompleteRebuiltTimestamp(long treeId, long ts) {
		byte[] value = new byte[ByteUtils.SIZEOF_LONG];
		ByteBuffer bbValue = ByteBuffer.wrap(value);
		bbValue.putLong(ts);
		byte[] key = generateMetaDataKey(MetaDataKey.FULL_REBUILT_TS, treeId);
		dbObj.put(key, value);
	}

	@Override
	public long getCompleteRebuiltTimestamp(long treeId) {
		byte[] key = generateMetaDataKey(MetaDataKey.FULL_REBUILT_TS, treeId);
		byte[] value = dbObj.get(key);
		return (value == null) ? 0 : ByteUtils.toLong(value, 0);
	}

	@Override
	public void deleteTree(long treeId) {
		DBIterator dbItr;
		byte[] temp = new byte[LEN_BASEKEY_AND_TREEID];
		for (BaseKey keyPrefix : BaseKey.values()) {
			dbItr = dbObj.iterator();
			ByteBuffer wrap = ByteBuffer.wrap(temp);
			fillBaseKey(wrap, keyPrefix, treeId);
			dbItr.seek(wrap.array());
			for (; dbItr.hasNext(); dbItr.next()) {
				if (ByteUtils.compareTo(temp, 0, temp.length, dbItr.peekNext()
						.getKey(), 0, temp.length) != 0)
					break;
				dbObj.delete(dbItr.peekNext().getKey());
			}
		}
	}

	@Override
	public void putSegmentData(long treeId, int segId, ByteBuffer key,
			ByteBuffer digest) {
		byte[] dbKey = generateSegmentDataKey(treeId, segId, key);
		dbObj.put(dbKey, digest.array());
	}

	@Override
	public SegmentData getSegmentData(long treeId, int segId, ByteBuffer key) {
		byte[] dbKey = generateSegmentDataKey(treeId, segId, key);
		byte[] value = dbObj.get(dbKey);
		if (value != null) {
			ByteBuffer intKeyBB = ByteBuffer.wrap(key.array());
			ByteBuffer valueBB = ByteBuffer.wrap(value);
			return new SegmentData(segId, intKeyBB, valueBB);
		}
		return null;
	}

	@Override
	public void deleteSegmentData(long treeId, int segId, ByteBuffer key) {
		byte[] dbKey = generateSegmentDataKey(treeId, segId, key);
		dbObj.delete(dbKey);
	}

	@Override
	public Iterator<SegmentData> getSegmentDataIterator(long treeId) {
		byte[] startKey = generateBaseKey(BaseKey.SEG_DATA, treeId);
		DBIterator iterator = dbObj.iterator();
		iterator.seek(startKey);
		return new DataFilterableIterator<>(startKey, false,
				KVBYTES_TO_SEGDATA_CONVERTER, iterator);
	}

	@Override
	public Iterator<SegmentData> getSegmentDataIterator(long treeId,
			int fromSegId, int toSegId) throws IOException {
		byte[] startKey = generateSegmentDataKey(treeId, fromSegId);
		byte[] endKey = generateSegmentDataKey(treeId, toSegId);
		DBIterator iterator = dbObj.iterator();
		iterator.seek(startKey);
		return new DataFilterableIterator<>(endKey, false,
				KVBYTES_TO_SEGDATA_CONVERTER, iterator);
	}

	@Override
	public List<SegmentData> getSegment(long treeId, int segId) {
		byte[] startKey = generateSegmentDataKey(treeId, segId);
		DBIterator iterator = dbObj.iterator();
		iterator.seek(startKey);
		return Lists.newArrayList(new DataFilterableIterator<>(startKey, true,
				KVBYTES_TO_SEGDATA_CONVERTER, iterator));
	}

	@Override
	public void markSegments(long treeId, List<Integer> segIds) {
		for (int segId : segIds) {
			byte[] key = generateRebuildMarkerKey(treeId, segId);
			dbObj.put(key, EMPTY_VALUE);
		}
	}

	@Override
	public void unmarkSegments(long treeId, List<Integer> segIds) {
		for (int segId : segIds) {
			byte[] key = generateRebuildMarkerKey(treeId, segId);
			dbObj.delete(key);
		}
	}

	@Override
	public List<Integer> getMarkedSegments(long treeId) {
		byte[] startKey = generateBaseKey(BaseKey.REBUILD_MARKER, treeId);
		DBIterator itr = dbObj.iterator();
		itr.seek(startKey);

		return Lists.newArrayList(new DataFilterableIterator<>(startKey, true,
				KVBYTES_TO_SEGID_CONVERTER, itr));
	}

	/**
	 * Deletes the db files.
	 * 
	 */
	public void delete() {
		stop();
		File dbDirObj = new File(dbDir);
		if (dbDirObj.exists())
			FileUtils.deleteQuietly(dbDirObj);
	}

	@Override
	public void start() {
		// Nothing to do.
	}

	@Override
	public void stop() {
		try {
			dbObj.close();
		} catch (IOException e) {
			LOG.warn("Exception occurred while closing leveldb connection.");
		}
	}

} public interface HashTreesStore extends Service {

	/**
	 * A segment data is the value inside a segment block.
	 * 
	 * @param treeId
	 * @param segId
	 * @param key
	 * @param digest
	 */
	void putSegmentData(long treeId, int segId, ByteBuffer key,
			ByteBuffer digest) throws IOException;

	/**
	 * Returns the SegmentData for the given key if available, otherwise returns
	 * null.
	 * 
	 * @param treeId
	 * @param segId
	 * @param key
	 * @return
	 */
	SegmentData getSegmentData(long treeId, int segId, ByteBuffer key)
			throws IOException;

	/**
	 * Deletes the given segement data from the block.
	 * 
	 * @param treeId
	 * @param segId
	 * @param key
	 */
	void deleteSegmentData(long treeId, int segId, ByteBuffer key)
			throws IOException;

	/**
	 * Returns an iterator to read all the segment data of the given tree id.
	 * 
	 * Note: Iterator implementations should throw
	 * {@link HashTreesCustomRuntimeException} so that failure cases can be
	 * handled properly by {@link HashTrees}
	 * 
	 * @param treeId
	 * @return
	 */
	Iterator<SegmentData> getSegmentDataIterator(long treeId)
			throws IOException;

	/**
	 * Returns an iterator to read all the segment data of the given tree id,
	 * and starting from a given segId and to segId.
	 * 
	 * Note: Iterator implementations should throw
	 * {@link HashTreesCustomRuntimeException} so that failure cases can be
	 * handled properly by {@link HashTrees}
	 * 
	 * @param treeId
	 * @param fromSegId
	 *            , inclusive
	 * @param toSegId
	 *            , inclusive
	 * @return
	 * @throws IOException
	 */
	Iterator<SegmentData> getSegmentDataIterator(long treeId, int fromSegId,
			int toSegId) throws IOException;

	/**
	 * Given a segment id, returns the list of all segment data in the
	 * individual segment block.
	 * 
	 * @param treeId
	 * @param segId
	 * @return
	 */
	List<SegmentData> getSegment(long treeId, int segId) throws IOException;

	/**
	 * Segment hash is the hash of all data inside a segment block. A segment
	 * hash is stored on a tree node.
	 * 
	 * @param treeId
	 * @param nodeId
	 *            , identifier of the node in the hash tree.
	 * @param digest
	 */
	void putSegmentHash(long treeId, int nodeId, ByteBuffer digest)
			throws IOException;

	/**
	 * 
	 * @param treeId
	 * @param nodeId
	 * @return
	 */
	SegmentHash getSegmentHash(long treeId, int nodeId) throws IOException;

	/**
	 * Returns the data inside the nodes of the hash tree. If the node id is not
	 * present in the hash tree, the entry will be missing in the result.
	 * 
	 * @param treeId
	 * @param nodeIds
	 *            , internal tree node ids.
	 * @return
	 */
	List<SegmentHash> getSegmentHashes(long treeId, Collection<Integer> nodeIds)
			throws IOException;

	/**
	 * Marks a segment as a dirty.
	 * 
	 * @param treeId
	 * @param segId
	 * @return previousValue
	 */
	boolean setDirtySegment(long treeId, int segId) throws IOException;

	/**
	 * Clears the segments, which are passed as an argument.
	 * 
	 * @param treeId
	 * @param segId
	 * @return previousValue
	 */
	boolean clearDirtySegment(long treeId, int segId) throws IOException;

	/**
	 * Gets the dirty segments without clearing those bits.
	 * 
	 * @param treeId
	 * @return
	 */
	List<Integer> getDirtySegments(long treeId) throws IOException;

	/**
	 * Sets flags for the given segments. Used during rebuild process by
	 * {@link HashTrees#rebuildHashTree(long, boolean)}. If the process crashes
	 * in the middle of rebuilding we don't want to loose track. This has to
	 * persist that information, so that we can reuse it after the process
	 * recovery.
	 * 
	 * @param segIds
	 */
	void markSegments(long treeId, List<Integer> segIds) throws IOException;

	/**
	 * Gets the marked segments for the given treeId.
	 * 
	 * @param treeId
	 * @return
	 */
	List<Integer> getMarkedSegments(long treeId) throws IOException;

	/**
	 * Unsets flags for the given segments. Used during rebuild process by
	 * {@link HashTrees#rebuildHashTree(long, boolean)}. After rebuilding is
	 * done, this will be cleared.
	 * 
	 * @param segIds
	 */
	void unmarkSegments(long treeId, List<Integer> segIds) throws IOException;

	/**
	 * Deletes the segment hashes, and segment data for the given treeId.
	 * 
	 * @param treeId
	 */
	void deleteTree(long treeId) throws IOException;

	/**
	 * Stores the timestamp at which the complete HashTree was rebuilt. This
	 * method updates the value in store only if the given value is higher than
	 * the existing timestamp, otherwise a noop.
	 * 
	 * @param timestamp
	 */
	void setCompleteRebuiltTimestamp(long treeId, long timestamp)
			throws IOException;

	/**
	 * Returns the timestamp at which the complete HashTree was rebuilt.
	 * 
	 * @return
	 */
	long getCompleteRebuiltTimestamp(long treeId) throws IOException;
} public class SimpleMemStore extends BaseStore {

	private final ConcurrentHashMap<ByteBuffer, ByteBuffer> kvMap = new ConcurrentHashMap<>();

	@Override
	public byte[] get(byte[] key) {
		ByteBuffer intKey = ByteBuffer.wrap(key);
		ByteBuffer valueBB = kvMap.get(intKey);
		if (valueBB != null) {
			byte[] value = new byte[valueBB.remaining()];
			valueBB.duplicate().get(value);
			return value;
		}
		return null;
	}

	@Override
	public void put(byte[] key, byte[] value) throws IOException {
		ByteBuffer intKey = ByteBuffer.wrap(key);
		ByteBuffer intValue = ByteBuffer.wrap(value);
		kvMap.put(intKey, intValue);
		super.put(key, value);
	}

	@Override
	public void delete(byte[] key) throws IOException {
		ByteBuffer intKey = ByteBuffer.wrap(key);
		kvMap.remove(intKey);
		super.delete(key);
	}

	/**
	 * Just returns all the key values in the store, irrespective of which
	 * treeId it belongs to.
	 */
	@Override
	public Iterator<Map.Entry<byte[], byte[]>> iterator(long treeId) {
		return iterator();
	}

	@Override
	public int hashCode() {
		return kvMap.hashCode();
	}

	@Override
	public boolean equals(Object other) {
		if (other == null || !(other instanceof SimpleMemStore))
			return false;
		SimpleMemStore that = (SimpleMemStore) other;
		return this.kvMap.equals(that.kvMap);
	}

	public Iterator<Map.Entry<byte[], byte[]>> iterator() {
		return Iterators
				.transform(
						kvMap.entrySet().iterator(),
						new Function<Map.Entry<ByteBuffer, ByteBuffer>, Map.Entry<byte[], byte[]>>() {

							@Override
							public Entry<byte[], byte[]> apply(
									final Entry<ByteBuffer, ByteBuffer> input) {
								return Maps.immutableEntry(input.getKey()
										.array(), input.getValue().array());
							};
						});
	}

	@Override
	public boolean contains(byte[] key) {
		ByteBuffer intKey = ByteBuffer.wrap(key);
		return kvMap.containsKey(intKey);
	}

} public interface Store {

	boolean contains(byte[] key) throws IOException;

	/**
	 * Adds the key and value to the local store. Also this call should forward
	 * the request to {@link HashTrees#hPut(ByteBuffer, ByteBuffer)} with the
	 * HashTrees object (If there is a hash trees registered already through
	 * {@link #registerHashTrees(HashTrees)}.
	 * 
	 * @param key
	 * @param value
	 * @throws IOException
	 */
	void put(byte[] key, byte[] value) throws IOException;

	byte[] get(byte[] key) throws IOException;

	/**
	 * Removes the key from the local store. Also this call should forward the
	 * request to {@link HashTrees#hRemove(ByteBuffer)} with the HashTrees
	 * object (If there is a hash trees registered already through
	 * {@link #registerHashTrees(HashTrees)}.
	 * 
	 * @param key
	 * @return
	 * @throws IOException
	 */
	void delete(byte[] key) throws IOException;

	/**
	 * This should return only (key,value) that are belonging to treeId. Used by
	 * {@link HashTrees#rebuildHashTree(long, boolean)} while rebuilding a
	 * specific hashtree.
	 * 
	 * Note: Iterator implementations should throw
	 * {@link HashTreesCustomRuntimeException} so that failure cases can be
	 * handled properly by {@link HashTrees}
	 * 
	 * @param treeId
	 * @return
	 * @throws IOException
	 */
	Iterator<Map.Entry<byte[], byte[]>> iterator(long treeId)
			throws IOException;

	/**
	 * Register hashtrees with storage implementation. On modifications to the
	 * storage, it should forward the calls to HashTrees.
	 * 
	 * @param hashTrees
	 */
	void registerHashTrees(HashTrees hashTrees);
} public class HashTreesConstants {

	public static final int DEFAULT_HASH_TREE_SERVER_PORT_NO = 4554;
	public static final String DEFAULT_STORE_FILE_PATH = "/tmp/hashtree.db";
} public class HashTreesImplTestUtils {

	private static final Random RANDOM = new Random(System.currentTimeMillis());
	public static final int ROOT_NODE = 0;
	public static final int DEFAULT_TREE_ID = 1;
	public static final int DEFAULT_SEG_DATA_BLOCKS_COUNT = 1 << 5;
	public static final int DEFAULT_HTREE_SERVER_PORT_NO = 11111;
	public static final SegIdProviderTest SEG_ID_PROVIDER = new SegIdProviderTest();
	public static final SimpleTreeIdProvider TREE_ID_PROVIDER = new SimpleTreeIdProvider();

	/**
	 * Default SegId provider which expects the key to be an integer wrapped as
	 * bytes.
	 * 
	 */
	public static class SegIdProviderTest implements SegmentIdProvider {

		@Override
		public int getSegmentId(byte[] key) {
			try {
				ByteBuffer bb = ByteBuffer.wrap(key);
				return bb.getInt();
			} catch (NumberFormatException e) {
				throw new IllegalArgumentException(
						"Exception occurred while converting the string");
			}
		}

	}

	public static class HTreeComponents {

		public final HashTreesStore hTStore;
		public final SimpleMemStore store;
		public final HashTreesImpl hTree;

		public HTreeComponents(final HashTreesStore hTStore,
				final SimpleMemStore store, final HashTreesImpl hTree) {
			this.hTStore = hTStore;
			this.store = store;
			this.hTree = hTree;
		}
	}

	public static byte[] randomBytes() {
		byte[] emptyBuffer = new byte[8];
		RANDOM.nextBytes(emptyBuffer);
		return emptyBuffer;
	}

	public static ByteBuffer randomByteBuffer() {
		byte[] random = new byte[8];
		RANDOM.nextBytes(random);
		return ByteBuffer.wrap(random);
	}

	public static String randomDirName() {
		return "/tmp/test/random" + RANDOM.nextInt();
	}

	public static HTreeComponents createHashTree(int noOfSegDataBlocks,
			boolean enabledNonBlockingCalls,
			final HashTreesIdProvider treeIdProv,
			final SegmentIdProvider segIdPro, final HashTreesStore hTStore) {
		SimpleMemStore store = new SimpleMemStore();
		HashTreesImpl hTree = new HashTreesImpl.Builder(store, treeIdProv,
				hTStore).setNoOfSegments(noOfSegDataBlocks)
				.setEnabledNonBlockingCalls(enabledNonBlockingCalls)
				.setSegmentIdProvider(segIdPro).build();
		store.registerHashTrees(hTree);
		return new HTreeComponents(hTStore, store, hTree);
	}

	public static HTreeComponents createHashTree(int noOfSegments,
			boolean enabledNonBlockingCalls, final HashTreesStore hTStore) {
		SimpleMemStore store = new SimpleMemStore();
		ModuloSegIdProvider segIdProvider = new ModuloSegIdProvider(
				noOfSegments);
		HashTreesImpl hTree = new HashTreesImpl.Builder(store,
				TREE_ID_PROVIDER, hTStore).setNoOfSegments(noOfSegments)
				.setEnabledNonBlockingCalls(enabledNonBlockingCalls)
				.setSegmentIdProvider(segIdProvider).build();
		store.registerHashTrees(hTree);
		return new HTreeComponents(hTStore, store, hTree);
	}

	public static HashTreesMemStore generateInMemoryStore() {
		return new HashTreesMemStore();
	}

	public static HashTreesPersistentStore generatePersistentStore()
			throws IOException {
		return new HashTreesPersistentStore(randomDirName());
	}

	public static HashTreesStore[] generateInMemoryAndPersistentStores()
			throws IOException {
		HashTreesStore[] stores = new HashTreesStore[2];
		stores[0] = generateInMemoryStore();
		stores[1] = generatePersistentStore();
		return stores;
	}

	public static void closeStores(HashTreesStore... stores) {
		for (HashTreesStore store : stores) {
			if (store instanceof HashTreesPersistentStore) {
				HashTreesPersistentStore pStore = (HashTreesPersistentStore) store;
				FileUtils.deleteQuietly(new File(pStore.getDbDir()));
			}
		}
	}
} public static class HTreeComponents {

	public final HashTreesStore hTStore;
	public final SimpleMemStore store;
	public final HashTreesImpl hTree;

	public HTreeComponents(final HashTreesStore hTStore,
			final SimpleMemStore store, final HashTreesImpl hTree) {
		this.hTStore = hTStore;
		this.store = store;
		this.hTree = hTree;
	}
} public class KeyValue implements org.apache.thrift.TBase<KeyValue, KeyValue._Fields>, java.io.Serializable, Cloneable {
  private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("KeyValue");

  private static final org.apache.thrift.protocol.TField KEY_FIELD_DESC = new org.apache.thrift.protocol.TField("key", org.apache.thrift.protocol.TType.STRING, (short)1);
  private static final org.apache.thrift.protocol.TField VALUE_FIELD_DESC = new org.apache.thrift.protocol.TField("value", org.apache.thrift.protocol.TType.STRING, (short)2);

  private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
  static {
    schemes.put(StandardScheme.class, new KeyValueStandardSchemeFactory());
    schemes.put(TupleScheme.class, new KeyValueTupleSchemeFactory());
  }

  public ByteBuffer key; // required
  public ByteBuffer value; // required

  /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
  public enum _Fields implements org.apache.thrift.TFieldIdEnum {
    KEY((short)1, "key"),
    VALUE((short)2, "value");

    private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();

    static {
      for (_Fields field : EnumSet.allOf(_Fields.class)) {
        byName.put(field.getFieldName(), field);
      }
    }

    /**
     * Find the _Fields constant that matches fieldId, or null if its not found.
     */
    public static _Fields findByThriftId(int fieldId) {
      switch(fieldId) {
        case 1: // KEY
          return KEY;
        case 2: // VALUE
          return VALUE;
        default:
          return null;
      }
    }

    /**
     * Find the _Fields constant that matches fieldId, throwing an exception
     * if it is not found.
     */
    public static _Fields findByThriftIdOrThrow(int fieldId) {
      _Fields fields = findByThriftId(fieldId);
      if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
      return fields;
    }

    /**
     * Find the _Fields constant that matches name, or null if its not found.
     */
    public static _Fields findByName(String name) {
      return byName.get(name);
    }

    private final short _thriftId;
    private final String _fieldName;

    _Fields(short thriftId, String fieldName) {
      _thriftId = thriftId;
      _fieldName = fieldName;
    }

    public short getThriftFieldId() {
      return _thriftId;
    }

    public String getFieldName() {
      return _fieldName;
    }
  }

  // isset id assignments
  public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
  static {
    Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
    tmpMap.put(_Fields.KEY, new org.apache.thrift.meta_data.FieldMetaData("key", org.apache.thrift.TFieldRequirementType.REQUIRED, 
        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING        , true)));
    tmpMap.put(_Fields.VALUE, new org.apache.thrift.meta_data.FieldMetaData("value", org.apache.thrift.TFieldRequirementType.REQUIRED, 
        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING        , true)));
    metaDataMap = Collections.unmodifiableMap(tmpMap);
    org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(KeyValue.class, metaDataMap);
  }

  public KeyValue() {
  }

  public KeyValue(
    ByteBuffer key,
    ByteBuffer value)
  {
    this();
    this.key = key;
    this.value = value;
  }

  /**
   * Performs a deep copy on <i>other</i>.
   */
  public KeyValue(KeyValue other) {
    if (other.isSetKey()) {
      this.key = org.apache.thrift.TBaseHelper.copyBinary(other.key);
;
    }
    if (other.isSetValue()) {
      this.value = org.apache.thrift.TBaseHelper.copyBinary(other.value);
;
    }
  }

  public KeyValue deepCopy() {
    return new KeyValue(this);
  }

  @Override
  public void clear() {
    this.key = null;
    this.value = null;
  }

  public byte[] getKey() {
    setKey(org.apache.thrift.TBaseHelper.rightSize(key));
    return key == null ? null : key.array();
  }

  public ByteBuffer bufferForKey() {
    return key;
  }

  public KeyValue setKey(byte[] key) {
    setKey(key == null ? (ByteBuffer)null : ByteBuffer.wrap(key));
    return this;
  }

  public KeyValue setKey(ByteBuffer key) {
    this.key = key;
    return this;
  }

  public void unsetKey() {
    this.key = null;
  }

  /** Returns true if field key is set (has been assigned a value) and false otherwise */
  public boolean isSetKey() {
    return this.key != null;
  }

  public void setKeyIsSet(boolean value) {
    if (!value) {
      this.key = null;
    }
  }

  public byte[] getValue() {
    setValue(org.apache.thrift.TBaseHelper.rightSize(value));
    return value == null ? null : value.array();
  }

  public ByteBuffer bufferForValue() {
    return value;
  }

  public KeyValue setValue(byte[] value) {
    setValue(value == null ? (ByteBuffer)null : ByteBuffer.wrap(value));
    return this;
  }

  public KeyValue setValue(ByteBuffer value) {
    this.value = value;
    return this;
  }

  public void unsetValue() {
    this.value = null;
  }

  /** Returns true if field value is set (has been assigned a value) and false otherwise */
  public boolean isSetValue() {
    return this.value != null;
  }

  public void setValueIsSet(boolean value) {
    if (!value) {
      this.value = null;
    }
  }

  public void setFieldValue(_Fields field, Object value) {
    switch (field) {
    case KEY:
      if (value == null) {
        unsetKey();
      } else {
        setKey((ByteBuffer)value);
      }
      break;

    case VALUE:
      if (value == null) {
        unsetValue();
      } else {
        setValue((ByteBuffer)value);
      }
      break;

    }
  }

  public Object getFieldValue(_Fields field) {
    switch (field) {
    case KEY:
      return getKey();

    case VALUE:
      return getValue();

    }
    throw new IllegalStateException();
  }

  /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
  public boolean isSet(_Fields field) {
    if (field == null) {
      throw new IllegalArgumentException();
    }

    switch (field) {
    case KEY:
      return isSetKey();
    case VALUE:
      return isSetValue();
    }
    throw new IllegalStateException();
  }

  @Override
  public boolean equals(Object that) {
    if (that == null)
      return false;
    if (that instanceof KeyValue)
      return this.equals((KeyValue)that);
    return false;
  }

  public boolean equals(KeyValue that) {
    if (that == null)
      return false;

    boolean this_present_key = true && this.isSetKey();
    boolean that_present_key = true && that.isSetKey();
    if (this_present_key || that_present_key) {
      if (!(this_present_key && that_present_key))
        return false;
      if (!this.key.equals(that.key))
        return false;
    }

    boolean this_present_value = true && this.isSetValue();
    boolean that_present_value = true && that.isSetValue();
    if (this_present_value || that_present_value) {
      if (!(this_present_value && that_present_value))
        return false;
      if (!this.value.equals(that.value))
        return false;
    }

    return true;
  }

  @Override
  public int hashCode() {
    return 0;
  }

  public int compareTo(KeyValue other) {
    if (!getClass().equals(other.getClass())) {
      return getClass().getName().compareTo(other.getClass().getName());
    }

    int lastComparison = 0;
    KeyValue typedOther = (KeyValue)other;

    lastComparison = Boolean.valueOf(isSetKey()).compareTo(typedOther.isSetKey());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetKey()) {
      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.key, typedOther.key);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    lastComparison = Boolean.valueOf(isSetValue()).compareTo(typedOther.isSetValue());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetValue()) {
      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.value, typedOther.value);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    return 0;
  }

  public _Fields fieldForId(int fieldId) {
    return _Fields.findByThriftId(fieldId);
  }

  public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
    schemes.get(iprot.getScheme()).getScheme().read(iprot, this);
  }

  public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
    schemes.get(oprot.getScheme()).getScheme().write(oprot, this);
  }

  @Override
  public String toString() {
    StringBuilder sb = new StringBuilder("KeyValue(");
    boolean first = true;

    sb.append("key:");
    if (this.key == null) {
      sb.append("null");
    } else {
      org.apache.thrift.TBaseHelper.toString(this.key, sb);
    }
    first = false;
    if (!first) sb.append(", ");
    sb.append("value:");
    if (this.value == null) {
      sb.append("null");
    } else {
      org.apache.thrift.TBaseHelper.toString(this.value, sb);
    }
    first = false;
    sb.append(")");
    return sb.toString();
  }

  public void validate() throws org.apache.thrift.TException {
    // check for required fields
    if (key == null) {
      throw new org.apache.thrift.protocol.TProtocolException("Required field 'key' was not present! Struct: " + toString());
    }
    if (value == null) {
      throw new org.apache.thrift.protocol.TProtocolException("Required field 'value' was not present! Struct: " + toString());
    }
  }

  private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
    try {
      write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
    } catch (org.apache.thrift.TException te) {
      throw new java.io.IOException(te);
    }
  }

  private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
    try {
      read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
    } catch (org.apache.thrift.TException te) {
      throw new java.io.IOException(te);
    }
  }

  private static class KeyValueStandardSchemeFactory implements SchemeFactory {
    public KeyValueStandardScheme getScheme() {
      return new KeyValueStandardScheme();
    }
  }

  private static class KeyValueStandardScheme extends StandardScheme<KeyValue> {

    public void read(org.apache.thrift.protocol.TProtocol iprot, KeyValue struct) throws org.apache.thrift.TException {
      org.apache.thrift.protocol.TField schemeField;
      iprot.readStructBegin();
      while (true)
      {
        schemeField = iprot.readFieldBegin();
        if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { 
          break;
        }
        switch (schemeField.id) {
          case 1: // KEY
            if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
              struct.key = iprot.readBinary();
              struct.setKeyIsSet(true);
            } else { 
              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
            }
            break;
          case 2: // VALUE
            if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
              struct.value = iprot.readBinary();
              struct.setValueIsSet(true);
            } else { 
              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
            }
            break;
          default:
            org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
        }
        iprot.readFieldEnd();
      }
      iprot.readStructEnd();

      // check for required fields of primitive type, which can't be checked in the validate method
      struct.validate();
    }

    public void write(org.apache.thrift.protocol.TProtocol oprot, KeyValue struct) throws org.apache.thrift.TException {
      struct.validate();

      oprot.writeStructBegin(STRUCT_DESC);
      if (struct.key != null) {
        oprot.writeFieldBegin(KEY_FIELD_DESC);
        oprot.writeBinary(struct.key);
        oprot.writeFieldEnd();
      }
      if (struct.value != null) {
        oprot.writeFieldBegin(VALUE_FIELD_DESC);
        oprot.writeBinary(struct.value);
        oprot.writeFieldEnd();
      }
      oprot.writeFieldStop();
      oprot.writeStructEnd();
    }

  }

  private static class KeyValueTupleSchemeFactory implements SchemeFactory {
    public KeyValueTupleScheme getScheme() {
      return new KeyValueTupleScheme();
    }
  }

  private static class KeyValueTupleScheme extends TupleScheme<KeyValue> {

    @Override
    public void write(org.apache.thrift.protocol.TProtocol prot, KeyValue struct) throws org.apache.thrift.TException {
      TTupleProtocol oprot = (TTupleProtocol) prot;
      oprot.writeBinary(struct.key);
      oprot.writeBinary(struct.value);
    }

    @Override
    public void read(org.apache.thrift.protocol.TProtocol prot, KeyValue struct) throws org.apache.thrift.TException {
      TTupleProtocol iprot = (TTupleProtocol) prot;
      struct.key = iprot.readBinary();
      struct.setKeyIsSet(true);
      struct.value = iprot.readBinary();
      struct.setValueIsSet(true);
    }
  }

} public class SegmentData implements org.apache.thrift.TBase<SegmentData, SegmentData._Fields>, java.io.Serializable, Cloneable {
  private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("SegmentData");

  private static final org.apache.thrift.protocol.TField SEG_ID_FIELD_DESC = new org.apache.thrift.protocol.TField("segId", org.apache.thrift.protocol.TType.I32, (short)1);
  private static final org.apache.thrift.protocol.TField KEY_FIELD_DESC = new org.apache.thrift.protocol.TField("key", org.apache.thrift.protocol.TType.STRING, (short)2);
  private static final org.apache.thrift.protocol.TField DIGEST_FIELD_DESC = new org.apache.thrift.protocol.TField("digest", org.apache.thrift.protocol.TType.STRING, (short)3);

  private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
  static {
    schemes.put(StandardScheme.class, new SegmentDataStandardSchemeFactory());
    schemes.put(TupleScheme.class, new SegmentDataTupleSchemeFactory());
  }

  public int segId; // required
  public ByteBuffer key; // required
  public ByteBuffer digest; // required

  /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
  public enum _Fields implements org.apache.thrift.TFieldIdEnum {
    SEG_ID((short)1, "segId"),
    KEY((short)2, "key"),
    DIGEST((short)3, "digest");

    private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();

    static {
      for (_Fields field : EnumSet.allOf(_Fields.class)) {
        byName.put(field.getFieldName(), field);
      }
    }

    /**
     * Find the _Fields constant that matches fieldId, or null if its not found.
     */
    public static _Fields findByThriftId(int fieldId) {
      switch(fieldId) {
        case 1: // SEG_ID
          return SEG_ID;
        case 2: // KEY
          return KEY;
        case 3: // DIGEST
          return DIGEST;
        default:
          return null;
      }
    }

    /**
     * Find the _Fields constant that matches fieldId, throwing an exception
     * if it is not found.
     */
    public static _Fields findByThriftIdOrThrow(int fieldId) {
      _Fields fields = findByThriftId(fieldId);
      if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
      return fields;
    }

    /**
     * Find the _Fields constant that matches name, or null if its not found.
     */
    public static _Fields findByName(String name) {
      return byName.get(name);
    }

    private final short _thriftId;
    private final String _fieldName;

    _Fields(short thriftId, String fieldName) {
      _thriftId = thriftId;
      _fieldName = fieldName;
    }

    public short getThriftFieldId() {
      return _thriftId;
    }

    public String getFieldName() {
      return _fieldName;
    }
  }

  // isset id assignments
  private static final int __SEGID_ISSET_ID = 0;
  private BitSet __isset_bit_vector = new BitSet(1);
  public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
  static {
    Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
    tmpMap.put(_Fields.SEG_ID, new org.apache.thrift.meta_data.FieldMetaData("segId", org.apache.thrift.TFieldRequirementType.REQUIRED, 
        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I32)));
    tmpMap.put(_Fields.KEY, new org.apache.thrift.meta_data.FieldMetaData("key", org.apache.thrift.TFieldRequirementType.REQUIRED, 
        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING        , true)));
    tmpMap.put(_Fields.DIGEST, new org.apache.thrift.meta_data.FieldMetaData("digest", org.apache.thrift.TFieldRequirementType.REQUIRED, 
        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING        , true)));
    metaDataMap = Collections.unmodifiableMap(tmpMap);
    org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(SegmentData.class, metaDataMap);
  }

  public SegmentData() {
  }

  public SegmentData(
    int segId,
    ByteBuffer key,
    ByteBuffer digest)
  {
    this();
    this.segId = segId;
    setSegIdIsSet(true);
    this.key = key;
    this.digest = digest;
  }

  /**
   * Performs a deep copy on <i>other</i>.
   */
  public SegmentData(SegmentData other) {
    __isset_bit_vector.clear();
    __isset_bit_vector.or(other.__isset_bit_vector);
    this.segId = other.segId;
    if (other.isSetKey()) {
      this.key = org.apache.thrift.TBaseHelper.copyBinary(other.key);
;
    }
    if (other.isSetDigest()) {
      this.digest = org.apache.thrift.TBaseHelper.copyBinary(other.digest);
;
    }
  }

  public SegmentData deepCopy() {
    return new SegmentData(this);
  }

  @Override
  public void clear() {
    setSegIdIsSet(false);
    this.segId = 0;
    this.key = null;
    this.digest = null;
  }

  public int getSegId() {
    return this.segId;
  }

  public SegmentData setSegId(int segId) {
    this.segId = segId;
    setSegIdIsSet(true);
    return this;
  }

  public void unsetSegId() {
    __isset_bit_vector.clear(__SEGID_ISSET_ID);
  }

  /** Returns true if field segId is set (has been assigned a value) and false otherwise */
  public boolean isSetSegId() {
    return __isset_bit_vector.get(__SEGID_ISSET_ID);
  }

  public void setSegIdIsSet(boolean value) {
    __isset_bit_vector.set(__SEGID_ISSET_ID, value);
  }

  public byte[] getKey() {
    setKey(org.apache.thrift.TBaseHelper.rightSize(key));
    return key == null ? null : key.array();
  }

  public ByteBuffer bufferForKey() {
    return key;
  }

  public SegmentData setKey(byte[] key) {
    setKey(key == null ? (ByteBuffer)null : ByteBuffer.wrap(key));
    return this;
  }

  public SegmentData setKey(ByteBuffer key) {
    this.key = key;
    return this;
  }

  public void unsetKey() {
    this.key = null;
  }

  /** Returns true if field key is set (has been assigned a value) and false otherwise */
  public boolean isSetKey() {
    return this.key != null;
  }

  public void setKeyIsSet(boolean value) {
    if (!value) {
      this.key = null;
    }
  }

  public byte[] getDigest() {
    setDigest(org.apache.thrift.TBaseHelper.rightSize(digest));
    return digest == null ? null : digest.array();
  }

  public ByteBuffer bufferForDigest() {
    return digest;
  }

  public SegmentData setDigest(byte[] digest) {
    setDigest(digest == null ? (ByteBuffer)null : ByteBuffer.wrap(digest));
    return this;
  }

  public SegmentData setDigest(ByteBuffer digest) {
    this.digest = digest;
    return this;
  }

  public void unsetDigest() {
    this.digest = null;
  }

  /** Returns true if field digest is set (has been assigned a value) and false otherwise */
  public boolean isSetDigest() {
    return this.digest != null;
  }

  public void setDigestIsSet(boolean value) {
    if (!value) {
      this.digest = null;
    }
  }

  public void setFieldValue(_Fields field, Object value) {
    switch (field) {
    case SEG_ID:
      if (value == null) {
        unsetSegId();
      } else {
        setSegId((Integer)value);
      }
      break;

    case KEY:
      if (value == null) {
        unsetKey();
      } else {
        setKey((ByteBuffer)value);
      }
      break;

    case DIGEST:
      if (value == null) {
        unsetDigest();
      } else {
        setDigest((ByteBuffer)value);
      }
      break;

    }
  }

  public Object getFieldValue(_Fields field) {
    switch (field) {
    case SEG_ID:
      return Integer.valueOf(getSegId());

    case KEY:
      return getKey();

    case DIGEST:
      return getDigest();

    }
    throw new IllegalStateException();
  }

  /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
  public boolean isSet(_Fields field) {
    if (field == null) {
      throw new IllegalArgumentException();
    }

    switch (field) {
    case SEG_ID:
      return isSetSegId();
    case KEY:
      return isSetKey();
    case DIGEST:
      return isSetDigest();
    }
    throw new IllegalStateException();
  }

  @Override
  public boolean equals(Object that) {
    if (that == null)
      return false;
    if (that instanceof SegmentData)
      return this.equals((SegmentData)that);
    return false;
  }

  public boolean equals(SegmentData that) {
    if (that == null)
      return false;

    boolean this_present_segId = true;
    boolean that_present_segId = true;
    if (this_present_segId || that_present_segId) {
      if (!(this_present_segId && that_present_segId))
        return false;
      if (this.segId != that.segId)
        return false;
    }

    boolean this_present_key = true && this.isSetKey();
    boolean that_present_key = true && that.isSetKey();
    if (this_present_key || that_present_key) {
      if (!(this_present_key && that_present_key))
        return false;
      if (!this.key.equals(that.key))
        return false;
    }

    boolean this_present_digest = true && this.isSetDigest();
    boolean that_present_digest = true && that.isSetDigest();
    if (this_present_digest || that_present_digest) {
      if (!(this_present_digest && that_present_digest))
        return false;
      if (!this.digest.equals(that.digest))
        return false;
    }

    return true;
  }

  @Override
  public int hashCode() {
    return 0;
  }

  public int compareTo(SegmentData other) {
    if (!getClass().equals(other.getClass())) {
      return getClass().getName().compareTo(other.getClass().getName());
    }

    int lastComparison = 0;
    SegmentData typedOther = (SegmentData)other;

    lastComparison = Boolean.valueOf(isSetSegId()).compareTo(typedOther.isSetSegId());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetSegId()) {
      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.segId, typedOther.segId);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    lastComparison = Boolean.valueOf(isSetKey()).compareTo(typedOther.isSetKey());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetKey()) {
      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.key, typedOther.key);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    lastComparison = Boolean.valueOf(isSetDigest()).compareTo(typedOther.isSetDigest());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetDigest()) {
      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.digest, typedOther.digest);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    return 0;
  }

  public _Fields fieldForId(int fieldId) {
    return _Fields.findByThriftId(fieldId);
  }

  public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
    schemes.get(iprot.getScheme()).getScheme().read(iprot, this);
  }

  public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
    schemes.get(oprot.getScheme()).getScheme().write(oprot, this);
  }

  @Override
  public String toString() {
    StringBuilder sb = new StringBuilder("SegmentData(");
    boolean first = true;

    sb.append("segId:");
    sb.append(this.segId);
    first = false;
    if (!first) sb.append(", ");
    sb.append("key:");
    if (this.key == null) {
      sb.append("null");
    } else {
      org.apache.thrift.TBaseHelper.toString(this.key, sb);
    }
    first = false;
    if (!first) sb.append(", ");
    sb.append("digest:");
    if (this.digest == null) {
      sb.append("null");
    } else {
      org.apache.thrift.TBaseHelper.toString(this.digest, sb);
    }
    first = false;
    sb.append(")");
    return sb.toString();
  }

  public void validate() throws org.apache.thrift.TException {
    // check for required fields
    // alas, we cannot check 'segId' because it's a primitive and you chose the non-beans generator.
    if (key == null) {
      throw new org.apache.thrift.protocol.TProtocolException("Required field 'key' was not present! Struct: " + toString());
    }
    if (digest == null) {
      throw new org.apache.thrift.protocol.TProtocolException("Required field 'digest' was not present! Struct: " + toString());
    }
  }

  private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
    try {
      write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
    } catch (org.apache.thrift.TException te) {
      throw new java.io.IOException(te);
    }
  }

  private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
    try {
      // it doesn't seem like you should have to do this, but java serialization is wacky, and doesn't call the default constructor.
      __isset_bit_vector = new BitSet(1);
      read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
    } catch (org.apache.thrift.TException te) {
      throw new java.io.IOException(te);
    }
  }

  private static class SegmentDataStandardSchemeFactory implements SchemeFactory {
    public SegmentDataStandardScheme getScheme() {
      return new SegmentDataStandardScheme();
    }
  }

  private static class SegmentDataStandardScheme extends StandardScheme<SegmentData> {

    public void read(org.apache.thrift.protocol.TProtocol iprot, SegmentData struct) throws org.apache.thrift.TException {
      org.apache.thrift.protocol.TField schemeField;
      iprot.readStructBegin();
      while (true)
      {
        schemeField = iprot.readFieldBegin();
        if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { 
          break;
        }
        switch (schemeField.id) {
          case 1: // SEG_ID
            if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
              struct.segId = iprot.readI32();
              struct.setSegIdIsSet(true);
            } else { 
              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
            }
            break;
          case 2: // KEY
            if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
              struct.key = iprot.readBinary();
              struct.setKeyIsSet(true);
            } else { 
              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
            }
            break;
          case 3: // DIGEST
            if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
              struct.digest = iprot.readBinary();
              struct.setDigestIsSet(true);
            } else { 
              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
            }
            break;
          default:
            org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
        }
        iprot.readFieldEnd();
      }
      iprot.readStructEnd();

      // check for required fields of primitive type, which can't be checked in the validate method
      if (!struct.isSetSegId()) {
        throw new org.apache.thrift.protocol.TProtocolException("Required field 'segId' was not found in serialized data! Struct: " + toString());
      }
      struct.validate();
    }

    public void write(org.apache.thrift.protocol.TProtocol oprot, SegmentData struct) throws org.apache.thrift.TException {
      struct.validate();

      oprot.writeStructBegin(STRUCT_DESC);
      oprot.writeFieldBegin(SEG_ID_FIELD_DESC);
      oprot.writeI32(struct.segId);
      oprot.writeFieldEnd();
      if (struct.key != null) {
        oprot.writeFieldBegin(KEY_FIELD_DESC);
        oprot.writeBinary(struct.key);
        oprot.writeFieldEnd();
      }
      if (struct.digest != null) {
        oprot.writeFieldBegin(DIGEST_FIELD_DESC);
        oprot.writeBinary(struct.digest);
        oprot.writeFieldEnd();
      }
      oprot.writeFieldStop();
      oprot.writeStructEnd();
    }

  }

  private static class SegmentDataTupleSchemeFactory implements SchemeFactory {
    public SegmentDataTupleScheme getScheme() {
      return new SegmentDataTupleScheme();
    }
  }

  private static class SegmentDataTupleScheme extends TupleScheme<SegmentData> {

    @Override
    public void write(org.apache.thrift.protocol.TProtocol prot, SegmentData struct) throws org.apache.thrift.TException {
      TTupleProtocol oprot = (TTupleProtocol) prot;
      oprot.writeI32(struct.segId);
      oprot.writeBinary(struct.key);
      oprot.writeBinary(struct.digest);
    }

    @Override
    public void read(org.apache.thrift.protocol.TProtocol prot, SegmentData struct) throws org.apache.thrift.TException {
      TTupleProtocol iprot = (TTupleProtocol) prot;
      struct.segId = iprot.readI32();
      struct.setSegIdIsSet(true);
      struct.key = iprot.readBinary();
      struct.setKeyIsSet(true);
      struct.digest = iprot.readBinary();
      struct.setDigestIsSet(true);
    }
  }

} public class SegmentHash implements org.apache.thrift.TBase<SegmentHash, SegmentHash._Fields>, java.io.Serializable, Cloneable {
  private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("SegmentHash");

  private static final org.apache.thrift.protocol.TField NODE_ID_FIELD_DESC = new org.apache.thrift.protocol.TField("nodeId", org.apache.thrift.protocol.TType.I32, (short)1);
  private static final org.apache.thrift.protocol.TField HASH_FIELD_DESC = new org.apache.thrift.protocol.TField("hash", org.apache.thrift.protocol.TType.STRING, (short)2);

  private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
  static {
    schemes.put(StandardScheme.class, new SegmentHashStandardSchemeFactory());
    schemes.put(TupleScheme.class, new SegmentHashTupleSchemeFactory());
  }

  public int nodeId; // required
  public ByteBuffer hash; // required

  /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
  public enum _Fields implements org.apache.thrift.TFieldIdEnum {
    NODE_ID((short)1, "nodeId"),
    HASH((short)2, "hash");

    private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();

    static {
      for (_Fields field : EnumSet.allOf(_Fields.class)) {
        byName.put(field.getFieldName(), field);
      }
    }

    /**
     * Find the _Fields constant that matches fieldId, or null if its not found.
     */
    public static _Fields findByThriftId(int fieldId) {
      switch(fieldId) {
        case 1: // NODE_ID
          return NODE_ID;
        case 2: // HASH
          return HASH;
        default:
          return null;
      }
    }

    /**
     * Find the _Fields constant that matches fieldId, throwing an exception
     * if it is not found.
     */
    public static _Fields findByThriftIdOrThrow(int fieldId) {
      _Fields fields = findByThriftId(fieldId);
      if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
      return fields;
    }

    /**
     * Find the _Fields constant that matches name, or null if its not found.
     */
    public static _Fields findByName(String name) {
      return byName.get(name);
    }

    private final short _thriftId;
    private final String _fieldName;

    _Fields(short thriftId, String fieldName) {
      _thriftId = thriftId;
      _fieldName = fieldName;
    }

    public short getThriftFieldId() {
      return _thriftId;
    }

    public String getFieldName() {
      return _fieldName;
    }
  }

  // isset id assignments
  private static final int __NODEID_ISSET_ID = 0;
  private BitSet __isset_bit_vector = new BitSet(1);
  public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
  static {
    Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
    tmpMap.put(_Fields.NODE_ID, new org.apache.thrift.meta_data.FieldMetaData("nodeId", org.apache.thrift.TFieldRequirementType.REQUIRED, 
        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I32)));
    tmpMap.put(_Fields.HASH, new org.apache.thrift.meta_data.FieldMetaData("hash", org.apache.thrift.TFieldRequirementType.REQUIRED, 
        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING        , true)));
    metaDataMap = Collections.unmodifiableMap(tmpMap);
    org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(SegmentHash.class, metaDataMap);
  }

  public SegmentHash() {
  }

  public SegmentHash(
    int nodeId,
    ByteBuffer hash)
  {
    this();
    this.nodeId = nodeId;
    setNodeIdIsSet(true);
    this.hash = hash;
  }

  /**
   * Performs a deep copy on <i>other</i>.
   */
  public SegmentHash(SegmentHash other) {
    __isset_bit_vector.clear();
    __isset_bit_vector.or(other.__isset_bit_vector);
    this.nodeId = other.nodeId;
    if (other.isSetHash()) {
      this.hash = org.apache.thrift.TBaseHelper.copyBinary(other.hash);
;
    }
  }

  public SegmentHash deepCopy() {
    return new SegmentHash(this);
  }

  @Override
  public void clear() {
    setNodeIdIsSet(false);
    this.nodeId = 0;
    this.hash = null;
  }

  public int getNodeId() {
    return this.nodeId;
  }

  public SegmentHash setNodeId(int nodeId) {
    this.nodeId = nodeId;
    setNodeIdIsSet(true);
    return this;
  }

  public void unsetNodeId() {
    __isset_bit_vector.clear(__NODEID_ISSET_ID);
  }

  /** Returns true if field nodeId is set (has been assigned a value) and false otherwise */
  public boolean isSetNodeId() {
    return __isset_bit_vector.get(__NODEID_ISSET_ID);
  }

  public void setNodeIdIsSet(boolean value) {
    __isset_bit_vector.set(__NODEID_ISSET_ID, value);
  }

  public byte[] getHash() {
    setHash(org.apache.thrift.TBaseHelper.rightSize(hash));
    return hash == null ? null : hash.array();
  }

  public ByteBuffer bufferForHash() {
    return hash;
  }

  public SegmentHash setHash(byte[] hash) {
    setHash(hash == null ? (ByteBuffer)null : ByteBuffer.wrap(hash));
    return this;
  }

  public SegmentHash setHash(ByteBuffer hash) {
    this.hash = hash;
    return this;
  }

  public void unsetHash() {
    this.hash = null;
  }

  /** Returns true if field hash is set (has been assigned a value) and false otherwise */
  public boolean isSetHash() {
    return this.hash != null;
  }

  public void setHashIsSet(boolean value) {
    if (!value) {
      this.hash = null;
    }
  }

  public void setFieldValue(_Fields field, Object value) {
    switch (field) {
    case NODE_ID:
      if (value == null) {
        unsetNodeId();
      } else {
        setNodeId((Integer)value);
      }
      break;

    case HASH:
      if (value == null) {
        unsetHash();
      } else {
        setHash((ByteBuffer)value);
      }
      break;

    }
  }

  public Object getFieldValue(_Fields field) {
    switch (field) {
    case NODE_ID:
      return Integer.valueOf(getNodeId());

    case HASH:
      return getHash();

    }
    throw new IllegalStateException();
  }

  /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
  public boolean isSet(_Fields field) {
    if (field == null) {
      throw new IllegalArgumentException();
    }

    switch (field) {
    case NODE_ID:
      return isSetNodeId();
    case HASH:
      return isSetHash();
    }
    throw new IllegalStateException();
  }

  @Override
  public boolean equals(Object that) {
    if (that == null)
      return false;
    if (that instanceof SegmentHash)
      return this.equals((SegmentHash)that);
    return false;
  }

  public boolean equals(SegmentHash that) {
    if (that == null)
      return false;

    boolean this_present_nodeId = true;
    boolean that_present_nodeId = true;
    if (this_present_nodeId || that_present_nodeId) {
      if (!(this_present_nodeId && that_present_nodeId))
        return false;
      if (this.nodeId != that.nodeId)
        return false;
    }

    boolean this_present_hash = true && this.isSetHash();
    boolean that_present_hash = true && that.isSetHash();
    if (this_present_hash || that_present_hash) {
      if (!(this_present_hash && that_present_hash))
        return false;
      if (!this.hash.equals(that.hash))
        return false;
    }

    return true;
  }

  @Override
  public int hashCode() {
    return 0;
  }

  public int compareTo(SegmentHash other) {
    if (!getClass().equals(other.getClass())) {
      return getClass().getName().compareTo(other.getClass().getName());
    }

    int lastComparison = 0;
    SegmentHash typedOther = (SegmentHash)other;

    lastComparison = Boolean.valueOf(isSetNodeId()).compareTo(typedOther.isSetNodeId());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetNodeId()) {
      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.nodeId, typedOther.nodeId);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    lastComparison = Boolean.valueOf(isSetHash()).compareTo(typedOther.isSetHash());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetHash()) {
      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.hash, typedOther.hash);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    return 0;
  }

  public _Fields fieldForId(int fieldId) {
    return _Fields.findByThriftId(fieldId);
  }

  public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
    schemes.get(iprot.getScheme()).getScheme().read(iprot, this);
  }

  public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
    schemes.get(oprot.getScheme()).getScheme().write(oprot, this);
  }

  @Override
  public String toString() {
    StringBuilder sb = new StringBuilder("SegmentHash(");
    boolean first = true;

    sb.append("nodeId:");
    sb.append(this.nodeId);
    first = false;
    if (!first) sb.append(", ");
    sb.append("hash:");
    if (this.hash == null) {
      sb.append("null");
    } else {
      org.apache.thrift.TBaseHelper.toString(this.hash, sb);
    }
    first = false;
    sb.append(")");
    return sb.toString();
  }

  public void validate() throws org.apache.thrift.TException {
    // check for required fields
    // alas, we cannot check 'nodeId' because it's a primitive and you chose the non-beans generator.
    if (hash == null) {
      throw new org.apache.thrift.protocol.TProtocolException("Required field 'hash' was not present! Struct: " + toString());
    }
  }

  private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
    try {
      write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
    } catch (org.apache.thrift.TException te) {
      throw new java.io.IOException(te);
    }
  }

  private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
    try {
      // it doesn't seem like you should have to do this, but java serialization is wacky, and doesn't call the default constructor.
      __isset_bit_vector = new BitSet(1);
      read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
    } catch (org.apache.thrift.TException te) {
      throw new java.io.IOException(te);
    }
  }

  private static class SegmentHashStandardSchemeFactory implements SchemeFactory {
    public SegmentHashStandardScheme getScheme() {
      return new SegmentHashStandardScheme();
    }
  }

  private static class SegmentHashStandardScheme extends StandardScheme<SegmentHash> {

    public void read(org.apache.thrift.protocol.TProtocol iprot, SegmentHash struct) throws org.apache.thrift.TException {
      org.apache.thrift.protocol.TField schemeField;
      iprot.readStructBegin();
      while (true)
      {
        schemeField = iprot.readFieldBegin();
        if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { 
          break;
        }
        switch (schemeField.id) {
          case 1: // NODE_ID
            if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
              struct.nodeId = iprot.readI32();
              struct.setNodeIdIsSet(true);
            } else { 
              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
            }
            break;
          case 2: // HASH
            if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
              struct.hash = iprot.readBinary();
              struct.setHashIsSet(true);
            } else { 
              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
            }
            break;
          default:
            org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
        }
        iprot.readFieldEnd();
      }
      iprot.readStructEnd();

      // check for required fields of primitive type, which can't be checked in the validate method
      if (!struct.isSetNodeId()) {
        throw new org.apache.thrift.protocol.TProtocolException("Required field 'nodeId' was not found in serialized data! Struct: " + toString());
      }
      struct.validate();
    }

    public void write(org.apache.thrift.protocol.TProtocol oprot, SegmentHash struct) throws org.apache.thrift.TException {
      struct.validate();

      oprot.writeStructBegin(STRUCT_DESC);
      oprot.writeFieldBegin(NODE_ID_FIELD_DESC);
      oprot.writeI32(struct.nodeId);
      oprot.writeFieldEnd();
      if (struct.hash != null) {
        oprot.writeFieldBegin(HASH_FIELD_DESC);
        oprot.writeBinary(struct.hash);
        oprot.writeFieldEnd();
      }
      oprot.writeFieldStop();
      oprot.writeStructEnd();
    }

  }

  private static class SegmentHashTupleSchemeFactory implements SchemeFactory {
    public SegmentHashTupleScheme getScheme() {
      return new SegmentHashTupleScheme();
    }
  }

  private static class SegmentHashTupleScheme extends TupleScheme<SegmentHash> {

    @Override
    public void write(org.apache.thrift.protocol.TProtocol prot, SegmentHash struct) throws org.apache.thrift.TException {
      TTupleProtocol oprot = (TTupleProtocol) prot;
      oprot.writeI32(struct.nodeId);
      oprot.writeBinary(struct.hash);
    }

    @Override
    public void read(org.apache.thrift.protocol.TProtocol prot, SegmentHash struct) throws org.apache.thrift.TException {
      TTupleProtocol iprot = (TTupleProtocol) prot;
      struct.nodeId = iprot.readI32();
      struct.setNodeIdIsSet(true);
      struct.hash = iprot.readBinary();
      struct.setHashIsSet(true);
    }
  }

} public class ServerName implements org.apache.thrift.TBase<ServerName, ServerName._Fields>, java.io.Serializable, Cloneable {
  private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("ServerName");

  private static final org.apache.thrift.protocol.TField HOST_NAME_FIELD_DESC = new org.apache.thrift.protocol.TField("hostName", org.apache.thrift.protocol.TType.STRING, (short)1);
  private static final org.apache.thrift.protocol.TField PORT_NO_FIELD_DESC = new org.apache.thrift.protocol.TField("portNo", org.apache.thrift.protocol.TType.I32, (short)2);

  private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
  static {
    schemes.put(StandardScheme.class, new ServerNameStandardSchemeFactory());
    schemes.put(TupleScheme.class, new ServerNameTupleSchemeFactory());
  }

  public String hostName; // required
  public int portNo; // required

  /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
  public enum _Fields implements org.apache.thrift.TFieldIdEnum {
    HOST_NAME((short)1, "hostName"),
    PORT_NO((short)2, "portNo");

    private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();

    static {
      for (_Fields field : EnumSet.allOf(_Fields.class)) {
        byName.put(field.getFieldName(), field);
      }
    }

    /**
     * Find the _Fields constant that matches fieldId, or null if its not found.
     */
    public static _Fields findByThriftId(int fieldId) {
      switch(fieldId) {
        case 1: // HOST_NAME
          return HOST_NAME;
        case 2: // PORT_NO
          return PORT_NO;
        default:
          return null;
      }
    }

    /**
     * Find the _Fields constant that matches fieldId, throwing an exception
     * if it is not found.
     */
    public static _Fields findByThriftIdOrThrow(int fieldId) {
      _Fields fields = findByThriftId(fieldId);
      if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
      return fields;
    }

    /**
     * Find the _Fields constant that matches name, or null if its not found.
     */
    public static _Fields findByName(String name) {
      return byName.get(name);
    }

    private final short _thriftId;
    private final String _fieldName;

    _Fields(short thriftId, String fieldName) {
      _thriftId = thriftId;
      _fieldName = fieldName;
    }

    public short getThriftFieldId() {
      return _thriftId;
    }

    public String getFieldName() {
      return _fieldName;
    }
  }

  // isset id assignments
  private static final int __PORTNO_ISSET_ID = 0;
  private BitSet __isset_bit_vector = new BitSet(1);
  public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
  static {
    Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
    tmpMap.put(_Fields.HOST_NAME, new org.apache.thrift.meta_data.FieldMetaData("hostName", org.apache.thrift.TFieldRequirementType.REQUIRED, 
        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
    tmpMap.put(_Fields.PORT_NO, new org.apache.thrift.meta_data.FieldMetaData("portNo", org.apache.thrift.TFieldRequirementType.REQUIRED, 
        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I32)));
    metaDataMap = Collections.unmodifiableMap(tmpMap);
    org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(ServerName.class, metaDataMap);
  }

  public ServerName() {
  }

  public ServerName(
    String hostName,
    int portNo)
  {
    this();
    this.hostName = hostName;
    this.portNo = portNo;
    setPortNoIsSet(true);
  }

  /**
   * Performs a deep copy on <i>other</i>.
   */
  public ServerName(ServerName other) {
    __isset_bit_vector.clear();
    __isset_bit_vector.or(other.__isset_bit_vector);
    if (other.isSetHostName()) {
      this.hostName = other.hostName;
    }
    this.portNo = other.portNo;
  }

  public ServerName deepCopy() {
    return new ServerName(this);
  }

  @Override
  public void clear() {
    this.hostName = null;
    setPortNoIsSet(false);
    this.portNo = 0;
  }

  public String getHostName() {
    return this.hostName;
  }

  public ServerName setHostName(String hostName) {
    this.hostName = hostName;
    return this;
  }

  public void unsetHostName() {
    this.hostName = null;
  }

  /** Returns true if field hostName is set (has been assigned a value) and false otherwise */
  public boolean isSetHostName() {
    return this.hostName != null;
  }

  public void setHostNameIsSet(boolean value) {
    if (!value) {
      this.hostName = null;
    }
  }

  public int getPortNo() {
    return this.portNo;
  }

  public ServerName setPortNo(int portNo) {
    this.portNo = portNo;
    setPortNoIsSet(true);
    return this;
  }

  public void unsetPortNo() {
    __isset_bit_vector.clear(__PORTNO_ISSET_ID);
  }

  /** Returns true if field portNo is set (has been assigned a value) and false otherwise */
  public boolean isSetPortNo() {
    return __isset_bit_vector.get(__PORTNO_ISSET_ID);
  }

  public void setPortNoIsSet(boolean value) {
    __isset_bit_vector.set(__PORTNO_ISSET_ID, value);
  }

  public void setFieldValue(_Fields field, Object value) {
    switch (field) {
    case HOST_NAME:
      if (value == null) {
        unsetHostName();
      } else {
        setHostName((String)value);
      }
      break;

    case PORT_NO:
      if (value == null) {
        unsetPortNo();
      } else {
        setPortNo((Integer)value);
      }
      break;

    }
  }

  public Object getFieldValue(_Fields field) {
    switch (field) {
    case HOST_NAME:
      return getHostName();

    case PORT_NO:
      return Integer.valueOf(getPortNo());

    }
    throw new IllegalStateException();
  }

  /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
  public boolean isSet(_Fields field) {
    if (field == null) {
      throw new IllegalArgumentException();
    }

    switch (field) {
    case HOST_NAME:
      return isSetHostName();
    case PORT_NO:
      return isSetPortNo();
    }
    throw new IllegalStateException();
  }

  @Override
  public boolean equals(Object that) {
    if (that == null)
      return false;
    if (that instanceof ServerName)
      return this.equals((ServerName)that);
    return false;
  }

  public boolean equals(ServerName that) {
    if (that == null)
      return false;

    boolean this_present_hostName = true && this.isSetHostName();
    boolean that_present_hostName = true && that.isSetHostName();
    if (this_present_hostName || that_present_hostName) {
      if (!(this_present_hostName && that_present_hostName))
        return false;
      if (!this.hostName.equals(that.hostName))
        return false;
    }

    boolean this_present_portNo = true;
    boolean that_present_portNo = true;
    if (this_present_portNo || that_present_portNo) {
      if (!(this_present_portNo && that_present_portNo))
        return false;
      if (this.portNo != that.portNo)
        return false;
    }

    return true;
  }

  @Override
  public int hashCode() {
    return 0;
  }

  public int compareTo(ServerName other) {
    if (!getClass().equals(other.getClass())) {
      return getClass().getName().compareTo(other.getClass().getName());
    }

    int lastComparison = 0;
    ServerName typedOther = (ServerName)other;

    lastComparison = Boolean.valueOf(isSetHostName()).compareTo(typedOther.isSetHostName());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetHostName()) {
      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.hostName, typedOther.hostName);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    lastComparison = Boolean.valueOf(isSetPortNo()).compareTo(typedOther.isSetPortNo());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetPortNo()) {
      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.portNo, typedOther.portNo);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    return 0;
  }

  public _Fields fieldForId(int fieldId) {
    return _Fields.findByThriftId(fieldId);
  }

  public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
    schemes.get(iprot.getScheme()).getScheme().read(iprot, this);
  }

  public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
    schemes.get(oprot.getScheme()).getScheme().write(oprot, this);
  }

  @Override
  public String toString() {
    StringBuilder sb = new StringBuilder("ServerName(");
    boolean first = true;

    sb.append("hostName:");
    if (this.hostName == null) {
      sb.append("null");
    } else {
      sb.append(this.hostName);
    }
    first = false;
    if (!first) sb.append(", ");
    sb.append("portNo:");
    sb.append(this.portNo);
    first = false;
    sb.append(")");
    return sb.toString();
  }

  public void validate() throws org.apache.thrift.TException {
    // check for required fields
    if (hostName == null) {
      throw new org.apache.thrift.protocol.TProtocolException("Required field 'hostName' was not present! Struct: " + toString());
    }
    // alas, we cannot check 'portNo' because it's a primitive and you chose the non-beans generator.
  }

  private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
    try {
      write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
    } catch (org.apache.thrift.TException te) {
      throw new java.io.IOException(te);
    }
  }

  private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
    try {
      // it doesn't seem like you should have to do this, but java serialization is wacky, and doesn't call the default constructor.
      __isset_bit_vector = new BitSet(1);
      read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
    } catch (org.apache.thrift.TException te) {
      throw new java.io.IOException(te);
    }
  }

  private static class ServerNameStandardSchemeFactory implements SchemeFactory {
    public ServerNameStandardScheme getScheme() {
      return new ServerNameStandardScheme();
    }
  }

  private static class ServerNameStandardScheme extends StandardScheme<ServerName> {

    public void read(org.apache.thrift.protocol.TProtocol iprot, ServerName struct) throws org.apache.thrift.TException {
      org.apache.thrift.protocol.TField schemeField;
      iprot.readStructBegin();
      while (true)
      {
        schemeField = iprot.readFieldBegin();
        if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { 
          break;
        }
        switch (schemeField.id) {
          case 1: // HOST_NAME
            if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
              struct.hostName = iprot.readString();
              struct.setHostNameIsSet(true);
            } else { 
              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
            }
            break;
          case 2: // PORT_NO
            if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
              struct.portNo = iprot.readI32();
              struct.setPortNoIsSet(true);
            } else { 
              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
            }
            break;
          default:
            org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
        }
        iprot.readFieldEnd();
      }
      iprot.readStructEnd();

      // check for required fields of primitive type, which can't be checked in the validate method
      if (!struct.isSetPortNo()) {
        throw new org.apache.thrift.protocol.TProtocolException("Required field 'portNo' was not found in serialized data! Struct: " + toString());
      }
      struct.validate();
    }

    public void write(org.apache.thrift.protocol.TProtocol oprot, ServerName struct) throws org.apache.thrift.TException {
      struct.validate();

      oprot.writeStructBegin(STRUCT_DESC);
      if (struct.hostName != null) {
        oprot.writeFieldBegin(HOST_NAME_FIELD_DESC);
        oprot.writeString(struct.hostName);
        oprot.writeFieldEnd();
      }
      oprot.writeFieldBegin(PORT_NO_FIELD_DESC);
      oprot.writeI32(struct.portNo);
      oprot.writeFieldEnd();
      oprot.writeFieldStop();
      oprot.writeStructEnd();
    }

  }

  private static class ServerNameTupleSchemeFactory implements SchemeFactory {
    public ServerNameTupleScheme getScheme() {
      return new ServerNameTupleScheme();
    }
  }

  private static class ServerNameTupleScheme extends TupleScheme<ServerName> {

    @Override
    public void write(org.apache.thrift.protocol.TProtocol prot, ServerName struct) throws org.apache.thrift.TException {
      TTupleProtocol oprot = (TTupleProtocol) prot;
      oprot.writeString(struct.hostName);
      oprot.writeI32(struct.portNo);
    }

    @Override
    public void read(org.apache.thrift.protocol.TProtocol prot, ServerName struct) throws org.apache.thrift.TException {
      TTupleProtocol iprot = (TTupleProtocol) prot;
      struct.hostName = iprot.readString();
      struct.setHostNameIsSet(true);
      struct.portNo = iprot.readI32();
      struct.setPortNoIsSet(true);
    }
  }

} public class ByteUtils {

	// ensuring non instantiability
	private ByteUtils() {

	}

	/**
	 * Size of boolean in bytes
	 */
	public static final int SIZEOF_BOOLEAN = Byte.SIZE / Byte.SIZE;

	/**
	 * Size of byte in bytes
	 */
	public static final int SIZEOF_BYTE = SIZEOF_BOOLEAN;

	/**
	 * Size of char in bytes
	 */
	public static final int SIZEOF_CHAR = Character.SIZE / Byte.SIZE;

	/**
	 * Size of double in bytes
	 */
	public static final int SIZEOF_DOUBLE = Double.SIZE / Byte.SIZE;

	/**
	 * Size of float in bytes
	 */
	public static final int SIZEOF_FLOAT = Float.SIZE / Byte.SIZE;

	/**
	 * Size of int in bytes
	 */
	public static final int SIZEOF_INT = Integer.SIZE / Byte.SIZE;

	/**
	 * Size of long in bytes
	 */
	public static final int SIZEOF_LONG = Long.SIZE / Byte.SIZE;

	/**
	 * Size of short in bytes
	 */
	public static final int SIZEOF_SHORT = Short.SIZE / Byte.SIZE;

	public static byte[] sha1(byte[] input) {
		return getDigest("SHA-1").digest(input);
	}

	public static MessageDigest getDigest(String algorithm) {
		try {
			return MessageDigest.getInstance(algorithm);
		} catch (NoSuchAlgorithmException e) {
			throw new IllegalStateException("Unknown algorithm: " + algorithm,
					e);
		}
	}

	/**
	 * @param left
	 *            left operand
	 * @param right
	 *            right operand
	 * @return 0 if equal, < 0 if left is less than right, etc.
	 */
	public static int compareTo(final byte[] left, final byte[] right) {
		return compareTo(left, 0, left.length, right, 0, right.length);
	}

	/**
	 * Lexographically compare two arrays.
	 * 
	 * @param buffer1
	 *            left operand
	 * @param buffer2
	 *            right operand
	 * @param offset1
	 *            Where to start comparing in the left buffer
	 * @param offset2
	 *            Where to start comparing in the right buffer
	 * @param length1
	 *            How much to compare from the left buffer
	 * @param length2
	 *            How much to compare from the right buffer
	 * @return 0 if equal, < 0 if left is less than right, etc.
	 */
	public static int compareTo(byte[] buffer1, int offset1, int length1,
			byte[] buffer2, int offset2, int length2) {
		// Bring WritableComparator code local
		int end1 = offset1 + length1;
		int end2 = offset2 + length2;
		for (int i = offset1, j = offset2; i < end1 && j < end2; i++, j++) {
			int a = (buffer1[i] & 0xff);
			int b = (buffer2[j] & 0xff);
			if (a != b) {
				return a - b;
			}
		}
		return length1 - length2;
	}

	/**
	 * Converts a byte array to a long value. Reverses {@link #toBytes(long)}
	 * 
	 * @param bytes
	 *            array
	 * @return the long value
	 */
	public static long toLong(byte[] bytes) {
		return toLong(bytes, 0, SIZEOF_LONG);
	}

	/**
	 * Converts a byte array to a long value. Assumes there will be
	 * {@link #SIZEOF_LONG} bytes available.
	 * 
	 * @param bytes
	 *            bytes
	 * @param offset
	 *            offset
	 * @return the long value
	 */
	public static long toLong(byte[] bytes, int offset) {
		return toLong(bytes, offset, SIZEOF_LONG);
	}

	/**
	 * Converts a byte array to a long value.
	 * 
	 * @param bytes
	 *            array of bytes
	 * @param offset
	 *            offset into array
	 * @param length
	 *            length of data (must be {@link #SIZEOF_LONG})
	 * @return the long value
	 * @throws IllegalArgumentException
	 *             if length is not {@link #SIZEOF_LONG} or if there's not
	 *             enough room in the array at the offset indicated.
	 */
	public static long toLong(byte[] bytes, int offset, final int length) {
		if (length != SIZEOF_LONG || offset + length > bytes.length) {
			throw explainWrongLengthOrOffset(bytes, offset, length, SIZEOF_LONG);
		}
		long l = 0;
		for (int i = offset; i < offset + length; i++) {
			l <<= 8;
			l ^= bytes[i] & 0xFF;
		}
		return l;
	}

	private static IllegalArgumentException explainWrongLengthOrOffset(
			final byte[] bytes, final int offset, final int length,
			final int expectedLength) {
		String reason;
		if (length != expectedLength) {
			reason = "Wrong length: " + length + ", expected " + expectedLength;
		} else {
			reason = "offset (" + offset + ") + length (" + length
					+ ") exceed the" + " capacity of the array: "
					+ bytes.length;
		}
		return new IllegalArgumentException(reason);
	}

	/**
	 * Copy the specified bytes into a new array
	 * 
	 * @param array
	 *            The array to copy from
	 * @param from
	 *            The index in the array to begin copying from
	 * @param to
	 *            The least index not copied
	 * @return A new byte[] containing the copied bytes
	 */
	public static byte[] copy(byte[] array, int from, int to) {
		if (to - from < 0) {
			return new byte[0];
		} else {
			byte[] a = new byte[to - from];
			System.arraycopy(array, from, a, 0, to - from);
			return a;
		}
	}

	public static int roundUpToPowerOf2(int number) {
		return (number > 1) ? Integer.highestOneBit((number - 1) << 1) : 1;
	}
} public static class QueueReachedMaxCapacityException extends
		RuntimeException {

	private static final long serialVersionUID = 1L;
	private static final String EXCEPTION_MSG = "Queue is full. Can not add more elements to the queue.";

	public QueueReachedMaxCapacityException() {
		super(EXCEPTION_MSG);
	}

	public QueueReachedMaxCapacityException(Throwable cause) {
		super(EXCEPTION_MSG, cause);
	}
}
import static org.hashtrees.test.utils.HashTreesImplTestUtils.DEFAULT_SEG_DATA_BLOCKS_COUNT;
import static org.hashtrees.test.utils.HashTreesImplTestUtils.DEFAULT_TREE_ID;
import static org.hashtrees.test.utils.HashTreesImplTestUtils.ROOT_NODE;
import static org.hashtrees.test.utils.HashTreesImplTestUtils.SEG_ID_PROVIDER;
import static org.hashtrees.test.utils.HashTreesImplTestUtils.TREE_ID_PROVIDER;
import static org.hashtrees.test.utils.HashTreesImplTestUtils.createHashTree;
import static org.hashtrees.test.utils.HashTreesImplTestUtils.generateInMemoryAndPersistentStores;
import static org.hashtrees.test.utils.HashTreesImplTestUtils.generateInMemoryStore;
import static org.hashtrees.test.utils.HashTreesImplTestUtils.generatePersistentStore;
import static org.hashtrees.test.utils.HashTreesImplTestUtils.randomByteBuffer;
import static org.hashtrees.test.utils.HashTreesImplTestUtils.randomBytes;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicIntegerArray;
import org.apache.commons.codec.binary.Hex;
import org.apache.thrift.transport.TTransportException;
import org.hashtrees.HashTrees;
import org.hashtrees.HashTreesImpl;
import org.hashtrees.HashTreesObserver;
import org.hashtrees.SimpleTreeIdProvider;
import org.hashtrees.SyncDiffResult;
import org.hashtrees.manager.EmptySyncListProvider;
import org.hashtrees.manager.HashTreesManager;
import org.hashtrees.manager.HashTreesRemoteClient;
import org.hashtrees.manager.HashTreesSynchListProvider;
import org.hashtrees.store.HashTreesMemStore;
import org.hashtrees.store.HashTreesPersistentStore;
import org.hashtrees.store.HashTreesStore;
import org.hashtrees.store.SimpleMemStore;
import org.hashtrees.store.Store;
import org.hashtrees.test.utils.HashTreesConstants;
import org.hashtrees.test.utils.HashTreesImplTestUtils;
import org.hashtrees.test.utils.HashTreesImplTestUtils.HTreeComponents;
import org.hashtrees.thrift.generated.KeyValue;
import org.hashtrees.thrift.generated.SegmentData;
import org.hashtrees.thrift.generated.SegmentHash;
import org.hashtrees.thrift.generated.ServerName;
import org.hashtrees.util.ByteUtils;
import org.hashtrees.util.NonBlockingQueuingTask.QueueReachedMaxCapacityException;
import org.junit.Assert;
import org.junit.Test;
/*
 * 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.
 */
package org.hashtrees.test;




public class HashTreesImplTest {

	private static byte[] generateBytesFrom(int... values) {
		byte[] result = new byte[values.length * ByteUtils.SIZEOF_INT];
		ByteBuffer bb = ByteBuffer.wrap(result);
		for (int value : values)
			bb.putInt(value);
		return result;
	}

	private static ByteBuffer generateRandomKeyWithPrefix(int prefixValue) {
		byte[] randomBytes = randomBytes();
		byte[] key = new byte[ByteUtils.SIZEOF_INT + randomBytes.length];
		ByteBuffer bb = ByteBuffer.wrap(key);
		bb.putInt(prefixValue);
		bb.put(randomBytes);
		return bb;
	}

	@Test
	public void testPut() throws IOException {

		HashTreesStore[] stores = generateInMemoryAndPersistentStores();

		try {
			for (HashTreesStore store : stores) {
				int segId = 1;

				HTreeComponents components = createHashTree(
						DEFAULT_SEG_DATA_BLOCKS_COUNT, false, TREE_ID_PROVIDER,
						SEG_ID_PROVIDER, store);
				HashTrees testTree = components.hTree;
				HashTreesStore testTreeStore = components.hTStore;

				ByteBuffer key = generateRandomKeyWithPrefix(segId);
				ByteBuffer value = ByteBuffer.wrap(randomBytes());
				testTree.hPut(key, value);
				ByteBuffer expectedDigest = ByteBuffer.wrap(ByteUtils
						.sha1(value.array()));

				SegmentData actualKeyAndDigest = testTree.getSegmentData(
						SimpleTreeIdProvider.TREE_ID, segId, key);
				Assert.assertNotNull(actualKeyAndDigest);
				Assert.assertTrue(Arrays.equals(key.array(),
						actualKeyAndDigest.getKey()));
				Assert.assertTrue(Arrays.equals(expectedDigest.array(),
						actualKeyAndDigest.getDigest()));

				List<Integer> dirtySegs = testTreeStore
						.getDirtySegments(SimpleTreeIdProvider.TREE_ID);
				Assert.assertEquals(1, dirtySegs.size());
				Assert.assertEquals(1, dirtySegs.get(0).intValue());
			}
		} finally {
			HashTreesImplTestUtils.closeStores(stores);
		}
	}

	@Test
	public void testRemove() throws IOException {

		HashTreesStore[] stores = generateInMemoryAndPersistentStores();
		int segId = 2;

		try {
			for (HashTreesStore store : stores) {
				HTreeComponents components = createHashTree(
						DEFAULT_SEG_DATA_BLOCKS_COUNT, false, TREE_ID_PROVIDER,
						SEG_ID_PROVIDER, store);
				HashTrees testTree = components.hTree;
				HashTreesStore testTreeStore = components.hTStore;

				ByteBuffer key = generateRandomKeyWithPrefix(segId);
				ByteBuffer value = ByteBuffer.wrap(randomBytes());
				testTree.hPut(key, value);
				testTree.hRemove(key);

				SegmentData segData = testTree.getSegmentData(
						SimpleTreeIdProvider.TREE_ID, segId, key);
				Assert.assertNull(segData);

				List<Integer> dirtySegs = testTreeStore
						.getDirtySegments(SimpleTreeIdProvider.TREE_ID);
				Assert.assertEquals(1, dirtySegs.size());
				Assert.assertEquals(2, dirtySegs.get(0).intValue());
			}
		} finally {
			HashTreesImplTestUtils.closeStores(stores);
		}
	}

	@Test
	public void testCompleteRebuild() throws IOException {
		int rootNodeId = 0;
		int nodeId = 2;
		int segId = 1;
		int noOfSegments = 2;
		HashTreesStore[] stores = generateInMemoryAndPersistentStores();

		try {
			for (HashTreesStore store : stores) {
				HTreeComponents components = createHashTree(noOfSegments,
						false, TREE_ID_PROVIDER, SEG_ID_PROVIDER, store);
				HashTrees testTree = components.hTree;
				SimpleMemStore kvStore = components.store;

				ByteBuffer expectedKey = generateRandomKeyWithPrefix(segId);
				ByteBuffer expectedValue = ByteBuffer.wrap(randomBytes());
				StringBuffer sb = new StringBuffer();
				ByteBuffer expectedDigest = ByteBuffer.wrap(ByteUtils
						.sha1(expectedValue.array()));
				sb.append(HashTreesImpl.getHexString(expectedKey,