test/unit/org/apache/cassandra/streaming/StreamingTransferTest.java
/**
     * Create and transfer a single sstable, and return the keys that should have been transferred.
     * The Mutator must create the given column, but it may also create any other columns it pleases.
     */
Create and transfer a single sstable, and return the keys that should have been transferred.
The Mutator must create the given column, but it may also create any other columns it pleases.
Create and transfer a single sstable and return the keys that should have been transferred . The Mutator must create the given column but it may also create any other columns it pleases .
private List<String> createAndTransfer(Table table, ColumnFamilyStore cfs, Mutator mutator) throws Exception
    {
        logger.debug("Mutating " + cfs.columnFamily);
        long timestamp = 1234;
        for (int i = 1; i <= 3; i++)
            mutator.mutate("key" + i, "col" + i, timestamp);
        cfs.forceBlockingFlush();
        Util.compactAll(cfs).get();
        assertEquals(1, cfs.getSSTables().size());
        SSTableReader sstable = cfs.getSSTables().iterator().next();
        cfs.clearUnsafe();
        logger.debug("Transferring " + cfs.columnFamily);
        transfer(table, sstable);
        assertEquals(1, cfs.getSSTables().size());
        int[] offs = new int[]{1, 3};
        List<Row> rows = Util.getRangeSlice(cfs);
        assertEquals(offs.length, rows.size());
        for (int i = 0; i < offs.length; i++)
        {
            String key = "key" + offs[i];
            String col = "col" + offs[i];
            assert null != cfs.getColumnFamily(QueryFilter.getIdentityFilter(Util.dk(key),
                                               new QueryPath(cfs.columnFamily)));
            assert rows.get(i).key.key.equals(ByteBufferUtil.bytes(key));
            assert rows.get(i).cf.getColumn(ByteBufferUtil.bytes(col)) != null;
        }
        assertEquals(timestamp, cfs.getSSTables().iterator().next().getMaxTimestamp());
        List<String> keys = new ArrayList<String>();
        for (int off : offs)
            keys.add("key" + off);
        logger.debug("... everything looks good for " + cfs.columnFamily);
        return keys;
    }
private List < String > createAndTransfer ( Table table , ColumnFamilyStore cfs , Mutator mutator ) throws Exception { logger . debug ( "Mutating " + cfs . columnFamily ) ; long timestamp = 1234 ; for ( int i = 1 ; i <= 3 ; i ++ ) mutator . mutate ( "key" + i , "col" + i , timestamp ) ; cfs . forceBlockingFlush ( ) ; Util . compactAll ( cfs ) . get ( ) ; assertEquals ( 1 , cfs . getSSTables ( ) . size ( ) ) ; SSTableReader sstable = cfs . getSSTables ( ) . iterator ( ) . next ( ) ; cfs . clearUnsafe ( ) ; logger . debug ( "Transferring " + cfs . columnFamily ) ; transfer ( table , sstable ) ; assertEquals ( 1 , cfs . getSSTables ( ) . size ( ) ) ; int [ ] offs = new int [ ] { 1 , 3 } ; List < Row > rows = Util . getRangeSlice ( cfs ) ; assertEquals ( offs . length , rows . size ( ) ) ; for ( int i = 0 ; i < offs . length ; i ++ ) { String key = "key" + offs [ i ] ; String col = "col" + offs [ i ] ; assert null != cfs . getColumnFamily ( QueryFilter . getIdentityFilter ( Util . dk ( key ) , new QueryPath ( cfs . columnFamily ) ) ) ; assert rows . get ( i ) . key . key . equals ( ByteBufferUtil . bytes ( key ) ) ; assert rows . get ( i ) . cf . getColumn ( ByteBufferUtil . bytes ( col ) ) != null ; } assertEquals ( timestamp , cfs . getSSTables ( ) . iterator ( ) . next ( ) . getMaxTimestamp ( ) ) ; List < String > keys = new ArrayList < String > ( ) ; for ( int off : offs ) keys . add ( "key" + off ) ; logger . debug ( "... everything looks good for " + cfs . columnFamily ) ; return keys ; }
Create and transfer a single sstable, and return the keys that should have been transferred.
Create and transfer a single sstable and return the keys that should have been transferred .
// write a temporary SSTable, and unregister it // transfer the first and last key // confirm that a single SSTable was transferred and registered // and that the index and filter were properly recovered // and that the max timestamp for the file was rediscovered