package com.yahoo.ads.pb.store;

import java.util.Iterator;
import java.util.Map;
import java.util.WeakHashMap;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.ByteBufferOutput;
import com.esotericsoftware.kryo.io.Input;
import com.yahoo.ads.pb.kafka.KeyValue;

import kyotocabinet.Cursor;

public class PistachiosTkIterator implements Iterator{
	public volatile static Map<Long,PistachiosTkIterator> iteratorMap = new WeakHashMap<Long,PistachiosTkIterator>();
	private Cursor cursor;
	private Kryo kryo = new Kryo();
	private static final ThreadLocal<ByteBufferOutput> threadByteBuffer =
      new ThreadLocal<ByteBufferOutput>() {
          @Override protected ByteBufferOutput initialValue() {
              return new ByteBufferOutput(10240);
          }
      };
	public static PistachiosTkIterator getPistachiosTkIterator(long id){
		if(iteratorMap.get(id) == null){
			synchronized(iteratorMap){
				if(iteratorMap.get(id) == null){
					iteratorMap.put(id, new PistachiosTkIterator());
				}
			}
		}
		return iteratorMap.get(id);
	}
	
	public boolean isCursorSet() {
		return cursor == null ? false : true;
	}
	public void setCursor(Cursor cursor) {
		this.cursor = cursor;
		cursor.jump();
	}
	
	@Override
    public boolean hasNext() {
		if(cursor.step()){
			cursor.step_back();
			return true;
		}else{
			return false;
		}
    }

	@Override
    public Object next() {
		if(cursor.step()){
			byte[] key =  cursor.get_key(false);
			byte[] value = cursor.get_value(false);
			KeyValue keyValue = new KeyValue();
			keyValue.key = key;
			keyValue.value = value;
			threadByteBuffer.get().clear();
			kryo.writeObject(threadByteBuffer.get(), keyValue);
			return threadByteBuffer.get().toBytes();
		}
	    return null;
    }

	@Override
    public void remove() {
		cursor.remove();
    }

	public void jump(byte[] key){
		cursor.jump(key);
	}
}

package com.yahoo.ads.pb.store;

import java.util.Iterator;
import java.util.Map;
import java.util.WeakHashMap;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.ByteBufferOutput;
import com.esotericsoftware.kryo.io.Input;
import com.yahoo.ads.pb.kafka.KeyValue;

import kyotocabinet.Cursor;

public class PistachiosTkIterator implements Iterator{
	public volatile static Map<Long,PistachiosTkIterator> iteratorMap = new WeakHashMap<Long,PistachiosTkIterator>();
	private Cursor cursor;
	private Kryo kryo = new Kryo();
	private static final ThreadLocal<ByteBufferOutput> threadByteBuffer =
      new ThreadLocal<ByteBufferOutput>() {
          @Override protected ByteBufferOutput initialValue() {
              return new ByteBufferOutput(10240);
          }
      };
	public static PistachiosTkIterator getPistachiosTkIterator(long id){
		if(iteratorMap.get(id) == null){
			synchronized(iteratorMap){
				if(iteratorMap.get(id) == null){
					iteratorMap.put(id, new PistachiosTkIterator());
				}
			}
		}
		return iteratorMap.get(id);
	}
	
	public boolean isCursorSet() {
		return cursor == null ? false : true;
	}
	public void setCursor(Cursor cursor) {
		this.cursor = cursor;
		cursor.jump();
	}
	
	@Override
    public boolean hasNext() {
		if(cursor.step()){
			cursor.step_back();
			return true;
		}else{
			return false;
		}
    }

	@Override
    public Object next() {
			byte[] key =  cursor.get_key(false);
			byte[] value = cursor.get_value(false);
			KeyValue keyValue = new KeyValue();
			keyValue.key = key;
			keyValue.value = value;
			threadByteBuffer.get().clear();
			kryo.writeObject(threadByteBuffer.get(), keyValue);
			cursor.step();
			return threadByteBuffer.get().toBytes();
    }

	@Override
    public void remove() {
		cursor.remove();
    }

	public void jump(byte[] key){
		cursor.jump(key);
	}
}

fix a bug