diff --git a/CHANGES.txt b/CHANGES.txt
index 040af7c87c..86d4e6f2e6 100644
--- a/CHANGES.txt
+++ b/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)
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index dc71c0f039..d3c91efd21 100644
--- a/conf/cassandra.yaml
+++ b/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
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index e8e43e2032..5317fb8110 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/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
diff --git a/src/java/org/apache/cassandra/io/util/FileUtils.java b/src/java/org/apache/cassandra/io/util/FileUtils.java
index e091465678..a12745cd19 100644
--- a/src/java/org/apache/cassandra/io/util/FileUtils.java
+++ b/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;
diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java
index d8c56f1096..0970776b0a 100644
--- a/src/java/org/apache/cassandra/service/CassandraDaemon.java
+++ b/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);
+                    }
                 }
             }
         });