You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ma...@apache.org on 2021/04/22 11:19:00 UTC
[cassandra] branch trunk updated: Minor nodetool verify fixes
This is an automated email from the ASF dual-hosted git repository.
marcuse pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/trunk by this push:
new 2bd07ec Minor nodetool verify fixes
2bd07ec is described below
commit 2bd07ecab6121ebfc8a192243c5c6fb41eb85515
Author: Marcus Eriksson <ma...@apache.org>
AuthorDate: Wed Apr 14 13:17:24 2021 +0200
Minor nodetool verify fixes
Patch by marcuse; reviewed by Caleb Rackliffe and Jon Meredith for CASSANDRA-16608
---
.../apache/cassandra/db/compaction/Verifier.java | 49 +++++++++++---------
test/unit/org/apache/cassandra/db/VerifyTest.java | 54 +++++++++++++++++++++-
2 files changed, 81 insertions(+), 22 deletions(-)
diff --git a/src/java/org/apache/cassandra/db/compaction/Verifier.java b/src/java/org/apache/cassandra/db/compaction/Verifier.java
index 5a04235..18b415c 100644
--- a/src/java/org/apache/cassandra/db/compaction/Verifier.java
+++ b/src/java/org/apache/cassandra/db/compaction/Verifier.java
@@ -24,6 +24,7 @@ import com.google.common.collect.ImmutableSet;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
+import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.Component;
@@ -57,6 +58,7 @@ import java.io.IOError;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
+import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.locks.Lock;
@@ -142,7 +144,7 @@ public class Verifier implements Closeable
catch (Throwable t)
{
outputHandler.warn(t.getMessage());
- markAndThrow(false);
+ markAndThrow(t, false);
}
try
@@ -153,7 +155,7 @@ public class Verifier implements Closeable
catch (Throwable t)
{
outputHandler.warn(t.getMessage());
- markAndThrow();
+ markAndThrow(t);
}
try
@@ -165,7 +167,7 @@ public class Verifier implements Closeable
{
outputHandler.output("Index summary is corrupt - if it is removed it will get rebuilt on startup "+sstable.descriptor.filenameFor(Component.SUMMARY));
outputHandler.warn(t.getMessage());
- markAndThrow(false);
+ markAndThrow(t, false);
}
try
@@ -177,10 +179,10 @@ public class Verifier implements Closeable
catch (Throwable t)
{
outputHandler.warn(t.getMessage());
- markAndThrow();
+ markAndThrow(t);
}
- if (options.checkOwnsTokens && !isOffline)
+ if (options.checkOwnsTokens && !isOffline && !(cfs.getPartitioner() instanceof LocalPartitioner))
{
outputHandler.debug("Checking that all tokens are owned by the current node");
try (KeyIterator iter = new KeyIterator(sstable.descriptor, sstable.metadata()))
@@ -198,7 +200,7 @@ public class Verifier implements Closeable
catch (Throwable t)
{
outputHandler.warn(t.getMessage());
- markAndThrow();
+ markAndThrow(t);
}
}
@@ -225,7 +227,7 @@ public class Verifier implements Closeable
catch (IOException e)
{
outputHandler.warn(e.getMessage());
- markAndThrow();
+ markAndThrow(e);
}
finally
{
@@ -243,7 +245,7 @@ public class Verifier implements Closeable
{
long firstRowPositionFromIndex = rowIndexEntrySerializer.deserializePositionAndSkip(indexFile);
if (firstRowPositionFromIndex != 0)
- markAndThrow();
+ markAndThrow(new RuntimeException("firstRowPositionFromIndex != 0: "+firstRowPositionFromIndex));
}
List<Range<Token>> ownedRanges = isOffline ? Collections.emptyList() : Range.normalize(tokenLookup.apply(cfs.metadata().keyspace));
@@ -270,7 +272,7 @@ public class Verifier implements Closeable
// check for null key below
}
- if (options.checkOwnsTokens && ownedRanges.size() > 0)
+ if (options.checkOwnsTokens && ownedRanges.size() > 0 && !(cfs.getPartitioner() instanceof LocalPartitioner))
{
try
{
@@ -279,7 +281,7 @@ public class Verifier implements Closeable
catch (Throwable t)
{
outputHandler.warn(String.format("Key %s in sstable %s not owned by local ranges %s", key, sstable, ownedRanges), t);
- markAndThrow();
+ markAndThrow(t);
}
}
@@ -294,7 +296,7 @@ public class Verifier implements Closeable
}
catch (Throwable th)
{
- markAndThrow();
+ markAndThrow(th);
}
long dataStart = dataFile.getFilePointer();
@@ -312,7 +314,7 @@ public class Verifier implements Closeable
try
{
if (key == null || dataSize > dataFile.length())
- markAndThrow();
+ markAndThrow(new RuntimeException(String.format("key = %s, dataSize=%d, dataFile.length() = %d", key, dataSize, dataFile.length())));
//mimic the scrub read path, intentionally unused
try (UnfilteredRowIterator iterator = SSTableIdentityIterator.create(sstable, dataFile, key))
@@ -320,7 +322,7 @@ public class Verifier implements Closeable
}
if ( (prevKey != null && prevKey.compareTo(key) > 0) || !key.getKey().equals(currentIndexKey) || dataStart != dataStartFromIndex )
- markAndThrow();
+ markAndThrow(new RuntimeException("Key out of order: previous = "+prevKey + " : current = " + key));
goodRows++;
prevKey = key;
@@ -331,7 +333,7 @@ public class Verifier implements Closeable
}
catch (Throwable th)
{
- markAndThrow();
+ markAndThrow(th);
}
}
}
@@ -442,9 +444,14 @@ public class Verifier implements Closeable
private void deserializeBloomFilter(SSTableReader sstable) throws IOException
{
- try (DataInputStream stream = new DataInputStream(new BufferedInputStream(Files.newInputStream(Paths.get(sstable.descriptor.filenameFor(Component.FILTER)))));
- IFilter bf = BloomFilterSerializer.deserialize(stream, sstable.descriptor.version.hasOldBfFormat()))
- {}
+ Path bfPath = Paths.get(sstable.descriptor.filenameFor(Component.FILTER));
+ if (Files.exists(bfPath))
+ {
+ try (DataInputStream stream = new DataInputStream(new BufferedInputStream(Files.newInputStream(bfPath)));
+ IFilter bf = BloomFilterSerializer.deserialize(stream, sstable.descriptor.version.hasOldBfFormat()))
+ {
+ }
+ }
}
public void close()
@@ -467,12 +474,12 @@ public class Verifier implements Closeable
throw (Error) th;
}
- private void markAndThrow()
+ private void markAndThrow(Throwable cause)
{
- markAndThrow(true);
+ markAndThrow(cause, true);
}
- private void markAndThrow(boolean mutateRepaired)
+ private void markAndThrow(Throwable cause, boolean mutateRepaired)
{
if (mutateRepaired && options.mutateRepairStatus) // if we are able to mutate repaired flag, an incremental repair should be enough
{
@@ -486,7 +493,7 @@ public class Verifier implements Closeable
outputHandler.output("Error mutating repairedAt for SSTable " + sstable.getFilename() + ", as part of markAndThrow");
}
}
- Exception e = new Exception(String.format("Invalid SSTable %s, please force %srepair", sstable.getFilename(), (mutateRepaired && options.mutateRepairStatus) ? "" : "a full "));
+ Exception e = new Exception(String.format("Invalid SSTable %s, please force %srepair", sstable.getFilename(), (mutateRepaired && options.mutateRepairStatus) ? "" : "a full "), cause);
if (options.invokeDiskFailurePolicy)
throw new CorruptSSTableException(e, sstable.getFilename());
else
diff --git a/test/unit/org/apache/cassandra/db/VerifyTest.java b/test/unit/org/apache/cassandra/db/VerifyTest.java
index df2acb4..4b73f27 100644
--- a/test/unit/org/apache/cassandra/db/VerifyTest.java
+++ b/test/unit/org/apache/cassandra/db/VerifyTest.java
@@ -22,6 +22,8 @@ import com.google.common.base.Charsets;
import org.apache.cassandra.OrderedJUnit4ClassRunner;
import org.apache.cassandra.Util;
+import org.apache.cassandra.batchlog.Batch;
+import org.apache.cassandra.batchlog.BatchlogManager;
import org.apache.cassandra.cache.ChunkCache;
import org.apache.cassandra.UpdateBuilder;
import org.apache.cassandra.db.compaction.AbstractCompactionStrategy;
@@ -52,11 +54,13 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import java.io.*;
+import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.zip.CRC32;
import java.util.zip.CheckedInputStream;
@@ -65,6 +69,7 @@ import static org.apache.cassandra.SchemaLoader.counterCFMD;
import static org.apache.cassandra.SchemaLoader.createKeyspace;
import static org.apache.cassandra.SchemaLoader.loadSchema;
import static org.apache.cassandra.SchemaLoader.standardCFMD;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -87,6 +92,7 @@ public class VerifyTest
public static final String CORRUPTCOUNTER_CF2 = "CounterCorrupt2";
public static final String CF_UUID = "UUIDKeys";
+ public static final String BF_ALWAYS_PRESENT = "BfAlwaysPresent";
@BeforeClass
public static void defineSchema() throws ConfigurationException
@@ -108,7 +114,8 @@ public class VerifyTest
counterCFMD(KEYSPACE, COUNTER_CF4),
counterCFMD(KEYSPACE, CORRUPTCOUNTER_CF),
counterCFMD(KEYSPACE, CORRUPTCOUNTER_CF2),
- standardCFMD(KEYSPACE, CF_UUID, 0, UUIDType.instance));
+ standardCFMD(KEYSPACE, CF_UUID, 0, UUIDType.instance),
+ standardCFMD(KEYSPACE, BF_ALWAYS_PRESENT).bloomFilterFpChance(1.0));
}
@@ -676,6 +683,51 @@ public class VerifyTest
new Verifier.RangeOwnHelper(Collections.emptyList()).validate(dk(1));
}
+ @Test
+ public void testVerifyLocalPartitioner() throws UnknownHostException
+ {
+ TokenMetadata tmd = StorageService.instance.getTokenMetadata();
+ byte[] tk1 = new byte[1], tk2 = new byte[1];
+ tk1[0] = 2;
+ tk2[0] = 1;
+ tmd.updateNormalToken(new ByteOrderedPartitioner.BytesToken(tk1), InetAddressAndPort.getByName("127.0.0.1"));
+ tmd.updateNormalToken(new ByteOrderedPartitioner.BytesToken(tk2), InetAddressAndPort.getByName("127.0.0.2"));
+ // write some bogus to a localpartitioner table
+ Batch bogus = Batch.createLocal(UUID.randomUUID(), 0, Collections.emptyList());
+ BatchlogManager.store(bogus);
+ ColumnFamilyStore cfs = Keyspace.open("system").getColumnFamilyStore("batches");
+ cfs.forceBlockingFlush();
+ for (SSTableReader sstable : cfs.getLiveSSTables())
+ {
+
+ try (Verifier verifier = new Verifier(cfs, sstable, false, Verifier.options().checkOwnsTokens(true).build()))
+ {
+ verifier.verify();
+ }
+ }
+ }
+
+ @Test
+ public void testNoFilterFile()
+ {
+ CompactionManager.instance.disableAutoCompaction();
+ Keyspace keyspace = Keyspace.open(KEYSPACE);
+ ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(BF_ALWAYS_PRESENT);
+ fillCF(cfs, 100);
+ assertEquals(1.0, cfs.metadata().params.bloomFilterFpChance, 0.0);
+ for (SSTableReader sstable : cfs.getLiveSSTables())
+ {
+ File f = new File(sstable.descriptor.filenameFor(Component.FILTER));
+ assertFalse(f.exists());
+ try (Verifier verifier = new Verifier(cfs, sstable, false, Verifier.options().build()))
+ {
+ verifier.verify();
+ }
+ }
+ }
+
+
+
private DecoratedKey dk(long l)
{
return new BufferDecoratedKey(t(l), ByteBufferUtil.EMPTY_BYTE_BUFFER);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org