Merge branch 'cassandra-2.0' into cassandra-2.1

Conflicts:
	CHANGES.txt
	src/java/org/apache/cassandra/service/CassandraDaemon.java

CHANGES.txt

@@ -39,6 +39,7 @@ Merged from 2.0:
  * 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)
  * add extra SSL cipher suites (CASSANDRA-6613)
  * fix nodetool getsstables for blob PK (CASSANDRA-6803)
 

conf/cassandra.yaml

@@ -117,6 +117,7 @@ commitlog_directory: /var/lib/cassandra/commitlog
 flush_directory: /var/lib/cassandra/flush
 
 # 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

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

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

@@ -31,12 +31,14 @@ import java.util.Arrays;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.config.Config;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.BlacklistedDirectories;
 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
@@ -406,29 +408,20 @@ 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:
-                if (StorageService.instance.isInitialized())
-                {
-                    logger.error("Stopping gossiper");
-                    StorageService.instance.stopGossiping();
-                }
-
-                if (StorageService.instance.isRPCServerRunning())
-                {
-                    logger.error("Stopping RPC server");
-                    StorageService.instance.stopRPCServer();
-                }
-
-                if (StorageService.instance.isNativeTransportRunning())
-                {
-                    logger.error("Stopping native transport");
-                    StorageService.instance.stopNativeTransport();
-                }
+                StorageService.instance.stopTransports();
                 break;
             case best_effort:
                 // for both read and write errors mark the path as unwritable.

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

@@ -34,6 +34,8 @@ import javax.management.StandardMBean;
 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.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -176,6 +178,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);
+                    }
                 }
             }
         });

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

@@ -354,6 +354,25 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         return daemon.nativeServer.isRunning();
     }
 
+    public void stopTransports()
+    {
+        if (isInitialized())
+        {
+            logger.error("Stopping gossiper");
+            stopGossiping();
+        }
+        if (isRPCServerRunning())
+        {
+            logger.error("Stopping RPC server");
+            stopRPCServer();
+        }
+        if (isNativeTransportRunning())
+        {
+            logger.error("Stopping native transport");
+            stopNativeTransport();
+        }
+    }
+
     private void shutdownClientServers()
     {
         stopRPCServer();