cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From xe...@apache.org
Subject [2/2] git commit: identify and blacklist corrupted SSTables from future compactions patch by Benjamin Coverston and Pavel Yaskevich; reviewed by Brandon Williams for CASSANDRA-2261
Date Tue, 28 Feb 2012 20:47:09 GMT
identify and blacklist corrupted SSTables from future compactions
patch by Benjamin Coverston and Pavel Yaskevich; reviewed by Brandon Williams for CASSANDRA-2261


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/9483e147
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/9483e147
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/9483e147

Branch: refs/heads/trunk
Commit: 9483e147ec675c6024b5f1a55c799cc15aa2239e
Parents: 2cc612c
Author: Pavel Yaskevich <xedin@apache.org>
Authored: Tue Feb 28 22:45:24 2012 +0300
Committer: Pavel Yaskevich <xedin@apache.org>
Committed: Tue Feb 28 23:37:45 2012 +0300

----------------------------------------------------------------------
 CHANGES.txt                                        |    2 +-
 .../db/columniterator/IndexedSliceReader.java      |    1 +
 .../db/columniterator/SSTableNamesIterator.java    |    2 +
 .../db/columniterator/SSTableSliceIterator.java    |    1 +
 .../db/columniterator/SimpleSliceReader.java       |    1 +
 .../db/compaction/AbstractCompactionStrategy.java  |   24 +++-
 .../cassandra/db/compaction/LeveledManifest.java   |   28 ++++-
 .../compaction/SizeTieredCompactionStrategy.java   |    5 +-
 .../io/sstable/SSTableBoundedScanner.java          |    1 +
 .../io/sstable/SSTableIdentityIterator.java        |    1 +
 .../apache/cassandra/io/sstable/SSTableReader.java |   15 ++
 .../cassandra/io/sstable/SSTableScanner.java       |    5 +
 .../cassandra/db/compaction/CompactionsTest.java   |  113 ++++++++++++++-
 13 files changed, 191 insertions(+), 8 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/9483e147/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 6a2a56e..591ae9f 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,6 +1,6 @@
 1.1.1-dev
  * optimize commitlog checksumming (CASSANDRA-3610)
-
+ * identify and blacklist corrupted SSTables from future compactions (CASSANDRA-2261)
 
 1.1-dev
  * start hint replay as soon as FD notifies that the target is back up

http://git-wip-us.apache.org/repos/asf/cassandra/blob/9483e147/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java b/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java
index d2d9e9b..68c9a4c 100644
--- a/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java
+++ b/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java
@@ -76,6 +76,7 @@ class IndexedSliceReader extends AbstractIterator<IColumn> implements
IColumnIte
         }
         catch (IOException e)
         {
+            sstable.markSuspect();
             throw new IOError(e);
         }
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/9483e147/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java b/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java
index fdb0c5c..ac8938c 100644
--- a/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java
+++ b/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java
@@ -73,6 +73,7 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement
         }
         catch (IOException e)
         {
+            sstable.markSuspect();
             throw new IOError(e);
         }
         finally
@@ -93,6 +94,7 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement
         }
         catch (IOException ioe)
         {
+            sstable.markSuspect();
             throw new IOError(ioe);
         }
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/9483e147/src/java/org/apache/cassandra/db/columniterator/SSTableSliceIterator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/columniterator/SSTableSliceIterator.java b/src/java/org/apache/cassandra/db/columniterator/SSTableSliceIterator.java
index 72fe1bf..5c97729 100644
--- a/src/java/org/apache/cassandra/db/columniterator/SSTableSliceIterator.java
+++ b/src/java/org/apache/cassandra/db/columniterator/SSTableSliceIterator.java
@@ -60,6 +60,7 @@ public class SSTableSliceIterator implements IColumnIterator
         }
         catch (IOException e)
         {
+            sstable.markSuspect();
             throw new IOError(e);
         }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/9483e147/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java b/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java
index 383777f..4e49ebb 100644
--- a/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java
+++ b/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java
@@ -62,6 +62,7 @@ class SimpleSliceReader extends AbstractIterator<IColumn> implements
IColumnIter
         }
         catch (IOException e)
         {
+            sstable.markSuspect();
             throw new IOError(e);
         }
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/9483e147/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
index 0455f95..2e70c46 100644
--- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
@@ -18,9 +18,7 @@
 
 package org.apache.cassandra.db.compaction;
 
-import java.util.Collection;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.cassandra.db.ColumnFamilyStore;
@@ -108,4 +106,24 @@ public abstract class AbstractCompactionStrategy
      * is going to be expensive
      */
     public abstract boolean isKeyExistenceExpensive(Set<? extends SSTable> sstablesToIgnore);
+
+    /**
+     * Filters SSTables that are to be blacklisted from the given collection
+     *
+     * @param originalCandidates The collection to check for blacklisted SSTables
+     *
+     * @return list of the SSTables with blacklisted ones filtered out
+     */
+    public static List<SSTableReader> filterSuspectSSTables(Collection<SSTableReader>
originalCandidates)
+    {
+        List<SSTableReader> filteredCandidates = new ArrayList<SSTableReader>();
+
+        for (SSTableReader candidate : originalCandidates)
+        {
+            if (!candidate.isMarkedSuspect())
+                filteredCandidates.add(candidate);
+        }
+
+        return filteredCandidates;
+    }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/9483e147/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java b/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
index 4c63180..2661c6d 100644
--- a/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
+++ b/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
@@ -44,6 +44,8 @@ import org.codehaus.jackson.JsonGenerator;
 import org.codehaus.jackson.JsonNode;
 import org.codehaus.jackson.map.ObjectMapper;
 
+import static org.apache.cassandra.db.compaction.AbstractCompactionStrategy.filterSuspectSSTables;
+
 public class LeveledManifest
 {
     private static final Logger logger = LoggerFactory.getLogger(LeveledManifest.class);
@@ -249,15 +251,39 @@ public class LeveledManifest
             if (score > 1.001 || i == 0)
             {
                 Collection<SSTableReader> candidates = getCandidatesFor(i);
+
                 if (logger.isDebugEnabled())
                     logger.debug("Compaction candidates for L{} are {}", i, toString(candidates));
-                return candidates;
+
+                // check if have any SSTables marked as suspected,
+                // saves us filter time when no SSTables are suspects
+                return hasSuspectSSTables(candidates)
+                        ? filterSuspectSSTables(candidates)
+                        : candidates;
             }
         }
 
         return Collections.emptyList();
     }
 
+    /**
+     * Go through candidates collection and check if any of the SSTables are marked as suspected.
+     *
+     * @param candidates The SSTable collection to examine.
+     *
+     * @return true if collection has at least one SSTable marked as suspected, false otherwise.
+     */
+    private boolean hasSuspectSSTables(Collection<SSTableReader> candidates)
+    {
+        for (SSTableReader candidate : candidates)
+        {
+            if (candidate.isMarkedSuspect())
+                return true;
+        }
+
+        return false;
+    }
+
     public int getLevelSize(int i)
     {
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/9483e147/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java
b/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java
index caead37..0281f08 100644
--- a/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/SizeTieredCompactionStrategy.java
@@ -55,7 +55,8 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy
             return null;
         }
 
-        List<List<SSTableReader>> buckets = getBuckets(createSSTableAndLengthPairs(cfs.getUncompactingSSTables()),
minSSTableSize);
+        Set<SSTableReader> candidates = cfs.getUncompactingSSTables();
+        List<List<SSTableReader>> buckets = getBuckets(createSSTableAndLengthPairs(filterSuspectSSTables(candidates)),
minSSTableSize);
         updateEstimatedCompactionsByTasks(buckets);
 
         List<List<SSTableReader>> prunedBuckets = new ArrayList<List<SSTableReader>>();
@@ -102,7 +103,7 @@ public class SizeTieredCompactionStrategy extends AbstractCompactionStrategy
 
     public AbstractCompactionTask getMaximalTask(final int gcBefore)
     {
-        return cfs.getSSTables().isEmpty() ? null : new CompactionTask(cfs, cfs.getSSTables(),
gcBefore);
+        return cfs.getSSTables().isEmpty() ? null : new CompactionTask(cfs, filterSuspectSSTables(cfs.getSSTables()),
gcBefore);
     }
 
     public AbstractCompactionTask getUserDefinedTask(Collection<SSTableReader> sstables,
final int gcBefore)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/9483e147/src/java/org/apache/cassandra/io/sstable/SSTableBoundedScanner.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableBoundedScanner.java b/src/java/org/apache/cassandra/io/sstable/SSTableBoundedScanner.java
index 262394d..280d7fc 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableBoundedScanner.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableBoundedScanner.java
@@ -49,6 +49,7 @@ public class SSTableBoundedScanner extends SSTableScanner
             }
             catch (IOException e)
             {
+                sstable.markSuspect();
                 throw new RuntimeException(e);
             }
         }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/9483e147/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
index fbb0788..275f983 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
@@ -151,6 +151,7 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
         }
         catch (IOException e)
         {
+            sstable.markSuspect();
             throw new IOError(e);
         }
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/9483e147/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
index 79b71ae..8c1d540 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
@@ -92,6 +92,7 @@ public class SSTableReader extends SSTable
     // technically isCompacted is not necessary since it should never be unreferenced unless
it is also compacted,
     // but it seems like a good extra layer of protection against reference counting bugs
to not delete data based on that alone
     private final AtomicBoolean isCompacted = new AtomicBoolean(false);
+    private final AtomicBoolean isSuspect = new AtomicBoolean(false);
     private final SSTableDeletingTask deletingTask;
 
     private final SSTableMetadata sstableMetadata;
@@ -713,6 +714,7 @@ public class SSTableReader extends SSTable
             }
             catch (IOException e)
             {
+                markSuspect();
                 throw new IOError(e);
             }
             finally
@@ -799,6 +801,19 @@ public class SSTableReader extends SSTable
         return true;
     }
 
+    public void markSuspect()
+    {
+        if (logger.isDebugEnabled())
+            logger.debug("Marking " + getFilename() + " as a suspect for blacklisting.");
+
+        isSuspect.getAndSet(true);
+    }
+
+    public boolean isMarkedSuspect()
+    {
+        return isSuspect.get();
+    }
+
     /**
      *
      * @param filter filter to use when reading the columns

http://git-wip-us.apache.org/repos/asf/cassandra/blob/9483e147/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
index 0a5240a..5e4f269 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
@@ -57,6 +57,7 @@ public class SSTableScanner implements CloseableIterator<IColumnIterator>
         }
         catch (IOException e)
         {
+            sstable.markSuspect();
             throw new IOError(e);
         }
         this.sstable = sstable;
@@ -74,6 +75,7 @@ public class SSTableScanner implements CloseableIterator<IColumnIterator>
         }
         catch (IOException e)
         {
+            sstable.markSuspect();
             throw new IOError(e);
         }
         this.sstable = sstable;
@@ -100,6 +102,7 @@ public class SSTableScanner implements CloseableIterator<IColumnIterator>
         }
         catch (IOException e)
         {
+            sstable.markSuspect();
             throw new RuntimeException("corrupt sstable", e);
         }
     }
@@ -154,6 +157,7 @@ public class SSTableScanner implements CloseableIterator<IColumnIterator>
             }
             catch (IOException e)
             {
+                sstable.markSuspect();
                 throw new RuntimeException(e);
             }
         }
@@ -185,6 +189,7 @@ public class SSTableScanner implements CloseableIterator<IColumnIterator>
             }
             catch (IOException e)
             {
+                sstable.markSuspect();
                 throw new RuntimeException(SSTableScanner.this + " failed to provide next
columns from " + this, e);
             }
         }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/9483e147/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
index ff6636d..23a2657 100644
--- a/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
@@ -18,7 +18,7 @@
 */
 package org.apache.cassandra.db.compaction;
 
-import java.io.IOException;
+import java.io.*;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.HashSet;
@@ -30,6 +30,7 @@ import java.util.concurrent.Future;
 
 import org.junit.Test;
 import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.assertNotNull;
 
 import org.apache.cassandra.CleanupHelper;
 import org.apache.cassandra.Util;
@@ -38,6 +39,7 @@ import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.filter.QueryFilter;
 import org.apache.cassandra.db.filter.QueryPath;
 import org.apache.cassandra.io.sstable.*;
+import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
 
@@ -46,6 +48,18 @@ public class CompactionsTest extends CleanupHelper
     public static final String TABLE1 = "Keyspace1";
 
     @Test
+    public void testBlacklistingWithSizeTieredCompactionStrategy() throws Exception
+    {
+        testBlacklisting(SizeTieredCompactionStrategy.class.getCanonicalName());
+    }
+
+    @Test
+    public void testBlacklistingWithLeveledCompactionStrategy() throws Exception
+    {
+        testBlacklisting(LeveledCompactionStrategy.class.getCanonicalName());
+    }
+
+    @Test
     public void testStandardColumnCompactions() throws IOException, ExecutionException, InterruptedException
     {
         // this test does enough rows to force multiple block indexes to be used
@@ -83,6 +97,7 @@ public class CompactionsTest extends CleanupHelper
 
         // make sure max timestamp of compacted sstables is recorded properly after compaction.
         assertMaxTimestamp(store, maxTimestampExpected);
+        store.truncate();
     }
 
 
@@ -269,4 +284,100 @@ public class CompactionsTest extends CleanupHelper
         cf = store.getColumnFamily(filter);
         assert cf == null || cf.isEmpty() : "should be empty: " + cf;
     }
+
+    public void testBlacklisting(String compactionStrategy) throws Exception
+    {
+        // this test does enough rows to force multiple block indexes to be used
+        Table table = Table.open(TABLE1);
+        final ColumnFamilyStore store = table.getColumnFamilyStore("Standard1");
+
+        final int ROWS_PER_SSTABLE = 10;
+        final int SSTABLES = DatabaseDescriptor.getIndexInterval() * 2 / ROWS_PER_SSTABLE;
+
+        store.setCompactionStrategyClass(compactionStrategy);
+
+        // disable compaction while flushing
+        store.disableAutoCompaction();
+        //test index corruption
+        //now create a few new SSTables
+        long maxTimestampExpected = Long.MIN_VALUE;
+        Set<DecoratedKey> inserted = new HashSet<DecoratedKey>();
+        for (int j = 0; j < SSTABLES; j++)
+        {
+            for (int i = 0; i < ROWS_PER_SSTABLE; i++)
+            {
+                DecoratedKey key = Util.dk(String.valueOf(i % 2));
+                RowMutation rm = new RowMutation(TABLE1, key.key);
+                long timestamp = j * ROWS_PER_SSTABLE + i;
+                rm.add(new QueryPath("Standard1", null, ByteBufferUtil.bytes(String.valueOf(i
/ 2))),
+                        ByteBufferUtil.EMPTY_BYTE_BUFFER,
+                        timestamp);
+                maxTimestampExpected = Math.max(timestamp, maxTimestampExpected);
+                rm.apply();
+                inserted.add(key);
+            }
+            store.forceBlockingFlush();
+            assertMaxTimestamp(store, maxTimestampExpected);
+            assertEquals(inserted.toString(), inserted.size(), Util.getRangeSlice(store).size());
+        }
+
+        Collection<SSTableReader> sstables = store.getSSTables();
+        int currentSSTable = 0;
+        int sstablesToCorrupt = 8;
+
+        // corrupt first 'sstablesToCorrupt' SSTables
+        for (SSTableReader sstable : sstables)
+        {
+            if(currentSSTable + 1 > sstablesToCorrupt)
+                break;
+
+            RandomAccessFile raf = null;
+
+            try
+            {
+                raf = new RandomAccessFile(sstable.getFilename(), "rw");
+                assertNotNull(raf);
+                raf.write(0xFFFFFF);
+            }
+            finally
+            {
+                FileUtils.closeQuietly(raf);
+            }
+
+            currentSSTable++;
+        }
+
+        int failures = 0;
+
+        // close error output steam to avoid printing ton of useless RuntimeException
+        System.err.close();
+
+        try
+        {
+            // in case something will go wrong we don't want to loop forever using for (;;)
+            for (int i = 0; i < sstables.size(); i++)
+            {
+                try
+                {
+                    store.forceMajorCompaction();
+                }
+                catch (Exception e)
+                {
+                    failures++;
+                    continue;
+                }
+
+                assertEquals(sstablesToCorrupt + 1, store.getSSTables().size());
+                break;
+            }
+        }
+        finally
+        {
+            System.setErr(new PrintStream(new ByteArrayOutputStream()));
+        }
+
+
+        store.truncate();
+        assertEquals(failures, sstablesToCorrupt);
+    }
 }


Mime
View raw message