// Path: src/main/java/com/mozilla/bagheera/util/IdUtil.java
// public class IdUtil {
// 
// 	public static final SimpleDateFormat SDF = new SimpleDateFormat("yyyyMMdd");
// 
// 	/**
// 	 * @param id
// 	 * @param timestamp
// 	 * @return
// 	 * @throws IOException
// 	 */
// 	public static byte[] bucketizeId(String id, long timestamp) throws IOException {
//         if (id == null) {
//             throw new IllegalArgumentException("id cannot be null");
//         }
//         
//         return nonRandByteBucketizeId(id, new Date(timestamp));
//     }
// 	
// 	/**
// 	 * Takes a given id and prefixes it with a byte character and the date in a non-random fashion
// 	 * This method expects id to be something like a UUID consisting only of hex characters.  If id 
// 	 * is something else this will produce unpredictable results.
// 	 * @param id
// 	 * @param d
// 	 * @return
// 	 * @throws IOException
// 	 */
// 	public static byte[] nonRandByteBucketizeId(String id, Date d) throws IOException {
// 		if (StringUtils.isBlank(id)) {
// 			throw new IllegalArgumentException("id cannot be null or empty");
// 		}
//         if (d == null) {
//             throw new IllegalArgumentException("date cannot be null");
//         }
//         
//         // bucket byte + SDF bytes + id bytes
//         ByteBuffer buf = ByteBuffer.allocate(9 + id.length());
// 		int bucket = 0;
// 		if (id.length() >= 2) {
// 		    // Munge two hex characters into the range of a single byte
// 		    bucket = Integer.parseInt(id.substring(0, 2), 16) - 128;
// 		} else {
// 		    bucket = Integer.parseInt(id, 16) - 128;
// 		}
// 		buf.put((byte)bucket);
// 		buf.put(SDF.format(d).getBytes());
// 		buf.put(id.getBytes());
// 
// 		return buf.array();
// 	}
// 
// 	/**
// 	 * Takes a given id and gives you an hbase shell compatible string that you can
// 	 * use for get command.
// 	 * @param id
// 	 * @param d
// 	 * @return
// 	 * @throws IOException
// 	 */
// 	public static String hbaseShellId(String id, Date d) throws IOException {
// 	    byte[] idBytes = IdUtil.nonRandByteBucketizeId(id, d);
// 	    StringBuilder sb = new StringBuilder("\"\\x");
// 	    sb.append(String.format("%02x", idBytes[0]));
// 	    sb.append((new String(idBytes)).substring(1));
// 	    sb.append("\"");
// 	    return sb.toString();
// 	}
// 	
// 	public static String hbaseShellId(byte[] idBytes) throws IOException {
//         StringBuilder sb = new StringBuilder("\"\\x");
//         sb.append(String.format("%02x", idBytes[0]));
//         sb.append((new String(idBytes)).substring(1));
//         sb.append("\"");
//         return sb.toString();
//     }
// 	
// 	public static void main(String[] args) throws IOException {
// 	    String id = "06b87727-2946-4b1e-b654-8bafb60bb995";
// 	    long timestamp = 1360493115891L;
// 	    Calendar cal = Calendar.getInstance();
// 	    //cal.set(2013, Calendar.FEBRUARY, 10);
// 	    cal.setTimeInMillis(timestamp);
// 	    System.out.println(IdUtil.hbaseShellId(id, cal.getTime()));
// 	}
// }


import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Row;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import com.mozilla.bagheera.util.IdUtil;
import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Gauge;
import com.yammer.metrics.core.Meter;
import com.yammer.metrics.core.MetricName;
import com.yammer.metrics.core.Timer;
import com.yammer.metrics.core.TimerContext;
    @Override
    public void store(String key, byte[] data) throws IOException {
        if (!isOversized(key, data)) {
            Put p = new Put(Bytes.toBytes(key));
            p.add(family, qualifier, data);
            rowQueue.add(p);
            if (rowQueueSize.incrementAndGet() >= batchSize) {
                flush();
            }
        }
    }


    // There is a max size for 'data', exceeding it causes
    //   java.lang.IllegalArgumentException: KeyValue size too large
    // Detect, log, and reject it.
    private boolean isOversized(String key, byte[] data) {
        boolean tooBig = false;
        if (data != null && data.length > maxKeyValueSize) {
            LOG.warn(String.format("Storing key '%s': Data exceeds max length (%d > %d)",
                    key, data.length, maxKeyValueSize));
            oversized.mark();
            tooBig = true;
        }
        return tooBig;
    }

    @Override
    public void store(String key, byte[] data, long timestamp) throws IOException {
        if (!isOversized(key, data)) {
// Path: src/main/java/com/mozilla/bagheera/util/IdUtil.java
// public class IdUtil {
// 
// 	public static final SimpleDateFormat SDF = new SimpleDateFormat("yyyyMMdd");
// 
// 	/**
// 	 * @param id
// 	 * @param timestamp
// 	 * @return
// 	 * @throws IOException
// 	 */
// 	public static byte[] bucketizeId(String id, long timestamp) throws IOException {
//         if (id == null) {
//             throw new IllegalArgumentException("id cannot be null");
//         }
//         
//         return nonRandByteBucketizeId(id, new Date(timestamp));
//     }
// 	
// 	/**
// 	 * Takes a given id and prefixes it with a byte character and the date in a non-random fashion
// 	 * This method expects id to be something like a UUID consisting only of hex characters.  If id 
// 	 * is something else this will produce unpredictable results.
// 	 * @param id
// 	 * @param d
// 	 * @return
// 	 * @throws IOException
// 	 */
// 	public static byte[] nonRandByteBucketizeId(String id, Date d) throws IOException {
// 		if (StringUtils.isBlank(id)) {
// 			throw new IllegalArgumentException("id cannot be null or empty");
// 		}
//         if (d == null) {
//             throw new IllegalArgumentException("date cannot be null");
//         }
//         
//         // bucket byte + SDF bytes + id bytes
//         ByteBuffer buf = ByteBuffer.allocate(9 + id.length());
// 		int bucket = 0;
// 		if (id.length() >= 2) {
// 		    // Munge two hex characters into the range of a single byte
// 		    bucket = Integer.parseInt(id.substring(0, 2), 16) - 128;
// 		} else {
// 		    bucket = Integer.parseInt(id, 16) - 128;
// 		}
// 		buf.put((byte)bucket);
// 		buf.put(SDF.format(d).getBytes());
// 		buf.put(id.getBytes());
// 
// 		return buf.array();
// 	}
// 
// 	/**
// 	 * Takes a given id and gives you an hbase shell compatible string that you can
// 	 * use for get command.
// 	 * @param id
// 	 * @param d
// 	 * @return
// 	 * @throws IOException
// 	 */
// 	public static String hbaseShellId(String id, Date d) throws IOException {
// 	    byte[] idBytes = IdUtil.nonRandByteBucketizeId(id, d);
// 	    StringBuilder sb = new StringBuilder("\"\\x");
// 	    sb.append(String.format("%02x", idBytes[0]));
// 	    sb.append((new String(idBytes)).substring(1));
// 	    sb.append("\"");
// 	    return sb.toString();
// 	}
// 	
// 	public static String hbaseShellId(byte[] idBytes) throws IOException {
//         StringBuilder sb = new StringBuilder("\"\\x");
//         sb.append(String.format("%02x", idBytes[0]));
//         sb.append((new String(idBytes)).substring(1));
//         sb.append("\"");
//         return sb.toString();
//     }
// 	
// 	public static void main(String[] args) throws IOException {
// 	    String id = "06b87727-2946-4b1e-b654-8bafb60bb995";
// 	    long timestamp = 1360493115891L;
// 	    Calendar cal = Calendar.getInstance();
// 	    //cal.set(2013, Calendar.FEBRUARY, 10);
// 	    cal.setTimeInMillis(timestamp);
// 	    System.out.println(IdUtil.hbaseShellId(id, cal.getTime()));
// 	}
// }


// Path: src/main/java/com/mozilla/bagheera/sink/HBaseSink.java
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Row;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import com.mozilla.bagheera.util.IdUtil;
import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Gauge;
import com.yammer.metrics.core.Meter;
import com.yammer.metrics.core.MetricName;
import com.yammer.metrics.core.Timer;
import com.yammer.metrics.core.TimerContext;

    @Override
    public void store(String key, byte[] data) throws IOException {
        if (!isOversized(key, data)) {
            Put p = new Put(Bytes.toBytes(key));
            p.add(family, qualifier, data);
            rowQueue.add(p);
            if (rowQueueSize.incrementAndGet() >= batchSize) {
                flush();
            }
        }
    }


    // There is a max size for 'data', exceeding it causes
    //   java.lang.IllegalArgumentException: KeyValue size too large
    // Detect, log, and reject it.
    private boolean isOversized(String key, byte[] data) {
        boolean tooBig = false;
        if (data != null && data.length > maxKeyValueSize) {
            LOG.warn(String.format("Storing key '%s': Data exceeds max length (%d > %d)",
                    key, data.length, maxKeyValueSize));
            oversized.mark();
            tooBig = true;
        }
        return tooBig;
    }

    @Override
    public void store(String key, byte[] data, long timestamp) throws IOException {
        if (!isOversized(key, data)) {
            byte[] k = prefixDate ? IdUtil.bucketizeId(key, timestamp) : Bytes.toBytes(key);