public static List<Integer> getImmediateChildren(int parentId, int k) {
	List<Integer> children = new ArrayList<Integer>(2);
	for (int i = 1; i <= k; i++) {
		children.add((k * parentId) + i);
	}
	return children;
} public static int getLeftMostChildNode(int parentId, int k, int height) {
	int parentNodeLevel = getLevelOfNode(parentId, k);
	int noOfItr = height - parentNodeLevel;
	int leftMostChild = parentId;
	while (noOfItr > 0) {
		leftMostChild = k * leftMostChild + 1;
		noOfItr--;
	}
	return leftMostChild;
} public static int getNoOfNodes(int h, int k) {
	int result = (((int) Math.pow(k, h + 1)) - 1) / (k - 1);
	return result;
} public static int getParent(int childId, int k) {
	if (childId <= k)
		return 0;
	return (childId % k == 0) ? ((childId / k) - 1) : (childId / k);
} public static int getRightMostChildNode(int parentId, int k, int height) {
	int parentNodeLevel = getLevelOfNode(parentId, k);
	int noOfItr = height - parentNodeLevel;
	int rightMostChild = parentId;
	while (noOfItr > 0) {
		rightMostChild = k * rightMostChild + k;
		noOfItr--;
	}
	return rightMostChild;
} public static int height(int noOfLeaves, int k) {
	int height = -1;
	while (noOfLeaves > 0) {
		noOfLeaves /= k;
		height++;
	}
	return height;
} public static int roundUpToPowerOf2(int number) {
	return (number > 1) ? Integer.highestOneBit((number - 1) << 1) : 1;
} public static byte[] sha1(byte[] input) {
	return getDigest("SHA-1").digest(input);
} @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 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 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 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 abstract class NonBlockingQueuingTask<T> extends StoppableTask {

	// A special marker to note down a stop operation has been requested on this
	// thread.
	private final T stopMarker;
	private final BlockingQueue<T> que;

	/**
	 * @param stopMarker
	 *            specifies a special data which is used for indicating stop
	 *            operation. Subclasses should not use this marker for any
	 *            operation.
	 * @param queueSize
	 *            how much data can the queue can hold. Should be higher enough
	 *            based on the number of producers, otherwise exception will be
	 *            thrown on adding the elements to the queue.
	 */
	public NonBlockingQueuingTask(T stopMarker, int queueSize) {
		this.stopMarker = stopMarker;
		que = new LinkedBlockingQueue<T>(queueSize);
	}

	public void enque(T element) {
		if (hasStopRequested() && element != stopMarker) {
			throw new QueuingTaskIsStoppedException();
		}
		boolean status = que.offer(element);
		if (!status)
			throw new QueueReachedMaxCapacityException();
	}

	@Override
	public synchronized void stopAsync() {
		super.stopAsync();
		enque(stopMarker);
	}

	@Override
	public synchronized void stopAsync(final CountDownLatch shutDownLatch) {
		super.stopAsync(shutDownLatch);
		enque(stopMarker);
	}

	@Override
	public void runImpl() {
		for (;;) {
			try {
				T data = que.take();
				if (data == stopMarker)
					continue;
				handleElement(data);
			} catch (InterruptedException e) {
				throw new RuntimeException(
						"Exception occurred while removing the element from the queue",
						e);
			} finally {
				if (hasStopRequested() && que.isEmpty()) {
					return;
				}
			}
		}
	}

	protected abstract void handleElement(T element);

	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);
		}
	}

	public static class QueuingTaskIsStoppedException extends RuntimeException {

		private static final long serialVersionUID = 1L;
		private static final String EXCEPTION_MSG = "Queuing task is stopped. Can not add elements to the queue.";

		public QueuingTaskIsStoppedException() {
			super(EXCEPTION_MSG);
		}

		public QueuingTaskIsStoppedException(Throwable cause) {
			super(EXCEPTION_MSG, cause);
		}
	}
} public class Pair<F, S> {

	private final F f;
	private final S s;

	public Pair(F f, S s) {
		this.f = f;
		this.s = s;
	}

	public static <F, S> Pair<F, S> create(F f, S s) {
		return new Pair<F, S>(f, s);
	}

	public F getFirst() {
		return f;
	}

	public S getSecond() {
		return s;
	}

	@Override
	public boolean equals(Object that) {
		if (that == null || !(that instanceof Pair<?, ?>))
			return false;
		Pair<?, ?> thatObj = (Pair<?, ?>) that;
		return Objects.equal(f, thatObj.f) && Objects.equal(s, thatObj.s);
	}

	@Override
	public int hashCode() {
		return Objects.hashCode(f, s);
	}

	@Override
	public String toString() {
		return MoreObjects.toStringHelper(this).add("first", f)
				.add("second", s).toString();
	}
} public interface Service {

	void start();

	void stop();

}
import static org.hashtrees.TreeUtils.getImmediateChildren;
import static org.hashtrees.TreeUtils.getLeftMostChildNode;
import static org.hashtrees.TreeUtils.getNoOfNodes;
import static org.hashtrees.TreeUtils.getParent;
import static org.hashtrees.TreeUtils.getRightMostChildNode;
import static org.hashtrees.TreeUtils.height;
import static org.hashtrees.util.ByteUtils.roundUpToPowerOf2;
import static org.hashtrees.util.ByteUtils.sha1;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CountDownLatch;
import javax.annotation.concurrent.NotThreadSafe;
import javax.annotation.concurrent.ThreadSafe;
import org.apache.commons.codec.binary.Hex;
import org.hashtrees.store.HashTreesMemStore;
import org.hashtrees.store.HashTreesPersistentStore;
import org.hashtrees.store.HashTreesStore;
import org.hashtrees.store.Store;
import org.hashtrees.thrift.generated.KeyValue;
import org.hashtrees.thrift.generated.SegmentData;
import org.hashtrees.thrift.generated.SegmentHash;
import org.hashtrees.util.ByteUtils;
import org.hashtrees.util.LockedBy;
import org.hashtrees.util.NonBlockingQueuingTask;
import org.hashtrees.util.Pair;
import org.hashtrees.util.Service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.collect.Iterators;
import com.google.common.collect.PeekingIterator;
/*
 * 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;






/**
 * HashTrees has segment blocks and segment trees.
 * 
 * 1) Segment blocks, where the (key, hash of value) pairs are stored. All the
 * pairs are stored in sorted order. Whenever a key addition/removal happens on
 * the node, HashTree segment is updated. Keys are distributed using uniform
 * hash distribution.
 * 
 * 2) Segment trees, where the segments' hashes are updated and maintained. Tree
 * is not updated on every update on a segment. Rather, tree update is happening
 * at regular intervals. Binary tree is used currently. It should be easier to
 * extend to support higher no of children.
 * 
 */
@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;