// Path: src/main/java/com/knewton/mapreduce/constant/PropertyConstants.java
// public enum PropertyConstants {
// 
//     /**
//      * The Keyspace Name for a data run
//      */
//     KEYSPACE_NAME("com.knewton.inputformat.cassandra.keyspace"),
//     /**
//      * Column family type needs to be set if the column family type is Super.
//      */
//     COLUMN_FAMILY_TYPE("com.knewton.cassandra.cftype"),
//     /**
//      * Comparator class name for columns.
//      */
//     COLUMN_COMPARATOR("com.knewton.cassandra.column.comparator"),
//     /**
//      * A nonstandard comparator requires a subcomparator
//      */
//     COLUMN_SUBCOMPARATOR("com.knewton.cassandra.column.subcomparator"),
//     /**
//      * Partitioner for decorating keys.
//      */
//     PARTITIONER("com.knewton.partitioner"),
//     /**
//      * True if the columns are sparse, false if they're dense.
//      */
//     SPARSE_COLUMN("com.knewton.column.sparse"),
//     /**
//      * Boolean variable. Set true if compression enabled.
//      */
//     COMPRESSION_ENABLED("com.knewton.cassandra.backup.compression"),
//     /**
//      * Buffer size for decompression operation.
//      */
//     DECOMPRESS_BUFFER("com.knewton.cassandra.backup.compress.buffersize"),
//     /**
//      * Start and end dates for student events processing
//      */
//     START_DATE("com.knewton.studentevents.date.start"),
//     END_DATE("com.knewton.studentevents.date.end"),
//     /**
//      * The Column Family Name for a data run
//      */
//     COLUMN_FAMILY_NAME("com.knewton.inputformat.cassandra.columnfamily"),
//     /**
//      * MapReduce run environment. Should be one of the enum values in MREnvironment.java
//      */
//     MAPREDUCE_ENVIRONMENT("com.knewton.mapreduce.environment"),
//     /**
//      * Name of serialization factory class
//      */
//     SERIALIZATION_FACTORY_PARAMETER("com.knewton.thrift.serialization.protocol");
// 
//     public final String txt;
// 
//     private PropertyConstants(String propertyName) {
//         this.txt = propertyName;
//     }
// }


import com.knewton.mapreduce.constant.PropertyConstants;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.BufferCounterCell;
import org.apache.cassandra.db.BufferDecoratedKey;
import org.apache.cassandra.db.ColumnSerializer.Flag;
import org.apache.cassandra.db.CounterCell;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.composites.CellName;
import org.apache.cassandra.db.composites.CellNameType;
import org.apache.cassandra.db.composites.SimpleDenseCellNameType;
import org.apache.cassandra.db.context.CounterContext;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.OrderPreservingPartitioner.StringToken;
import org.apache.cassandra.dht.RandomPartitioner;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.Descriptor.Type;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.utils.CounterId;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.Spy;
import org.mockito.runners.MockitoJUnitRunner;
import java.io.File;
import java.nio.ByteBuffer;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
            .openSSTableReader(any(IPartitioner.class), any(CFMetaData.class));

        when(ssTableReader.estimatedKeys()).thenReturn(2L);
        when(ssTableReader.getScanner()).thenReturn(tableScanner);

        when(tableScanner.hasNext()).thenReturn(true, true, false);

        key = new BufferDecoratedKey(new StringToken("a"), ByteBuffer.wrap("b".getBytes()));
        CellNameType simpleDenseCellType = new SimpleDenseCellNameType(BytesType.instance);
        CellName cellName = simpleDenseCellType.cellFromByteBuffer(ByteBuffer.wrap("n".getBytes()));
        ByteBuffer counterBB = CounterContext.instance()
            .createGlobal(CounterId.fromInt(0), System.currentTimeMillis(), 123L);
        value = BufferCounterCell.create(cellName, counterBB, System.currentTimeMillis(), 0L,
                                         Flag.PRESERVE_SIZE);

        SSTableIdentityIterator row1 = getNewRow();
        SSTableIdentityIterator row2 = getNewRow();

        when(tableScanner.next()).thenReturn(row1, row2);
    }

    private SSTableIdentityIterator getNewRow() {
        SSTableIdentityIterator row = mock(SSTableIdentityIterator.class);
        when(row.hasNext()).thenReturn(true, true, false);
        when(row.getKey()).thenReturn(key);
        when(row.next()).thenReturn(value);
        return row;
    }

    private TaskAttemptContext getTaskAttemptContext() {
// Path: src/main/java/com/knewton/mapreduce/constant/PropertyConstants.java
// public enum PropertyConstants {
// 
//     /**
//      * The Keyspace Name for a data run
//      */
//     KEYSPACE_NAME("com.knewton.inputformat.cassandra.keyspace"),
//     /**
//      * Column family type needs to be set if the column family type is Super.
//      */
//     COLUMN_FAMILY_TYPE("com.knewton.cassandra.cftype"),
//     /**
//      * Comparator class name for columns.
//      */
//     COLUMN_COMPARATOR("com.knewton.cassandra.column.comparator"),
//     /**
//      * A nonstandard comparator requires a subcomparator
//      */
//     COLUMN_SUBCOMPARATOR("com.knewton.cassandra.column.subcomparator"),
//     /**
//      * Partitioner for decorating keys.
//      */
//     PARTITIONER("com.knewton.partitioner"),
//     /**
//      * True if the columns are sparse, false if they're dense.
//      */
//     SPARSE_COLUMN("com.knewton.column.sparse"),
//     /**
//      * Boolean variable. Set true if compression enabled.
//      */
//     COMPRESSION_ENABLED("com.knewton.cassandra.backup.compression"),
//     /**
//      * Buffer size for decompression operation.
//      */
//     DECOMPRESS_BUFFER("com.knewton.cassandra.backup.compress.buffersize"),
//     /**
//      * Start and end dates for student events processing
//      */
//     START_DATE("com.knewton.studentevents.date.start"),
//     END_DATE("com.knewton.studentevents.date.end"),
//     /**
//      * The Column Family Name for a data run
//      */
//     COLUMN_FAMILY_NAME("com.knewton.inputformat.cassandra.columnfamily"),
//     /**
//      * MapReduce run environment. Should be one of the enum values in MREnvironment.java
//      */
//     MAPREDUCE_ENVIRONMENT("com.knewton.mapreduce.environment"),
//     /**
//      * Name of serialization factory class
//      */
//     SERIALIZATION_FACTORY_PARAMETER("com.knewton.thrift.serialization.protocol");
// 
//     public final String txt;
// 
//     private PropertyConstants(String propertyName) {
//         this.txt = propertyName;
//     }
// }


// Path: src/test/java/com/knewton/mapreduce/SSTableColumnRecordReaderTest.java
import com.knewton.mapreduce.constant.PropertyConstants;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.BufferCounterCell;
import org.apache.cassandra.db.BufferDecoratedKey;
import org.apache.cassandra.db.ColumnSerializer.Flag;
import org.apache.cassandra.db.CounterCell;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.composites.CellName;
import org.apache.cassandra.db.composites.CellNameType;
import org.apache.cassandra.db.composites.SimpleDenseCellNameType;
import org.apache.cassandra.db.context.CounterContext;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.OrderPreservingPartitioner.StringToken;
import org.apache.cassandra.dht.RandomPartitioner;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.Descriptor.Type;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.utils.CounterId;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.Spy;
import org.mockito.runners.MockitoJUnitRunner;
import java.io.File;
import java.nio.ByteBuffer;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

            .openSSTableReader(any(IPartitioner.class), any(CFMetaData.class));

        when(ssTableReader.estimatedKeys()).thenReturn(2L);
        when(ssTableReader.getScanner()).thenReturn(tableScanner);

        when(tableScanner.hasNext()).thenReturn(true, true, false);

        key = new BufferDecoratedKey(new StringToken("a"), ByteBuffer.wrap("b".getBytes()));
        CellNameType simpleDenseCellType = new SimpleDenseCellNameType(BytesType.instance);
        CellName cellName = simpleDenseCellType.cellFromByteBuffer(ByteBuffer.wrap("n".getBytes()));
        ByteBuffer counterBB = CounterContext.instance()
            .createGlobal(CounterId.fromInt(0), System.currentTimeMillis(), 123L);
        value = BufferCounterCell.create(cellName, counterBB, System.currentTimeMillis(), 0L,
                                         Flag.PRESERVE_SIZE);

        SSTableIdentityIterator row1 = getNewRow();
        SSTableIdentityIterator row2 = getNewRow();

        when(tableScanner.next()).thenReturn(row1, row2);
    }

    private SSTableIdentityIterator getNewRow() {
        SSTableIdentityIterator row = mock(SSTableIdentityIterator.class);
        when(row.hasNext()).thenReturn(true, true, false);
        when(row.getKey()).thenReturn(key);
        when(row.next()).thenReturn(value);
        return row;
    }

    private TaskAttemptContext getTaskAttemptContext() {
        conf.set(PropertyConstants.COLUMN_COMPARATOR.txt, LongType.class.getName());