Merge branch 'cassandra-2.0' into trunk

Conflicts:
	CHANGES.txt
	src/java/org/apache/cassandra/tools/NodeCmd.java
	src/resources/org/apache/cassandra/tools/NodeToolHelp.yaml

CHANGES.txt

@@ -27,6 +27,10 @@
  * New counters implementation (CASSANDRA-6504)
 
 
+2.0.6
+ * Let scrub optionally skip broken counter partitions (CASSANDRA-5930)
+
+
 2.0.5
  * Reduce garbage generated by bloom filter lookups (CASSANDRA-6609)
  * Add ks.cf names to tombstone logging (CASSANDRA-6597)

NEWS.txt

@@ -44,11 +44,21 @@ Upgrading
      (See https://issues.apache.org/jira/browse/CASSANDRA-6504 and the dev
      blog post at http://www.datastax.com/dev/blog/<PLACEHOLDER> for details).
 
+2.0.6
+=====
+
+New features
+------------
+    - Scrub can now optionally skip corrupt counter partitions. Please note
+      that this will lead to the loss of all the counter updates in the skipped
+      partition. See the --skip-corrupted option.
+
+
 2.0.5
 =====
 
 New features
---------
+------------
     - Batchlog replay can be, and is throttled by default now.
       See batchlog_replay_throttle_in_kb setting in cassandra.yaml.
 

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -1303,12 +1303,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         CompactionManager.instance.performCleanup(ColumnFamilyStore.this);
     }
 
-    public void scrub(boolean disableSnapshot) throws ExecutionException, InterruptedException
+    public void scrub(boolean disableSnapshot, boolean skipCorrupted) throws ExecutionException, InterruptedException
     {
         // skip snapshot creation during scrub, SEE JIRA 5891
         if(!disableSnapshot)
             snapshotWithoutFlush("pre-scrub-" + System.currentTimeMillis());
-        CompactionManager.instance.performScrub(ColumnFamilyStore.this);
+        CompactionManager.instance.performScrub(ColumnFamilyStore.this, skipCorrupted);
     }
 
     public void sstablesRewrite(boolean excludeCurrentVersion) throws ExecutionException, InterruptedException

src/java/org/apache/cassandra/db/compaction/CompactionManager.java

@@ -231,13 +231,13 @@ public class CompactionManager implements CompactionManagerMBean
         executor.submit(runnable).get();
     }
 
-    public void performScrub(ColumnFamilyStore cfStore) throws InterruptedException, ExecutionException
+    public void performScrub(ColumnFamilyStore cfStore, final boolean skipCorrupted) throws InterruptedException, ExecutionException
     {
         performAllSSTableOperation(cfStore, new AllSSTablesOperation()
         {
             public void perform(ColumnFamilyStore store, Iterable<SSTableReader> sstables) throws IOException
             {
-                doScrub(store, sstables);
+                doScrub(store, sstables, skipCorrupted);
             }
         });
     }
@@ -420,16 +420,16 @@ public class CompactionManager implements CompactionManagerMBean
      *
      * @throws IOException
      */
-    private void doScrub(ColumnFamilyStore cfs, Iterable<SSTableReader> sstables) throws IOException
+    private void doScrub(ColumnFamilyStore cfs, Iterable<SSTableReader> sstables, boolean skipCorrupted) throws IOException
     {
         assert !cfs.isIndex();
         for (final SSTableReader sstable : sstables)
-            scrubOne(cfs, sstable);
+            scrubOne(cfs, sstable, skipCorrupted);
     }
 
-    private void scrubOne(ColumnFamilyStore cfs, SSTableReader sstable) throws IOException
+    private void scrubOne(ColumnFamilyStore cfs, SSTableReader sstable, boolean skipCorrupted) throws IOException
     {
-        Scrubber scrubber = new Scrubber(cfs, sstable);
+        Scrubber scrubber = new Scrubber(cfs, sstable, skipCorrupted);
 
         CompactionInfo.Holder scrubInfo = scrubber.getScrubInfo();
         metrics.beginCompaction(scrubInfo);

src/java/org/apache/cassandra/db/compaction/Scrubber.java

@@ -35,6 +35,7 @@ public class Scrubber implements Closeable
     public final ColumnFamilyStore cfs;
     public final SSTableReader sstable;
     public final File destination;
+    public final boolean skipCorrupted;
 
     private final CompactionController controller;
     private final boolean isCommutative;
@@ -63,16 +64,17 @@ public class Scrubber implements Closeable
     };
     private final SortedSet<Row> outOfOrderRows = new TreeSet<>(rowComparator);
 
-    public Scrubber(ColumnFamilyStore cfs, SSTableReader sstable) throws IOException
+    public Scrubber(ColumnFamilyStore cfs, SSTableReader sstable, boolean skipCorrupted) throws IOException
     {
-        this(cfs, sstable, new OutputHandler.LogOutput(), false);
+        this(cfs, sstable, skipCorrupted, new OutputHandler.LogOutput(), false);
     }
 
-    public Scrubber(ColumnFamilyStore cfs, SSTableReader sstable, OutputHandler outputHandler, boolean isOffline) throws IOException
+    public Scrubber(ColumnFamilyStore cfs, SSTableReader sstable, boolean skipCorrupted, OutputHandler outputHandler, boolean isOffline) throws IOException
     {
         this.cfs = cfs;
         this.sstable = sstable;
         this.outputHandler = outputHandler;
+        this.skipCorrupted = skipCorrupted;
 
         // Calculate the expected compacted filesize
         this.destination = cfs.directories.getDirectoryForNewSSTables();
@@ -166,7 +168,9 @@ public class Scrubber implements Closeable
                 if (!sstable.descriptor.version.hasRowSizeAndColumnCount)
                 {
                     dataSize = dataSizeFromIndex;
-                    outputHandler.debug(String.format("row %s is %s bytes", ByteBufferUtil.bytesToHex(key.key), dataSize));
+                    // avoid an NPE if key is null
+                    String keyName = key == null ? "(unreadable key)" : ByteBufferUtil.bytesToHex(key.key);
+                    outputHandler.debug(String.format("row %s is %s bytes", keyName, dataSize));
                 }
                 else
                 {
@@ -203,7 +207,7 @@ public class Scrubber implements Closeable
                 catch (Throwable th)
                 {
                     throwIfFatal(th);
-                    outputHandler.warn("Non-fatal error reading row (stacktrace follows)", th);
+                    outputHandler.warn("Error reading row (stacktrace follows):", th);
                     writer.resetAndTruncate();
 
                     if (currentIndexKey != null
@@ -231,9 +235,7 @@ public class Scrubber implements Closeable
                         catch (Throwable th2)
                         {
                             throwIfFatal(th2);
-                            // Skipping rows is dangerous for counters (see CASSANDRA-2759)
+                            throwIfCommutative(key, th2);
-                            if (isCommutative)
-                                throw new IOError(th2);
 
                             outputHandler.warn("Retry failed too. Skipping to next row (retry's stacktrace follows)", th2);
                             writer.resetAndTruncate();
@@ -243,11 +245,9 @@ public class Scrubber implements Closeable
                     }
                     else
                     {
-                        // Skipping rows is dangerous for counters (see CASSANDRA-2759)
+                        throwIfCommutative(key, th);
-                        if (isCommutative)
-                            throw new IOError(th);
 
-                        outputHandler.warn("Row at " + dataStart + " is unreadable; skipping to next");
+                        outputHandler.warn("Row starting at position " + dataStart + " is unreadable; skipping to next");
                         if (currentIndexKey != null)
                             dataFile.seek(nextRowPositionFromIndex);
                         badRows++;
@@ -324,6 +324,19 @@ public class Scrubber implements Closeable
             throw (Error) th;
     }
 
+    private void throwIfCommutative(DecoratedKey key, Throwable th)
+    {
+        if (isCommutative && !skipCorrupted)
+        {
+            outputHandler.warn(String.format("An error occurred while scrubbing the row with key '%s'.  Skipping corrupt " +
+                                             "rows in counter tables will result in undercounts for the affected " +
+                                             "counters (see CASSANDRA-2759 for more details), so by default the scrub will " +
+                                             "stop at this point.  If you would like to skip the row anyway and continue " +
+                                             "scrubbing, re-run the scrub with the --skip-corrupted option.", key));
+            throw new IOError(th);
+        }
+    }
+
     public void close()
     {
         FileUtils.closeQuietly(dataFile);

src/java/org/apache/cassandra/service/StorageService.java

@@ -2143,10 +2143,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         }
     }
 
-    public void scrub(boolean disableSnapshot, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException
+    public void scrub(boolean disableSnapshot, boolean skipCorrupted, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException
     {
         for (ColumnFamilyStore cfStore : getValidColumnFamilies(false, false, keyspaceName, columnFamilies))
-            cfStore.scrub(disableSnapshot);
+            cfStore.scrub(disableSnapshot, skipCorrupted);
     }
 
     public void upgradeSSTables(String keyspaceName, boolean excludeCurrentVersion, String... columnFamilies) throws IOException, ExecutionException, InterruptedException

src/java/org/apache/cassandra/service/StorageServiceMBean.java

@@ -244,7 +244,7 @@ public interface StorageServiceMBean extends NotificationEmitter
      *
      * Scrubbed CFs will be snapshotted first, if disableSnapshot is false
      */
-    public void scrub(boolean disableSnapshot, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException;
+    public void scrub(boolean disableSnapshot, boolean skipCorrupted, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException;
 
     /**
      * Rewrite all sstables to the latest version.

src/java/org/apache/cassandra/tools/NodeProbe.java

@@ -191,9 +191,9 @@ public class NodeProbe implements AutoCloseable
         ssProxy.forceKeyspaceCleanup(keyspaceName, columnFamilies);
     }
 
-    public void scrub(boolean disableSnapshot, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException
+    public void scrub(boolean disableSnapshot, boolean skipCorrupted, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException
     {
-        ssProxy.scrub(disableSnapshot, keyspaceName, columnFamilies);
+        ssProxy.scrub(disableSnapshot, skipCorrupted, keyspaceName, columnFamilies);
     }
 
     public void upgradeSSTables(String keyspaceName, boolean excludeCurrentVersion, String... columnFamilies) throws IOException, ExecutionException, InterruptedException

src/java/org/apache/cassandra/tools/NodeTool.java

@@ -971,9 +971,16 @@ public class NodeTool
         @Arguments(usage = "[<keyspace> <cfnames>...]", description = "The keyspace followed by one or many column families")
         private List<String> args = new ArrayList<>();
 
-        @Option(title = "disable_snapshot", name = {"-ns", "--no-snapshot"}, description = "Scrubbed CFs will be snapshotted first, if disableSnapshot is false. (default false)")
+        @Option(title = "disable_snapshot",
+                name = {"-ns", "--no-snapshot"},
+                description = "Scrubbed CFs will be snapshotted first, if disableSnapshot is false. (default false)")
         private boolean disableSnapshot = false;
 
+        @Option(title = "skip_corrupted",
+                name = {"-s", "--skip-corrupted"},
+                description = "Skip corrupted partitions even when scrubbing counter tables. (default false)")
+        private boolean skipCorrupted = false;
+
         @Override
         public void execute(NodeProbe probe)
         {
@@ -984,7 +991,7 @@ public class NodeTool
             {
                 try
                 {
-                    probe.scrub(disableSnapshot, keyspace, cfnames);
+                    probe.scrub(disableSnapshot, skipCorrupted, keyspace, cfnames);
                 } catch (Exception e)
                 {
                     throw new RuntimeException("Error occurred during flushing", e);

src/java/org/apache/cassandra/tools/StandaloneScrubber.java

@@ -43,6 +43,7 @@ public class StandaloneScrubber
     private static final String DEBUG_OPTION  = "debug";
     private static final String HELP_OPTION  = "help";
     private static final String MANIFEST_CHECK_OPTION  = "manifest-check";
+    private static final String SKIP_CORRUPTED_OPTION = "skip-corrupted";
 
     public static void main(String args[])
     {
@@ -106,7 +107,7 @@ public class StandaloneScrubber
                 {
                     try
                     {
-                        Scrubber scrubber = new Scrubber(cfs, sstable, handler, true);
+                        Scrubber scrubber = new Scrubber(cfs, sstable, options.skipCorrupted, handler, true);
                         try
                         {
                             scrubber.scrub();
@@ -171,6 +172,7 @@ public class StandaloneScrubber
         public boolean debug;
         public boolean verbose;
         public boolean manifestCheckOnly;
+        public boolean skipCorrupted;
 
         private Options(String keyspaceName, String cfName)
         {
@@ -209,6 +211,7 @@ public class StandaloneScrubber
                 opts.debug = cmd.hasOption(DEBUG_OPTION);
                 opts.verbose = cmd.hasOption(VERBOSE_OPTION);
                 opts.manifestCheckOnly = cmd.hasOption(MANIFEST_CHECK_OPTION);
+                opts.skipCorrupted = cmd.hasOption(SKIP_CORRUPTED_OPTION);
 
                 return opts;
             }
@@ -233,6 +236,7 @@ public class StandaloneScrubber
             options.addOption("v",  VERBOSE_OPTION,        "verbose output");
             options.addOption("h",  HELP_OPTION,           "display this help message");
             options.addOption("m",  MANIFEST_CHECK_OPTION, "only check and repair the leveled manifest, without actually scrubbing the sstables");
+            options.addOption("s",  SKIP_CORRUPTED_OPTION, "skip corrupt rows in counter tables");
             return options;
         }
 

test/unit/org/apache/cassandra/db/ScrubTest.java

@@ -20,13 +20,15 @@ package org.apache.cassandra.db;
  *
  */
 
-import java.io.File;
+import java.io.*;
-import java.io.IOException;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 
+import org.apache.cassandra.db.compaction.OperationType;
+import org.apache.commons.lang3.StringUtils;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -39,9 +41,11 @@ import org.apache.cassandra.db.compaction.Scrubber;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.db.columniterator.IdentityQueryFilter;
 import org.apache.cassandra.db.compaction.CompactionManager;
+import org.apache.cassandra.exceptions.WriteTimeoutException;
 import org.apache.cassandra.io.sstable.*;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
+import static org.apache.cassandra.Util.cellname;
 import static org.apache.cassandra.Util.column;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
@@ -52,6 +56,7 @@ public class ScrubTest extends SchemaLoader
     public String KEYSPACE = "Keyspace1";
     public String CF = "Standard1";
     public String CF3 = "Standard2";
+    public String COUNTER_CF = "Counter1";
 
     @Test
     public void testScrubOneRow() throws IOException, ExecutionException, InterruptedException, ConfigurationException
@@ -68,13 +73,60 @@ public class ScrubTest extends SchemaLoader
         rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
         assertEquals(1, rows.size());
 
-        CompactionManager.instance.performScrub(cfs);
+        CompactionManager.instance.performScrub(cfs, false);
 
         // check data is still there
         rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
         assertEquals(1, rows.size());
     }
 
+    @Test
+    public void testScrubCorruptedCounterRow() throws IOException, InterruptedException, ExecutionException, WriteTimeoutException
+    {
+        CompactionManager.instance.disableAutoCompaction();
+        Keyspace keyspace = Keyspace.open(KEYSPACE);
+        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(COUNTER_CF);
+        cfs.clearUnsafe();
+
+        fillCounterCF(cfs, 2);
+
+        List<Row> rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
+        assertEquals(2, rows.size());
+
+        SSTableReader sstable = cfs.getSSTables().iterator().next();
+
+        // overwrite one row with garbage
+        long row0Start = sstable.getPosition(RowPosition.forKey(ByteBufferUtil.bytes("0"), sstable.partitioner), SSTableReader.Operator.EQ).position;
+        long row1Start = sstable.getPosition(RowPosition.forKey(ByteBufferUtil.bytes("1"), sstable.partitioner), SSTableReader.Operator.EQ).position;
+        long startPosition = row0Start < row1Start ? row0Start : row1Start;
+        long endPosition = row0Start < row1Start ? row1Start : row0Start;
+
+        RandomAccessFile file = new RandomAccessFile(sstable.getFilename(), "rw");
+        file.seek(startPosition);
+        file.writeBytes(StringUtils.repeat('z', (int) (endPosition - startPosition)));
+        file.close();
+
+        // with skipCorrupted == false, the scrub is expected to fail
+        Scrubber scrubber = new Scrubber(cfs, sstable, false);
+        try
+        {
+            scrubber.scrub();
+            fail("Expected a CorruptSSTableException to be thrown");
+        }
+        catch (IOError err) {}
+
+        // with skipCorrupted == true, the corrupt row will be skipped
+        scrubber = new Scrubber(cfs, sstable, true);
+        scrubber.scrub();
+        scrubber.close();
+        cfs.replaceCompactedSSTables(Collections.singletonList(sstable), Collections.singletonList(scrubber.getNewSSTable()), OperationType.SCRUB);
+        assertEquals(1, cfs.getSSTables().size());
+
+        // verify that we can read all of the rows, and there is now one less row
+        rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
+        assertEquals(1, rows.size());
+    }
+
     @Test
     public void testScrubDeletedRow() throws IOException, ExecutionException, InterruptedException, ConfigurationException
     {
@@ -89,7 +141,7 @@ public class ScrubTest extends SchemaLoader
         rm.applyUnsafe();
         cfs.forceBlockingFlush();
 
-        CompactionManager.instance.performScrub(cfs);
+        CompactionManager.instance.performScrub(cfs, false);
         assert cfs.getSSTables().isEmpty();
     }
 
@@ -108,7 +160,7 @@ public class ScrubTest extends SchemaLoader
         rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
         assertEquals(10, rows.size());
 
-        CompactionManager.instance.performScrub(cfs);
+        CompactionManager.instance.performScrub(cfs, false);
 
         // check data is still there
         rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
@@ -145,7 +197,6 @@ public class ScrubTest extends SchemaLoader
         writer.closeAndOpenReader();
         */
 
-
         String root = System.getProperty("corrupt-sstable-root");
         assert root != null;
         File rootDir = new File(root);
@@ -171,7 +222,7 @@ public class ScrubTest extends SchemaLoader
         components.add(Component.TOC);
         SSTableReader sstable = SSTableReader.openNoValidation(desc, components, metadata);
 
-        Scrubber scrubber = new Scrubber(cfs, sstable);
+        Scrubber scrubber = new Scrubber(cfs, sstable, false);
         scrubber.scrub();
 
         cfs.loadNewSSTables();
@@ -207,4 +258,20 @@ public class ScrubTest extends SchemaLoader
 
         cfs.forceBlockingFlush();
     }
+
+    protected void fillCounterCF(ColumnFamilyStore cfs, int rowsPerSSTable) throws ExecutionException, InterruptedException, IOException, WriteTimeoutException
+    {
+        for (int i = 0; i < rowsPerSSTable; i++)
+        {
+            String key = String.valueOf(i);
+            ColumnFamily cf = TreeMapBackedSortedColumns.factory.create(KEYSPACE, COUNTER_CF);
+            Mutation rm = new Mutation(KEYSPACE, ByteBufferUtil.bytes(key), cf);
+            rm.addCounter(COUNTER_CF, cellname("Column1"), 100);
+            CounterMutation cm = new CounterMutation(rm, ConsistencyLevel.ONE);
+            cm.apply();
+        }
+
+        cfs.forceBlockingFlush();
+    }
+
 }