diff --git a/CHANGES.txt b/CHANGES.txt index 6ca163abc7..f9da65c561 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -27,6 +27,10 @@ * New counters implementation (CASSANDRA-6504) +2.0.6 + * Let scrub optionally skip broken counter partitions (CASSANDRA-5930) + + 2.0.5 * Reduce garbage generated by bloom filter lookups (CASSANDRA-6609) * Add ks.cf names to tombstone logging (CASSANDRA-6597) diff --git a/NEWS.txt b/NEWS.txt index 72b898ef78..185f60c6e2 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -44,11 +44,21 @@ Upgrading (See https://issues.apache.org/jira/browse/CASSANDRA-6504 and the dev blog post at http://www.datastax.com/dev/blog/ for details). +2.0.6 +===== + +New features +------------ + - Scrub can now optionally skip corrupt counter partitions. Please note + that this will lead to the loss of all the counter updates in the skipped + partition. See the --skip-corrupted option. + + 2.0.5 ===== New features --------- +------------ - Batchlog replay can be, and is throttled by default now. See batchlog_replay_throttle_in_kb setting in cassandra.yaml. diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index a46cde54ef..14b35718d2 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1303,12 +1303,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean CompactionManager.instance.performCleanup(ColumnFamilyStore.this); } - public void scrub(boolean disableSnapshot) throws ExecutionException, InterruptedException + public void scrub(boolean disableSnapshot, boolean skipCorrupted) throws ExecutionException, InterruptedException { // skip snapshot creation during scrub, SEE JIRA 5891 if(!disableSnapshot) snapshotWithoutFlush("pre-scrub-" + System.currentTimeMillis()); - CompactionManager.instance.performScrub(ColumnFamilyStore.this); + CompactionManager.instance.performScrub(ColumnFamilyStore.this, skipCorrupted); } public void sstablesRewrite(boolean excludeCurrentVersion) throws ExecutionException, InterruptedException diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index eeed45f142..7a04249d67 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -231,13 +231,13 @@ public class CompactionManager implements CompactionManagerMBean executor.submit(runnable).get(); } - public void performScrub(ColumnFamilyStore cfStore) throws InterruptedException, ExecutionException + public void performScrub(ColumnFamilyStore cfStore, final boolean skipCorrupted) throws InterruptedException, ExecutionException { performAllSSTableOperation(cfStore, new AllSSTablesOperation() { public void perform(ColumnFamilyStore store, Iterable sstables) throws IOException { - doScrub(store, sstables); + doScrub(store, sstables, skipCorrupted); } }); } @@ -420,16 +420,16 @@ public class CompactionManager implements CompactionManagerMBean * * @throws IOException */ - private void doScrub(ColumnFamilyStore cfs, Iterable sstables) throws IOException + private void doScrub(ColumnFamilyStore cfs, Iterable sstables, boolean skipCorrupted) throws IOException { assert !cfs.isIndex(); for (final SSTableReader sstable : sstables) - scrubOne(cfs, sstable); + scrubOne(cfs, sstable, skipCorrupted); } - private void scrubOne(ColumnFamilyStore cfs, SSTableReader sstable) throws IOException + private void scrubOne(ColumnFamilyStore cfs, SSTableReader sstable, boolean skipCorrupted) throws IOException { - Scrubber scrubber = new Scrubber(cfs, sstable); + Scrubber scrubber = new Scrubber(cfs, sstable, skipCorrupted); CompactionInfo.Holder scrubInfo = scrubber.getScrubInfo(); metrics.beginCompaction(scrubInfo); diff --git a/src/java/org/apache/cassandra/db/compaction/Scrubber.java b/src/java/org/apache/cassandra/db/compaction/Scrubber.java index 978865c1b8..666b933241 100644 --- a/src/java/org/apache/cassandra/db/compaction/Scrubber.java +++ b/src/java/org/apache/cassandra/db/compaction/Scrubber.java @@ -35,6 +35,7 @@ public class Scrubber implements Closeable public final ColumnFamilyStore cfs; public final SSTableReader sstable; public final File destination; + public final boolean skipCorrupted; private final CompactionController controller; private final boolean isCommutative; @@ -63,16 +64,17 @@ public class Scrubber implements Closeable }; private final SortedSet outOfOrderRows = new TreeSet<>(rowComparator); - public Scrubber(ColumnFamilyStore cfs, SSTableReader sstable) throws IOException + public Scrubber(ColumnFamilyStore cfs, SSTableReader sstable, boolean skipCorrupted) throws IOException { - this(cfs, sstable, new OutputHandler.LogOutput(), false); + this(cfs, sstable, skipCorrupted, new OutputHandler.LogOutput(), false); } - public Scrubber(ColumnFamilyStore cfs, SSTableReader sstable, OutputHandler outputHandler, boolean isOffline) throws IOException + public Scrubber(ColumnFamilyStore cfs, SSTableReader sstable, boolean skipCorrupted, OutputHandler outputHandler, boolean isOffline) throws IOException { this.cfs = cfs; this.sstable = sstable; this.outputHandler = outputHandler; + this.skipCorrupted = skipCorrupted; // Calculate the expected compacted filesize this.destination = cfs.directories.getDirectoryForNewSSTables(); @@ -166,7 +168,9 @@ public class Scrubber implements Closeable if (!sstable.descriptor.version.hasRowSizeAndColumnCount) { dataSize = dataSizeFromIndex; - outputHandler.debug(String.format("row %s is %s bytes", ByteBufferUtil.bytesToHex(key.key), dataSize)); + // avoid an NPE if key is null + String keyName = key == null ? "(unreadable key)" : ByteBufferUtil.bytesToHex(key.key); + outputHandler.debug(String.format("row %s is %s bytes", keyName, dataSize)); } else { @@ -203,7 +207,7 @@ public class Scrubber implements Closeable catch (Throwable th) { throwIfFatal(th); - outputHandler.warn("Non-fatal error reading row (stacktrace follows)", th); + outputHandler.warn("Error reading row (stacktrace follows):", th); writer.resetAndTruncate(); if (currentIndexKey != null @@ -231,9 +235,7 @@ public class Scrubber implements Closeable catch (Throwable th2) { throwIfFatal(th2); - // Skipping rows is dangerous for counters (see CASSANDRA-2759) - if (isCommutative) - throw new IOError(th2); + throwIfCommutative(key, th2); outputHandler.warn("Retry failed too. Skipping to next row (retry's stacktrace follows)", th2); writer.resetAndTruncate(); @@ -243,11 +245,9 @@ public class Scrubber implements Closeable } else { - // Skipping rows is dangerous for counters (see CASSANDRA-2759) - if (isCommutative) - throw new IOError(th); + throwIfCommutative(key, th); - outputHandler.warn("Row at " + dataStart + " is unreadable; skipping to next"); + outputHandler.warn("Row starting at position " + dataStart + " is unreadable; skipping to next"); if (currentIndexKey != null) dataFile.seek(nextRowPositionFromIndex); badRows++; @@ -324,6 +324,19 @@ public class Scrubber implements Closeable throw (Error) th; } + private void throwIfCommutative(DecoratedKey key, Throwable th) + { + if (isCommutative && !skipCorrupted) + { + outputHandler.warn(String.format("An error occurred while scrubbing the row with key '%s'. Skipping corrupt " + + "rows in counter tables will result in undercounts for the affected " + + "counters (see CASSANDRA-2759 for more details), so by default the scrub will " + + "stop at this point. If you would like to skip the row anyway and continue " + + "scrubbing, re-run the scrub with the --skip-corrupted option.", key)); + throw new IOError(th); + } + } + public void close() { FileUtils.closeQuietly(dataFile); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 1c59351083..c40b28d935 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -2143,10 +2143,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } } - public void scrub(boolean disableSnapshot, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException + public void scrub(boolean disableSnapshot, boolean skipCorrupted, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException { for (ColumnFamilyStore cfStore : getValidColumnFamilies(false, false, keyspaceName, columnFamilies)) - cfStore.scrub(disableSnapshot); + cfStore.scrub(disableSnapshot, skipCorrupted); } public void upgradeSSTables(String keyspaceName, boolean excludeCurrentVersion, String... columnFamilies) throws IOException, ExecutionException, InterruptedException diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index 390795e08e..c7b51d64a2 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -244,7 +244,7 @@ public interface StorageServiceMBean extends NotificationEmitter * * Scrubbed CFs will be snapshotted first, if disableSnapshot is false */ - public void scrub(boolean disableSnapshot, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException; + public void scrub(boolean disableSnapshot, boolean skipCorrupted, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException; /** * Rewrite all sstables to the latest version. diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index 830097cdf5..e44b576569 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -191,9 +191,9 @@ public class NodeProbe implements AutoCloseable ssProxy.forceKeyspaceCleanup(keyspaceName, columnFamilies); } - public void scrub(boolean disableSnapshot, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException + public void scrub(boolean disableSnapshot, boolean skipCorrupted, String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException { - ssProxy.scrub(disableSnapshot, keyspaceName, columnFamilies); + ssProxy.scrub(disableSnapshot, skipCorrupted, keyspaceName, columnFamilies); } public void upgradeSSTables(String keyspaceName, boolean excludeCurrentVersion, String... columnFamilies) throws IOException, ExecutionException, InterruptedException diff --git a/src/java/org/apache/cassandra/tools/NodeTool.java b/src/java/org/apache/cassandra/tools/NodeTool.java index 7c49e236bb..cc94b3553f 100644 --- a/src/java/org/apache/cassandra/tools/NodeTool.java +++ b/src/java/org/apache/cassandra/tools/NodeTool.java @@ -971,9 +971,16 @@ public class NodeTool @Arguments(usage = "[ ...]", description = "The keyspace followed by one or many column families") private List args = new ArrayList<>(); - @Option(title = "disable_snapshot", name = {"-ns", "--no-snapshot"}, description = "Scrubbed CFs will be snapshotted first, if disableSnapshot is false. (default false)") + @Option(title = "disable_snapshot", + name = {"-ns", "--no-snapshot"}, + description = "Scrubbed CFs will be snapshotted first, if disableSnapshot is false. (default false)") private boolean disableSnapshot = false; + @Option(title = "skip_corrupted", + name = {"-s", "--skip-corrupted"}, + description = "Skip corrupted partitions even when scrubbing counter tables. (default false)") + private boolean skipCorrupted = false; + @Override public void execute(NodeProbe probe) { @@ -984,7 +991,7 @@ public class NodeTool { try { - probe.scrub(disableSnapshot, keyspace, cfnames); + probe.scrub(disableSnapshot, skipCorrupted, keyspace, cfnames); } catch (Exception e) { throw new RuntimeException("Error occurred during flushing", e); diff --git a/src/java/org/apache/cassandra/tools/StandaloneScrubber.java b/src/java/org/apache/cassandra/tools/StandaloneScrubber.java index 5531aee39e..ecd530436a 100644 --- a/src/java/org/apache/cassandra/tools/StandaloneScrubber.java +++ b/src/java/org/apache/cassandra/tools/StandaloneScrubber.java @@ -43,6 +43,7 @@ public class StandaloneScrubber private static final String DEBUG_OPTION = "debug"; private static final String HELP_OPTION = "help"; private static final String MANIFEST_CHECK_OPTION = "manifest-check"; + private static final String SKIP_CORRUPTED_OPTION = "skip-corrupted"; public static void main(String args[]) { @@ -106,7 +107,7 @@ public class StandaloneScrubber { try { - Scrubber scrubber = new Scrubber(cfs, sstable, handler, true); + Scrubber scrubber = new Scrubber(cfs, sstable, options.skipCorrupted, handler, true); try { scrubber.scrub(); @@ -171,6 +172,7 @@ public class StandaloneScrubber public boolean debug; public boolean verbose; public boolean manifestCheckOnly; + public boolean skipCorrupted; private Options(String keyspaceName, String cfName) { @@ -209,6 +211,7 @@ public class StandaloneScrubber opts.debug = cmd.hasOption(DEBUG_OPTION); opts.verbose = cmd.hasOption(VERBOSE_OPTION); opts.manifestCheckOnly = cmd.hasOption(MANIFEST_CHECK_OPTION); + opts.skipCorrupted = cmd.hasOption(SKIP_CORRUPTED_OPTION); return opts; } @@ -233,6 +236,7 @@ public class StandaloneScrubber options.addOption("v", VERBOSE_OPTION, "verbose output"); options.addOption("h", HELP_OPTION, "display this help message"); options.addOption("m", MANIFEST_CHECK_OPTION, "only check and repair the leveled manifest, without actually scrubbing the sstables"); + options.addOption("s", SKIP_CORRUPTED_OPTION, "skip corrupt rows in counter tables"); return options; } diff --git a/test/unit/org/apache/cassandra/db/ScrubTest.java b/test/unit/org/apache/cassandra/db/ScrubTest.java index 38c8b62c46..d8ab9ff337 100644 --- a/test/unit/org/apache/cassandra/db/ScrubTest.java +++ b/test/unit/org/apache/cassandra/db/ScrubTest.java @@ -20,13 +20,15 @@ package org.apache.cassandra.db; * */ -import java.io.File; -import java.io.IOException; +import java.io.*; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; import java.util.concurrent.ExecutionException; +import org.apache.cassandra.db.compaction.OperationType; +import org.apache.commons.lang3.StringUtils; import org.junit.Test; import org.junit.runner.RunWith; @@ -39,9 +41,11 @@ import org.apache.cassandra.db.compaction.Scrubber; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.db.columniterator.IdentityQueryFilter; import org.apache.cassandra.db.compaction.CompactionManager; +import org.apache.cassandra.exceptions.WriteTimeoutException; import org.apache.cassandra.io.sstable.*; import org.apache.cassandra.utils.ByteBufferUtil; +import static org.apache.cassandra.Util.cellname; import static org.apache.cassandra.Util.column; import static org.junit.Assert.assertEquals; import static org.junit.Assert.fail; @@ -52,6 +56,7 @@ public class ScrubTest extends SchemaLoader public String KEYSPACE = "Keyspace1"; public String CF = "Standard1"; public String CF3 = "Standard2"; + public String COUNTER_CF = "Counter1"; @Test public void testScrubOneRow() throws IOException, ExecutionException, InterruptedException, ConfigurationException @@ -68,13 +73,60 @@ public class ScrubTest extends SchemaLoader rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000); assertEquals(1, rows.size()); - CompactionManager.instance.performScrub(cfs); + CompactionManager.instance.performScrub(cfs, false); // check data is still there rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000); assertEquals(1, rows.size()); } + @Test + public void testScrubCorruptedCounterRow() throws IOException, InterruptedException, ExecutionException, WriteTimeoutException + { + CompactionManager.instance.disableAutoCompaction(); + Keyspace keyspace = Keyspace.open(KEYSPACE); + ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(COUNTER_CF); + cfs.clearUnsafe(); + + fillCounterCF(cfs, 2); + + List rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000); + assertEquals(2, rows.size()); + + SSTableReader sstable = cfs.getSSTables().iterator().next(); + + // overwrite one row with garbage + long row0Start = sstable.getPosition(RowPosition.forKey(ByteBufferUtil.bytes("0"), sstable.partitioner), SSTableReader.Operator.EQ).position; + long row1Start = sstable.getPosition(RowPosition.forKey(ByteBufferUtil.bytes("1"), sstable.partitioner), SSTableReader.Operator.EQ).position; + long startPosition = row0Start < row1Start ? row0Start : row1Start; + long endPosition = row0Start < row1Start ? row1Start : row0Start; + + RandomAccessFile file = new RandomAccessFile(sstable.getFilename(), "rw"); + file.seek(startPosition); + file.writeBytes(StringUtils.repeat('z', (int) (endPosition - startPosition))); + file.close(); + + // with skipCorrupted == false, the scrub is expected to fail + Scrubber scrubber = new Scrubber(cfs, sstable, false); + try + { + scrubber.scrub(); + fail("Expected a CorruptSSTableException to be thrown"); + } + catch (IOError err) {} + + // with skipCorrupted == true, the corrupt row will be skipped + scrubber = new Scrubber(cfs, sstable, true); + scrubber.scrub(); + scrubber.close(); + cfs.replaceCompactedSSTables(Collections.singletonList(sstable), Collections.singletonList(scrubber.getNewSSTable()), OperationType.SCRUB); + assertEquals(1, cfs.getSSTables().size()); + + // verify that we can read all of the rows, and there is now one less row + rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000); + assertEquals(1, rows.size()); + } + @Test public void testScrubDeletedRow() throws IOException, ExecutionException, InterruptedException, ConfigurationException { @@ -89,7 +141,7 @@ public class ScrubTest extends SchemaLoader rm.applyUnsafe(); cfs.forceBlockingFlush(); - CompactionManager.instance.performScrub(cfs); + CompactionManager.instance.performScrub(cfs, false); assert cfs.getSSTables().isEmpty(); } @@ -108,7 +160,7 @@ public class ScrubTest extends SchemaLoader rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000); assertEquals(10, rows.size()); - CompactionManager.instance.performScrub(cfs); + CompactionManager.instance.performScrub(cfs, false); // check data is still there rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000); @@ -145,7 +197,6 @@ public class ScrubTest extends SchemaLoader writer.closeAndOpenReader(); */ - String root = System.getProperty("corrupt-sstable-root"); assert root != null; File rootDir = new File(root); @@ -171,7 +222,7 @@ public class ScrubTest extends SchemaLoader components.add(Component.TOC); SSTableReader sstable = SSTableReader.openNoValidation(desc, components, metadata); - Scrubber scrubber = new Scrubber(cfs, sstable); + Scrubber scrubber = new Scrubber(cfs, sstable, false); scrubber.scrub(); cfs.loadNewSSTables(); @@ -207,4 +258,20 @@ public class ScrubTest extends SchemaLoader cfs.forceBlockingFlush(); } + + protected void fillCounterCF(ColumnFamilyStore cfs, int rowsPerSSTable) throws ExecutionException, InterruptedException, IOException, WriteTimeoutException + { + for (int i = 0; i < rowsPerSSTable; i++) + { + String key = String.valueOf(i); + ColumnFamily cf = TreeMapBackedSortedColumns.factory.create(KEYSPACE, COUNTER_CF); + Mutation rm = new Mutation(KEYSPACE, ByteBufferUtil.bytes(key), cf); + rm.addCounter(COUNTER_CF, cellname("Column1"), 100); + CounterMutation cm = new CounterMutation(rm, ConsistencyLevel.ONE); + cm.apply(); + } + + cfs.forceBlockingFlush(); + } + }