// Path: src/main/java/io/izenecloud/couchbase/CouchbaseInputFormat.java
// static class CouchbaseSplit extends InputSplit implements Writable {
// 	final List<Integer> vbuckets;
// 
// 	CouchbaseSplit() {
// 		vbuckets = new ArrayList<Integer>();
// 	}
// 
// 	CouchbaseSplit(List<Integer> vblist) {
// 		vbuckets = vblist;
// 	}
// 
// 	public void readFields(DataInput in) throws IOException {
// 		short numvbuckets = in.readShort();
// 		for (int i = 0; i < numvbuckets; i++) {
// 			vbuckets.add(new Integer(in.readShort()));
// 		}
// 	}
// 
// 	public void write(DataOutput out) throws IOException {
// 		out.writeShort(vbuckets.size());
// 		for (Integer v : vbuckets) {
// 			out.writeShort(v.shortValue());
// 		}
// 	}
// 
// 	public long getLength() throws IOException {
// 		return vbuckets.size();
// 	}
// 
// 	public String[] getLocations() {
// 		return new String[0];
// 	}
// }


import io.izenecloud.couchbase.CouchbaseInputFormat.CouchbaseSplit;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import javax.naming.ConfigurationException;
import net.spy.memcached.tapmessage.RequestMessage;
import net.spy.memcached.tapmessage.ResponseMessage;
import net.spy.memcached.tapmessage.TapMagic;
import net.spy.memcached.tapmessage.TapOpcode;
import net.spy.memcached.tapmessage.TapRequestFlag;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import com.couchbase.client.TapClient;
			k.set(getCurrentKey());
			v.set(getCurrentValue());
			return true;
		}
		return false;
	}

	@Override
	public void initialize(InputSplit split, TaskAttemptContext context)
			throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		final URI ClusterURI;
		try {
			ClusterURI = new URI(conf.get(CouchbaseConfig.CB_INPUT_CLUSTER));
		} catch (URISyntaxException e) {
			throw new IOException(e);
		}
		final List<URI> ClientURIList = new ArrayList<URI>();
		ClientURIList.add(ClusterURI.resolve("/pools"));
		final String bucket = conf.get(CouchbaseConfig.CB_INPUT_BUCKET,
				"");
		final String password = conf.get(CouchbaseConfig.CB_INPUT_PASSWORD, "");

		RequestMessage tapReq = new RequestMessage();
		tapReq.setMagic(TapMagic.PROTOCOL_BINARY_REQ);
		tapReq.setOpcode(TapOpcode.REQUEST);
		tapReq.setFlags(TapRequestFlag.DUMP);
		tapReq.setFlags(TapRequestFlag.SUPPORT_ACK);
		tapReq.setFlags(TapRequestFlag.FIX_BYTEORDER);
		tapReq.setFlags(TapRequestFlag.LIST_VBUCKETS);
// Path: src/main/java/io/izenecloud/couchbase/CouchbaseInputFormat.java
// static class CouchbaseSplit extends InputSplit implements Writable {
// 	final List<Integer> vbuckets;
// 
// 	CouchbaseSplit() {
// 		vbuckets = new ArrayList<Integer>();
// 	}
// 
// 	CouchbaseSplit(List<Integer> vblist) {
// 		vbuckets = vblist;
// 	}
// 
// 	public void readFields(DataInput in) throws IOException {
// 		short numvbuckets = in.readShort();
// 		for (int i = 0; i < numvbuckets; i++) {
// 			vbuckets.add(new Integer(in.readShort()));
// 		}
// 	}
// 
// 	public void write(DataOutput out) throws IOException {
// 		out.writeShort(vbuckets.size());
// 		for (Integer v : vbuckets) {
// 			out.writeShort(v.shortValue());
// 		}
// 	}
// 
// 	public long getLength() throws IOException {
// 		return vbuckets.size();
// 	}
// 
// 	public String[] getLocations() {
// 		return new String[0];
// 	}
// }


// Path: src/main/java/io/izenecloud/couchbase/CouchbaseRecordReader.java
import io.izenecloud.couchbase.CouchbaseInputFormat.CouchbaseSplit;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import javax.naming.ConfigurationException;
import net.spy.memcached.tapmessage.RequestMessage;
import net.spy.memcached.tapmessage.ResponseMessage;
import net.spy.memcached.tapmessage.TapMagic;
import net.spy.memcached.tapmessage.TapOpcode;
import net.spy.memcached.tapmessage.TapRequestFlag;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import com.couchbase.client.TapClient;

			k.set(getCurrentKey());
			v.set(getCurrentValue());
			return true;
		}
		return false;
	}

	@Override
	public void initialize(InputSplit split, TaskAttemptContext context)
			throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		final URI ClusterURI;
		try {
			ClusterURI = new URI(conf.get(CouchbaseConfig.CB_INPUT_CLUSTER));
		} catch (URISyntaxException e) {
			throw new IOException(e);
		}
		final List<URI> ClientURIList = new ArrayList<URI>();
		ClientURIList.add(ClusterURI.resolve("/pools"));
		final String bucket = conf.get(CouchbaseConfig.CB_INPUT_BUCKET,
				"");
		final String password = conf.get(CouchbaseConfig.CB_INPUT_PASSWORD, "");

		RequestMessage tapReq = new RequestMessage();
		tapReq.setMagic(TapMagic.PROTOCOL_BINARY_REQ);
		tapReq.setOpcode(TapOpcode.REQUEST);
		tapReq.setFlags(TapRequestFlag.DUMP);
		tapReq.setFlags(TapRequestFlag.SUPPORT_ACK);
		tapReq.setFlags(TapRequestFlag.FIX_BYTEORDER);
		tapReq.setFlags(TapRequestFlag.LIST_VBUCKETS);
		final CouchbaseSplit couchSplit = (CouchbaseSplit) split;