Add paranoid disk failure option.

Patch by marcuse, reviewed by kohlisankalp and jbellis for CASSANDRA-6646

CHANGES.txt

@@ -14,6 +14,7 @@
  * Correctly compare BooleanType values other than 0 and 1 (CASSANDRA-6779)
  * Read message id as string from earlier versions (CASSANDRA-6840)
  * Properly use the Paxos consistency for (non-protocol) batch (CASSANDRA-6837)
+ * Add paranoid disk failure option (CASSANDRA-6646)
 Merged from 1.2:
  * add extra SSL cipher suites (CASSANDRA-6613)
  * fix nodetool getsstables for blob PK (CASSANDRA-6803)

conf/cassandra.yaml

@@ -111,6 +111,7 @@ data_file_directories:
 commitlog_directory: /var/lib/cassandra/commitlog
 
 # policy for data disk failures:
+# stop_paranoid: shut down gossip and Thrift even for single-sstable errors.
 # stop: shut down gossip and Thrift, leaving the node effectively dead, but
 #       can still be inspected via JMX.
 # best_effort: stop using the failed disk and respond to requests based on

src/java/org/apache/cassandra/config/Config.java

@@ -275,6 +275,7 @@ public class Config
         best_effort,
         stop,
         ignore,
+        stop_paranoid,
     }
 
     public static enum CommitFailurePolicy

src/java/org/apache/cassandra/io/util/FileUtils.java

@@ -38,6 +38,7 @@ import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.io.FSError;
 import org.apache.cassandra.io.FSReadError;
 import org.apache.cassandra.io.FSWriteError;
+import org.apache.cassandra.io.sstable.CorruptSSTableException;
 import org.apache.cassandra.service.StorageService;
 
 public class FileUtils
@@ -407,11 +408,18 @@ public class FileUtils
             n += skipped;
         }
     }
+
+    public static void handleCorruptSSTable(CorruptSSTableException e)
+    {
+        if (DatabaseDescriptor.getDiskFailurePolicy() == Config.DiskFailurePolicy.stop_paranoid)
+            StorageService.instance.stopTransports();
+    }
     
     public static void handleFSError(FSError e)
     {
         switch (DatabaseDescriptor.getDiskFailurePolicy())
         {
+            case stop_paranoid:
             case stop:
                 StorageService.instance.stopTransports();
                 break;

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

@@ -36,6 +36,8 @@ import com.addthis.metrics.reporter.config.ReporterConfig;
 
 import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.Uninterruptibles;
+
+import org.apache.cassandra.io.sstable.CorruptSSTableException;
 import org.apache.log4j.PropertyConfigurator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -207,6 +209,13 @@ public class CassandraDaemon
                             logger.error("Exception in thread " + t, e2);
                         FileUtils.handleFSError((FSError) e2);
                     }
+
+                    if (e2 instanceof CorruptSSTableException)
+                    {
+                        if (e2 != e)
+                            logger.error("Exception in thread " + t, e2);
+                        FileUtils.handleCorruptSSTable((CorruptSSTableException) e2);
+                    }
                 }
             }
         });