natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Handle corrupt files on startup 

Patch by stefania; reviewed by marcuse for CASSANDRA-9686
This commit is contained in:
Stefania Alborghetti 2015-07-01 11:15:15 +08:00 committed by Marcus Eriksson
parent 19c00f7212
commit 1ecccede7d
7 changed files with 150 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
2.1.9
* Handle corrupt files on startup (CASSANDRA-9686)
* Fix clientutil jar and tests (CASSANDRA-9760)
* (cqlsh) Allow the SSL protocol version to be specified through the
config file or environment variables (CASSANDRA-9544)

9
conf/cassandra.yaml
View File

@ -112,11 +112,12 @@ partitioner: org.apache.cassandra.dht.Murmur3Partitioner
# commitlog_directory: /var/lib/cassandra/commitlog
# policy for data disk failures:
# die: shut down gossip and Thrift and kill the JVM for any fs errors or
# die: shut down gossip and client transports and kill the JVM for any fs errors or
# single-sstable errors, so the node can be replaced.
# 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.
# stop_paranoid: shut down gossip and client transports even for single-sstable errors,
# kill the JVM for errors during startup.
# stop: shut down gossip and client transports, leaving the node effectively dead, but
# can still be inspected via JMX, kill the JVM for errors during startup.
# best_effort: stop using the failed disk and respond to requests based on
# remaining available sstables. This means you WILL see obsolete
# data at CL.ONE!

137
src/java/org/apache/cassandra/io/sstable/SSTableReader.java
View File

@ -69,6 +69,7 @@ import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.FSError;
import org.apache.cassandra.io.compress.CompressionMetadata;
import org.apache.cassandra.io.sstable.metadata.CompactionMetadata;
import org.apache.cassandra.io.sstable.metadata.MetadataComponent;
@ -477,19 +478,27 @@ public class SSTableReader extends SSTable implements SelfRefCounted<SSTableRead
statsMetadata,
OpenReason.NORMAL);
// load index and filter
long start = System.nanoTime();
sstable.load(validationMetadata);
logger.debug("INDEX LOAD TIME for {}: {} ms.", descriptor, TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
try
{
// load index and filter
long start = System.nanoTime();
sstable.load(validationMetadata);
logger.debug("INDEX LOAD TIME for {}: {} ms.", descriptor, TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
sstable.setup(!validate);
if (validate)
sstable.validate();
sstable.setup(!validate);
if (validate)
sstable.validate();
if (sstable.getKeyCache() != null)
logger.debug("key cache contains {}/{} keys", sstable.getKeyCache().size(), sstable.getKeyCache().getCapacity());
if (sstable.getKeyCache() != null)
logger.debug("key cache contains {}/{} keys", sstable.getKeyCache().size(), sstable.getKeyCache().getCapacity());
return sstable;
return sstable;
}
catch (Throwable t)
{
sstable.selfRef().release();
throw t;
}
}
public static void logOpenException(Descriptor descriptor, IOException e)
@ -518,9 +527,21 @@ public class SSTableReader extends SSTable implements SelfRefCounted<SSTableRead
{
sstable = open(entry.getKey(), entry.getValue(), metadata, partitioner);
}
catch (CorruptSSTableException ex)
{
FileUtils.handleCorruptSSTable(ex);
logger.error("Corrupt sstable {}; skipping table", entry, ex);
return;
}
catch (FSError ex)
{
FileUtils.handleFSError(ex);
logger.error("Cannot read sstable {}; file system error, skipping table", entry, ex);
return;
}
catch (IOException ex)
{
logger.error("Corrupt sstable {}; skipped", entry, ex);
logger.error("Cannot read sstable {}; other IO error, skipping table", entry, ex);
return;
}
sstables.add(sstable);
@ -704,46 +725,71 @@ public class SSTableReader extends SSTable implements SelfRefCounted<SSTableRead
*/
private void load(boolean recreateBloomFilter, boolean saveSummaryIfCreated) throws IOException
{
SegmentedFile.Builder ibuilder = SegmentedFile.getBuilder(DatabaseDescriptor.getIndexAccessMode());
SegmentedFile.Builder dbuilder = compression
try
{
SegmentedFile.Builder ibuilder = SegmentedFile.getBuilder(DatabaseDescriptor.getIndexAccessMode());
SegmentedFile.Builder dbuilder = compression
? SegmentedFile.getCompressedBuilder()
: SegmentedFile.getBuilder(DatabaseDescriptor.getDiskAccessMode());
boolean summaryLoaded = loadSummary(ibuilder, dbuilder);
boolean builtSummary = false;
if (recreateBloomFilter || !summaryLoaded)
{
buildSummary(recreateBloomFilter, ibuilder, dbuilder, summaryLoaded, Downsampling.BASE_SAMPLING_LEVEL);
builtSummary = true;
}
boolean summaryLoaded = loadSummary(ibuilder, dbuilder);
boolean builtSummary = false;
if (recreateBloomFilter || !summaryLoaded)
{
buildSummary(recreateBloomFilter, ibuilder, dbuilder, summaryLoaded, Downsampling.BASE_SAMPLING_LEVEL);
builtSummary = true;
}
if (components.contains(Component.PRIMARY_INDEX))
ifile = ibuilder.complete(descriptor.filenameFor(Component.PRIMARY_INDEX));
if (components.contains(Component.PRIMARY_INDEX))
ifile = ibuilder.complete(descriptor.filenameFor(Component.PRIMARY_INDEX));
dfile = dbuilder.complete(descriptor.filenameFor(Component.DATA));
// Check for an index summary that was downsampled even though the serialization format doesn't support
// that. If it was downsampled, rebuild it. See CASSANDRA-8993 for details.
if (!descriptor.version.hasSamplingLevel && !builtSummary && !validateSummarySamplingLevel() && ifile != null)
{
indexSummary.close();
ifile.close();
dfile.close();
logger.info("Detected erroneously downsampled index summary; will rebuild summary at full sampling");
FileUtils.deleteWithConfirm(new File(descriptor.filenameFor(Component.SUMMARY)));
ibuilder = SegmentedFile.getBuilder(DatabaseDescriptor.getIndexAccessMode());
dbuilder = compression
? SegmentedFile.getCompressedBuilder()
: SegmentedFile.getBuilder(DatabaseDescriptor.getDiskAccessMode());
buildSummary(false, ibuilder, dbuilder, false, Downsampling.BASE_SAMPLING_LEVEL);
ifile = ibuilder.complete(descriptor.filenameFor(Component.PRIMARY_INDEX));
dfile = dbuilder.complete(descriptor.filenameFor(Component.DATA));
saveSummary(ibuilder, dbuilder);
// Check for an index summary that was downsampled even though the serialization format doesn't support
// that. If it was downsampled, rebuild it. See CASSANDRA-8993 for details.
if (!descriptor.version.hasSamplingLevel && !builtSummary && !validateSummarySamplingLevel() && ifile != null)
{
indexSummary.close();
ifile.close();
dfile.close();
logger.info("Detected erroneously downsampled index summary; will rebuild summary at full sampling");
FileUtils.deleteWithConfirm(new File(descriptor.filenameFor(Component.SUMMARY)));
ibuilder = SegmentedFile.getBuilder(DatabaseDescriptor.getIndexAccessMode());
dbuilder = compression
? SegmentedFile.getCompressedBuilder()
: SegmentedFile.getBuilder(DatabaseDescriptor.getDiskAccessMode());
buildSummary(false, ibuilder, dbuilder, false, Downsampling.BASE_SAMPLING_LEVEL);
ifile = ibuilder.complete(descriptor.filenameFor(Component.PRIMARY_INDEX));
dfile = dbuilder.complete(descriptor.filenameFor(Component.DATA));
saveSummary(ibuilder, dbuilder);
}
else if (saveSummaryIfCreated && builtSummary)
{
saveSummary(ibuilder, dbuilder);
}
}
else if (saveSummaryIfCreated && builtSummary)
{
saveSummary(ibuilder, dbuilder);
catch (Throwable t)
{ // Because the tidier has not been set-up yet in SSTableReader.open(), we must release the files in case of error
if (ifile != null)
{
ifile.close();
ifile = null;
}
if (dfile != null)
{
dfile.close();
dfile = null;
}
if (indexSummary != null)
{
indexSummary.close();
indexSummary = null;
}
throw t;
}
}
@ -2174,7 +2220,8 @@ public class SSTableReader extends SSTable implements SelfRefCounted<SSTableRead
summary.close();
if (runOnClose != null)
runOnClose.run();
dfile.close();
if (dfile != null)
dfile.close();
if (ifile != null)
ifile.close();
typeRef.release();

22
src/java/org/apache/cassandra/io/util/FileUtils.java
View File

@ -395,6 +395,9 @@ public class FileUtils
public static void handleCorruptSSTable(CorruptSSTableException e)
{
if (!StorageService.instance.isSetupCompleted())
handleStartupFSError(e);
JVMStabilityInspector.inspectThrowable(e);
switch (DatabaseDescriptor.getDiskFailurePolicy())
{
@ -406,6 +409,9 @@ public class FileUtils
public static void handleFSError(FSError e)
{
if (!StorageService.instance.isSetupCompleted())
handleStartupFSError(e);
JVMStabilityInspector.inspectThrowable(e);
switch (DatabaseDescriptor.getDiskFailurePolicy())
{
@ -431,6 +437,22 @@ public class FileUtils
}
}
private static void handleStartupFSError(Throwable t)
{
switch (DatabaseDescriptor.getDiskFailurePolicy())
{
case stop_paranoid:
case stop:
case die:
logger.error("Exiting forcefully due to file system exception on startup, disk failure policy \"{}\"",
DatabaseDescriptor.getDiskFailurePolicy(),
t);
JVMStabilityInspector.killCurrentJVM(t, true);
break;
default:
break;
}
}
/**
* Get the size of a directory in bytes
* @param directory The directory for which we need size.

9
src/java/org/apache/cassandra/service/CassandraDaemon.java
View File

@ -134,7 +134,7 @@ public class CassandraDaemon
public Server thriftServer;
public Server nativeServer;
private boolean setupCompleted = false;
/**
* This is a hook for concrete daemons to initialize themselves suitably.
*
@ -425,6 +425,13 @@ public class CassandraDaemon
InetAddress nativeAddr = DatabaseDescriptor.getRpcAddress();
int nativePort = DatabaseDescriptor.getNativeTransportPort();
nativeServer = new org.apache.cassandra.transport.Server(nativeAddr, nativePort);
setupCompleted = true;
}
public boolean setupCompleted()
{
return setupCompleted;
}
/**

7
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -486,6 +486,13 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
return initialized;
}
public boolean isSetupCompleted()
{
return daemon == null
? false
: daemon.setupCompleted();
}
public void stopDaemon()
{
if (daemon == null)

17
src/java/org/apache/cassandra/utils/JVMStabilityInspector.java
View File

@ -72,6 +72,11 @@ public final class JVMStabilityInspector
inspectThrowable(t);
}
public static void killCurrentJVM(Throwable t, boolean quiet)
{
killer.killCurrentJVM(t, quiet);
}
@VisibleForTesting
public static Killer replaceKiller(Killer newKiller) {
Killer oldKiller = JVMStabilityInspector.killer;
@ -90,8 +95,16 @@ public final class JVMStabilityInspector
*/
protected void killCurrentJVM(Throwable t)
{
t.printStackTrace(System.err);
logger.error("JVM state determined to be unstable. Exiting forcefully due to:", t);
killCurrentJVM(t, false);
}
protected void killCurrentJVM(Throwable t, boolean quiet)
{
if (!quiet)
{
t.printStackTrace(System.err);
logger.error("JVM state determined to be unstable. Exiting forcefully due to:", t);
}
StorageService.instance.removeShutdownHook();
System.exit(100);
}