diff --git a/CHANGES.txt b/CHANGES.txt
index 71ce442cd4..4a7e174c1a 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -32,6 +32,7 @@
  * Fix streaming not holding ref when stream error (CASSANDRA-9295)
  * Fix canonical view returning early opened SSTables (CASSANDRA-9396)
 Merged from 2.0:
+ * Always mark sstable suspect when corrupted (CASSANDRA-9478)
  * Add database users and permissions to CQL3 documentation (CASSANDRA-7558)
  * Allow JVM_OPTS to be passed to standalone tools (CASSANDRA-5969)
  * Fix bad condition in RangeTombstoneList (CASSANDRA-9485)
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
index b784a7e691..498ad26527 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
@@ -38,6 +38,9 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
     private final boolean validateColumns;
     private final String filename;
 
+    // Not every SSTableIdentifyIterator is attached to a sstable, so this can be null.
+    private final SSTableReader sstable;
+
     /**
      * Used to iterate through the columns of a row.
      * @param sstable SSTable we are reading ffrom.
@@ -81,6 +84,7 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
         this.dataSize = dataSize;
         this.flag = flag;
         this.validateColumns = checkData;
+        this.sstable = sstable;
 
         Descriptor.Version dataVersion = sstable == null ? Descriptor.Version.CURRENT : sstable.descriptor.version;
         int expireBefore = (int) (System.currentTimeMillis() / 1000);
@@ -119,9 +123,15 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
         {
             // catch here b/c atomIterator is an AbstractIterator; hasNext reads the value
             if (e.getCause() instanceof IOException)
+            {
+                if (sstable != null)
+                    sstable.markSuspect();
                 throw new CorruptSSTableException((IOException)e.getCause(), filename);
+            }
             else
+            {
                 throw e;
+            }
         }
     }
 
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
index 43873a050f..04756655b5 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
@@ -956,14 +956,55 @@ public class SSTableReader extends SSTable implements SelfRefCounted<SSTableRead
         }
     }
 
+    /**
+     * Clone this reader with the provided start and open reason, and set the clone as replacement.
+     *
+     * @param newFirst the first key for the replacement (which can be different from the original due to the pre-emptive
+     * opening of compaction results).
+     * @param reason the {@code OpenReason} for the replacement.
+     *
+     * @return the cloned reader. That reader is set as a replacement by the method.
+     */
+    private SSTableReader cloneAndReplace(DecoratedKey newFirst, OpenReason reason)
+    {
+        return cloneAndReplace(newFirst, reason, indexSummary.sharedCopy());
+    }
+
+    /**
+     * Clone this reader with the new values and set the clone as replacement.
+     *
+     * @param newFirst the first key for the replacement (which can be different from the original due to the pre-emptive
+     * opening of compaction results).
+     * @param reason the {@code OpenReason} for the replacement.
+     * @param newSummary the index summary for the replacement.
+     *
+     * @return the cloned reader. That reader is set as a replacement by the method.
+     */
+    private SSTableReader cloneAndReplace(DecoratedKey newFirst, OpenReason reason, IndexSummary newSummary)
+    {
+        SSTableReader replacement = internalOpen(descriptor,
+                                                 components,
+                                                 metadata,
+                                                 partitioner,
+                                                 ifile.sharedCopy(),
+                                                 dfile.sharedCopy(),
+                                                 newSummary,
+                                                 bf.sharedCopy(),
+                                                 maxDataAge,
+                                                 sstableMetadata,
+                                                 reason);
+        replacement.first = newFirst;
+        replacement.last = last;
+        replacement.isSuspect.set(isSuspect.get());
+        setReplacedBy(replacement);
+        return replacement;
+    }
+
     public SSTableReader cloneWithNewStart(DecoratedKey newStart, final Runnable runOnClose)
     {
         synchronized (tidy.global)
         {
             assert openReason != OpenReason.EARLY;
-            SSTableReader replacement = new SSTableReader(descriptor, components, metadata, partitioner, ifile.sharedCopy(),
-                                                          dfile.sharedCopy(), indexSummary.sharedCopy(), bf.sharedCopy(),
-                                                          maxDataAge, sstableMetadata, OpenReason.MOVED_START);
             // TODO: make data/index start accurate for compressed files
             // TODO: merge with caller's firstKeyBeyond() work,to save time
             if (newStart.compareTo(first) > 0)
@@ -982,10 +1023,7 @@ public class SSTableReader extends SSTable implements SelfRefCounted<SSTableRead
                 };
             }
 
-            replacement.first = newStart;
-            replacement.last = this.last;
-            setReplacedBy(replacement);
-            return replacement;
+            return cloneAndReplace(newStart, OpenReason.MOVED_START);
         }
     }
 
@@ -1004,13 +1042,7 @@ public class SSTableReader extends SSTable implements SelfRefCounted<SSTableRead
                 }
             };
 
-            SSTableReader replacement = new SSTableReader(descriptor, components, metadata, partitioner, ifile.sharedCopy(),
-                                                          dfile.sharedCopy(), indexSummary.sharedCopy(), bf.sharedCopy(),
-                                                          maxDataAge, sstableMetadata, OpenReason.SHADOWED);
-            replacement.first = first;
-            replacement.last = last;
-            setReplacedBy(replacement);
-            return replacement;
+            return cloneAndReplace(first, OpenReason.SHADOWED);
         }
     }
 
@@ -1067,13 +1099,7 @@ public class SSTableReader extends SSTable implements SelfRefCounted<SSTableRead
             StorageMetrics.load.inc(newSize - oldSize);
             parent.metric.liveDiskSpaceUsed.inc(newSize - oldSize);
 
-            SSTableReader replacement = new SSTableReader(descriptor, components, metadata, partitioner, ifile.sharedCopy(),
-                                                          dfile.sharedCopy(), newSummary, bf.sharedCopy(), maxDataAge,
-                                                          sstableMetadata, OpenReason.METADATA_CHANGE);
-            replacement.first = this.first;
-            replacement.last = this.last;
-            setReplacedBy(replacement);
-            return replacement;
+            return cloneAndReplace(first, OpenReason.METADATA_CHANGE, newSummary);
         }
     }
 
diff --git a/test/unit/org/apache/cassandra/db/compaction/BlacklistingCompactionsTest.java b/test/unit/org/apache/cassandra/db/compaction/BlacklistingCompactionsTest.java
index 08e3fb32bb..572ad36a16 100644
--- a/test/unit/org/apache/cassandra/db/compaction/BlacklistingCompactionsTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/BlacklistingCompactionsTest.java
@@ -22,9 +22,7 @@ package org.apache.cassandra.db.compaction;
 
 
 import java.io.RandomAccessFile;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.Set;
+import java.util.*;
 
 import org.junit.After;
 import org.junit.AfterClass;
@@ -40,6 +38,7 @@ import org.apache.cassandra.utils.ByteBufferUtil;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 import static org.apache.cassandra.Util.cellname;
 
 public class BlacklistingCompactionsTest extends SchemaLoader
@@ -128,7 +127,14 @@ public class BlacklistingCompactionsTest extends SchemaLoader
             {
                 raf = new RandomAccessFile(sstable.getFilename(), "rw");
                 assertNotNull(raf);
-                raf.write(0xFFFFFF);
+                assertTrue(raf.length() > 20);
+                raf.seek(new Random().nextInt((int)(raf.length() - 20)));
+                // We want to write something large enough that the corruption cannot get undetected
+                // (even without compression)
+                byte[] corruption = new byte[20];
+                Arrays.fill(corruption, (byte)0xFF);
+                raf.write(corruption);
+
             }
             finally
             {
@@ -160,8 +166,7 @@ public class BlacklistingCompactionsTest extends SchemaLoader
             break;
         }
 
-
         cfs.truncateBlocking();
-        assertEquals(failures, sstablesToCorrupt);
+        assertEquals(sstablesToCorrupt, failures);
     }
 }