You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by dc...@apache.org on 2020/07/30 00:28:35 UTC
[cassandra] branch trunk updated (a2a3526 -> 3e393f2)
This is an automated email from the ASF dual-hosted git repository.
dcapwell pushed a change to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git.
from a2a3526 Merge branch 'cassandra-3.11' into trunk
new db03460 stop_paranoid disk failure policy is ignored on CorruptSSTableException after node is up
new 86b7727 Merge branch 'cassandra-3.0' into cassandra-3.11
new 3e393f2 Merge branch 'cassandra-3.11' into trunk
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CHANGES.txt | 1 +
.../AbstractLocalAwareExecutorService.java | 12 +-
src/java/org/apache/cassandra/db/Directories.java | 3 +-
.../commitlog/AbstractCommitLogSegmentManager.java | 1 -
.../cassandra/io/sstable/format/SSTableReader.java | 4 +-
.../org/apache/cassandra/io/util/FileUtils.java | 2 +-
.../apache/cassandra/service/CassandraDaemon.java | 21 +-
.../cassandra/service/DefaultFSErrorHandler.java | 6 +-
.../cassandra/utils/JVMStabilityInspector.java | 30 ++-
.../org/apache/cassandra/utils/binlog/BinLog.java | 3 +-
...bilityInspectorCorruptSSTableExceptionTest.java | 206 +++++++++++++++
.../io/sstable/format/ForwardingSSTableReader.java | 293 ++++++++++++++++-----
.../org/apache/cassandra/db/DirectoriesTest.java | 3 +-
.../cassandra/utils/JVMStabilityInspectorTest.java | 14 +
14 files changed, 486 insertions(+), 113 deletions(-)
create mode 100644 test/distributed/org/apache/cassandra/distributed/test/JVMStabilityInspectorCorruptSSTableExceptionTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[cassandra] 01/01: Merge branch 'cassandra-3.11' into trunk
Posted by dc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dcapwell pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 3e393f25bd928d66470b9ba8114032341df396c9
Merge: a2a3526 86b7727
Author: David Capwell <dc...@gmail.com>
AuthorDate: Wed Jul 29 17:21:43 2020 -0700
Merge branch 'cassandra-3.11' into trunk
CHANGES.txt | 1 +
.../AbstractLocalAwareExecutorService.java | 12 +-
src/java/org/apache/cassandra/db/Directories.java | 3 +-
.../commitlog/AbstractCommitLogSegmentManager.java | 1 -
.../cassandra/io/sstable/format/SSTableReader.java | 4 +-
.../org/apache/cassandra/io/util/FileUtils.java | 2 +-
.../apache/cassandra/service/CassandraDaemon.java | 21 +-
.../cassandra/service/DefaultFSErrorHandler.java | 6 +-
.../cassandra/utils/JVMStabilityInspector.java | 30 ++-
.../org/apache/cassandra/utils/binlog/BinLog.java | 3 +-
...bilityInspectorCorruptSSTableExceptionTest.java | 206 +++++++++++++++
.../io/sstable/format/ForwardingSSTableReader.java | 293 ++++++++++++++++-----
.../org/apache/cassandra/db/DirectoriesTest.java | 3 +-
.../cassandra/utils/JVMStabilityInspectorTest.java | 14 +
14 files changed, 486 insertions(+), 113 deletions(-)
diff --cc CHANGES.txt
index 1acd5a2,73f1a11..4a9ae14
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,12 -1,10 +1,13 @@@
-3.11.8
+4.0-beta2
+ * Forbid altering UDTs used in partition keys (CASSANDRA-15933)
+ * Fix version parsing logic when upgrading from 3.0 (CASSANDRA-15973)
+ * Optimize NoSpamLogger use in hot paths (CASSANDRA-15766)
+ * Verify sstable components on startup (CASSANDRA-15945)
+Merged from 3.11:
++ * stop_paranoid disk failure policy is ignored on CorruptSSTableException after node is up (CASSANDRA-15191)
* Frozen RawTuple is not annotated with frozen in the toString method (CASSANDRA-15857)
Merged from 3.0:
- * stop_paranoid disk failure policy is ignored on CorruptSSTableException after node is up (CASSANDRA-15191)
- * Forbid altering UDTs used in partition keys (CASSANDRA-15933)
* Fix empty/null json string representation (CASSANDRA-15896)
- * 3.x fails to start if commit log has range tombstones from a column which is also deleted (CASSANDRA-15970)
Merged from 2.2:
* Fix CQL parsing of collections when the column type is reversed (CASSANDRA-15814)
diff --cc src/java/org/apache/cassandra/db/Directories.java
index 2ddeb64,b5be69b..889f3a5
--- a/src/java/org/apache/cassandra/db/Directories.java
+++ b/src/java/org/apache/cassandra/db/Directories.java
@@@ -44,9 -44,10 +44,10 @@@ import org.apache.cassandra.io.FSError
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.sstable.*;
+import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.DirectorySizeCalculator;
-import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
+ import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.Pair;
/**
diff --cc src/java/org/apache/cassandra/service/CassandraDaemon.java
index 10d9cea,cf185ec..37cb0b1
--- a/src/java/org/apache/cassandra/service/CassandraDaemon.java
+++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java
@@@ -449,32 -445,6 +449,21 @@@ public class CassandraDaemo
nativeTransportService = new NativeTransportService();
}
+ @VisibleForTesting
+ public static void uncaughtException(Thread t, Throwable e)
+ {
+ StorageMetrics.uncaughtExceptions.inc();
- logger.error("Exception in thread " + t, e);
++ logger.error("Exception in thread {}", t, e);
+ Tracing.trace("Exception in thread {}", t, e);
+ for (Throwable e2 = e; e2 != null; e2 = e2.getCause())
+ {
- JVMStabilityInspector.inspectThrowable(e2);
-
- if (e2 instanceof FSError)
- {
- if (e2 != e) // make sure FSError gets logged exactly once.
- logger.error("Exception in thread " + t, e2);
- FileUtils.handleFSError((FSError) e2);
- }
-
- if (e2 instanceof CorruptSSTableException)
- {
- if (e2 != e)
- logger.error("Exception in thread " + t, e2);
- FileUtils.handleCorruptSSTable((CorruptSSTableException) e2);
- }
++ // make sure error gets logged exactly once.
++ if (e2 != e && (e2 instanceof FSError || e2 instanceof CorruptSSTableException))
++ logger.error("Exception in thread {}", t, e2);
+ }
++ JVMStabilityInspector.inspectThrowable(e);
+ }
+
/*
* Asynchronously load the row and key cache in one off threads and return a compound future of the result.
* Error handling is pushed into the cache load since cache loads are allowed to fail and are handled by logging.
diff --cc src/java/org/apache/cassandra/utils/JVMStabilityInspector.java
index 64403e7,5ac6de4..ae28410
--- a/src/java/org/apache/cassandra/utils/JVMStabilityInspector.java
+++ b/src/java/org/apache/cassandra/utils/JVMStabilityInspector.java
@@@ -56,13 -55,26 +58,31 @@@ public final class JVMStabilityInspecto
* @param t
* The Throwable to check for server-stop conditions
*/
- public static void inspectThrowable(Throwable t)
+ public static void inspectThrowable(Throwable t) throws OutOfMemoryError
+ {
+ inspectThrowable(t, true);
+ }
+
+ public static void inspectThrowable(Throwable t, boolean propagateOutOfMemory) throws OutOfMemoryError
{
- inspectThrowable(t, JVMStabilityInspector::inspectDiskError);
++ inspectThrowable(t, propagateOutOfMemory, JVMStabilityInspector::inspectDiskError);
+ }
+
+ public static void inspectCommitLogThrowable(Throwable t)
+ {
- inspectThrowable(t, JVMStabilityInspector::inspectCommitLogError);
++ inspectThrowable(t, true, JVMStabilityInspector::inspectCommitLogError);
+ }
+
+ private static void inspectDiskError(Throwable t)
+ {
+ if (t instanceof CorruptSSTableException)
+ FileUtils.handleCorruptSSTable((CorruptSSTableException) t);
+ else if (t instanceof FSError)
+ FileUtils.handleFSError((FSError) t);
+ }
+
- public static void inspectThrowable(Throwable t, Consumer<Throwable> fn) throws OutOfMemoryError
++ public static void inspectThrowable(Throwable t, boolean propagateOutOfMemory, Consumer<Throwable> fn) throws OutOfMemoryError
+ {
boolean isUnstable = false;
if (t instanceof OutOfMemoryError)
{
@@@ -103,10 -114,10 +125,10 @@@
killer.killCurrentJVM(t);
if (t.getCause() != null)
- inspectThrowable(t.getCause());
- inspectThrowable(t.getCause(), fn);
++ inspectThrowable(t.getCause(), propagateOutOfMemory, fn);
}
- public static void inspectCommitLogThrowable(Throwable t)
+ private static void inspectCommitLogError(Throwable t)
{
if (!StorageService.instance.isDaemonSetupCompleted())
{
diff --cc src/java/org/apache/cassandra/utils/binlog/BinLog.java
index 7f91761,0000000..f0a24cf
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/utils/binlog/BinLog.java
+++ b/src/java/org/apache/cassandra/utils/binlog/BinLog.java
@@@ -1,479 -1,0 +1,480 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.utils.binlog;
+
+import java.io.File;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import net.openhft.chronicle.queue.ChronicleQueue;
+import net.openhft.chronicle.queue.ChronicleQueueBuilder;
+import net.openhft.chronicle.queue.ExcerptAppender;
+import net.openhft.chronicle.queue.RollCycle;
+import net.openhft.chronicle.queue.RollCycles;
+import net.openhft.chronicle.wire.WireOut;
+import net.openhft.chronicle.wire.WriteMarshallable;
+import org.apache.cassandra.concurrent.NamedThreadFactory;
+import org.apache.cassandra.io.FSError;
+import org.apache.cassandra.io.util.FileUtils;
++import org.apache.cassandra.utils.JVMStabilityInspector;
+import org.apache.cassandra.utils.NoSpamLogger;
+import org.apache.cassandra.utils.Throwables;
+import org.apache.cassandra.utils.concurrent.WeightedQueue;
+
+/**
+ * Bin log is a is quick and dirty binary log that is kind of a NIH version of binary logging with a traditional logging
+ * framework. It's goal is good enough performance, predictable footprint, simplicity in terms of implementation and configuration
+ * and most importantly minimal impact on producers of log records.
+ *
+ * Performance safety is accomplished by feeding items to the binary log using a weighted queue and dropping records if the binary log falls
+ * sufficiently far behind.
+ *
+ * Simplicity and good enough perforamance is achieved by using a single log writing thread as well as Chronicle Queue
+ * to handle writing the log, making it available for readers, as well as log rolling.
+ *
+ */
+public class BinLog implements Runnable
+{
+ private static final Logger logger = LoggerFactory.getLogger(BinLog.class);
+ private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1, TimeUnit.MINUTES);
+ private static final NoSpamLogger.NoSpamLogStatement droppedSamplesStatement = noSpamLogger.getStatement("Dropped {} binary log samples", 1, TimeUnit.MINUTES);
+
+ public final Path path;
+
+ public static final String VERSION = "version";
+ public static final String TYPE = "type";
+
+ private ChronicleQueue queue;
+ private ExcerptAppender appender;
+ @VisibleForTesting
+ Thread binLogThread = new NamedThreadFactory("Binary Log thread").newThread(this);
+ final WeightedQueue<ReleaseableWriteMarshallable> sampleQueue;
+ private final BinLogArchiver archiver;
+ private final boolean blocking;
+
+ private final AtomicLong droppedSamplesSinceLastLog = new AtomicLong();
+
+ /*
+ This set contains all the paths we are currently logging to, it is used to make sure
+ we don't start writing audit and full query logs to the same path.
+ */
+ private static final Set<Path> currentPaths = Collections.synchronizedSet(new HashSet<>());
+
+ private static final ReleaseableWriteMarshallable NO_OP = new ReleaseableWriteMarshallable()
+ {
+ @Override
+ protected long version()
+ {
+ return 0;
+ }
+
+ @Override
+ protected String type()
+ {
+ return "no-op";
+ }
+
+ @Override
+ public void writeMarshallablePayload(WireOut wire)
+ {
+ }
+
+ @Override
+ public void release()
+ {
+ }
+ };
+
+ private volatile boolean shouldContinue = true;
+
+ /**
+ * @param path Path to store the BinLog. Can't be shared with anything else.
+ * @param rollCycle How often to roll the log file so it can potentially be deleted
+ * @param maxQueueWeight Maximum weight of in memory queue for records waiting to be written to the file before blocking or dropping
+ */
+ private BinLog(Path path, RollCycle rollCycle, int maxQueueWeight, BinLogArchiver archiver, boolean blocking)
+ {
+ Preconditions.checkNotNull(path, "path was null");
+ Preconditions.checkNotNull(rollCycle, "rollCycle was null");
+ Preconditions.checkArgument(maxQueueWeight > 0, "maxQueueWeight must be > 0");
+ ChronicleQueueBuilder builder = ChronicleQueueBuilder.single(path.toFile());
+ builder.rollCycle(rollCycle);
+
+ sampleQueue = new WeightedQueue<>(maxQueueWeight);
+ this.archiver = archiver;
+ builder.storeFileListener(this.archiver);
+ queue = builder.build();
+ appender = queue.acquireAppender();
+ this.blocking = blocking;
+ this.path = path;
+ }
+
+ /**
+ * Start the consumer thread that writes log records. Can only be done once.
+ */
+ @VisibleForTesting
+ void start()
+ {
+ if (!shouldContinue)
+ {
+ throw new IllegalStateException("Can't reuse stopped BinLog");
+ }
+ binLogThread.start();
+ }
+
+ /**
+ * Stop the consumer thread that writes log records. Can be called multiple times.
+ * @throws InterruptedException
+ */
+ public synchronized void stop() throws InterruptedException
+ {
+ if (!shouldContinue)
+ {
+ return;
+ }
+
+ shouldContinue = false;
+ sampleQueue.put(NO_OP);
+ binLogThread.join();
+ appender = null;
+ queue = null;
+ archiver.stop();
+ currentPaths.remove(path);
+ }
+
+ /**
+ * Offer a record to the log. If the in memory queue is full the record will be dropped and offer will return false.
+ * @param record The record to write to the log
+ * @return true if the record was queued and false otherwise
+ */
+ public boolean offer(ReleaseableWriteMarshallable record)
+ {
+ if (!shouldContinue)
+ {
+ return false;
+ }
+
+ return sampleQueue.offer(record);
+ }
+
+ /**
+ * Put a record into the log. If the in memory queue is full the putting thread will be blocked until there is space or it is interrupted.
+ * @param record The record to write to the log
+ * @throws InterruptedException
+ */
+ public void put(ReleaseableWriteMarshallable record) throws InterruptedException
+ {
+ if (!shouldContinue)
+ {
+ return;
+ }
+
+ //Resolve potential deadlock at shutdown when queue is full
+ while (shouldContinue)
+ {
+ if (sampleQueue.offer(record, 1, TimeUnit.SECONDS))
+ {
+ return;
+ }
+ }
+ }
+
+ private void processTasks(List<ReleaseableWriteMarshallable> tasks)
+ {
+ for (int ii = 0; ii < tasks.size(); ii++)
+ {
+ WriteMarshallable t = tasks.get(ii);
+ //Don't write an empty document
+ if (t == NO_OP)
+ {
+ continue;
+ }
+
+ appender.writeDocument(t);
+ }
+ }
+
+ @Override
+ public void run()
+ {
+ List<ReleaseableWriteMarshallable> tasks = new ArrayList<>(16);
+ while (shouldContinue)
+ {
+ try
+ {
+ tasks.clear();
+ ReleaseableWriteMarshallable task = sampleQueue.take();
+ tasks.add(task);
+ sampleQueue.drainTo(tasks, 15);
+
+ processTasks(tasks);
+ }
+ catch (Throwable t)
+ {
+ logger.error("Unexpected exception in binary log thread", t);
+ }
+ finally
+ {
+ for (int ii = 0; ii < tasks.size(); ii++)
+ {
+ tasks.get(ii).release();
+ }
+ }
+ }
+
+ //Clean up the buffers on thread exit, finalization will check again once this
+ //is no longer reachable ensuring there are no stragglers in the queue.
+ finalize();
+ }
+
+
+ /**
+ * There is a race where we might not release a buffer, going to let finalization
+ * catch it since it shouldn't happen to a lot of buffers. Only test code would run
+ * into it anyways.
+ */
+ @Override
+ public void finalize()
+ {
+ ReleaseableWriteMarshallable toRelease;
+ while (((toRelease = sampleQueue.poll()) != null))
+ {
+ toRelease.release();
+ }
+ }
+
+ // todo: refactor to helper class?
+ public void logRecord(ReleaseableWriteMarshallable record)
+ {
+ boolean putInQueue = false;
+ try
+ {
+ if (blocking)
+ {
+ try
+ {
+ put(record);
+ putInQueue = true;
+ }
+ catch (InterruptedException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+ else
+ {
+ if (!offer(record))
+ {
+ logDroppedSample();
+ }
+ else
+ {
+ putInQueue = true;
+ }
+ }
+ }
+ finally
+ {
+ if (!putInQueue)
+ {
+ record.release();
+ }
+ }
+ }
+
+ /**
+ * This is potentially lossy, but it's not super critical as we will always generally know
+ * when this is happening and roughly how bad it is.
+ */
+ private void logDroppedSample()
+ {
+ droppedSamplesSinceLastLog.incrementAndGet();
+ if (droppedSamplesStatement.warn(new Object[] {droppedSamplesSinceLastLog.get()}))
+ {
+ droppedSamplesSinceLastLog.set(0);
+ }
+ }
+
+
+ public abstract static class ReleaseableWriteMarshallable implements WriteMarshallable
+ {
+ @Override
+ public final void writeMarshallable(WireOut wire)
+ {
+ wire.write(VERSION).int16(version());
+ wire.write(TYPE).text(type());
+
+ writeMarshallablePayload(wire);
+ }
+
+ protected abstract long version();
+
+ protected abstract String type();
+
+ protected abstract void writeMarshallablePayload(WireOut wire);
+
+ public abstract void release();
+ }
+
+ public static class Builder
+ {
+ private Path path;
+ private String rollCycle;
+ private int maxQueueWeight;
+ private long maxLogSize;
+ private String archiveCommand;
+ private int maxArchiveRetries;
+ private boolean blocking;
+
+ public Builder path(Path path)
+ {
+ Preconditions.checkNotNull(path, "path was null");
+ File pathAsFile = path.toFile();
+ //Exists and is a directory or can be created
+ Preconditions.checkArgument((pathAsFile.exists() && pathAsFile.isDirectory()) || (!pathAsFile.exists() && pathAsFile.mkdirs()), "path exists and is not a directory or couldn't be created");
+ Preconditions.checkArgument(pathAsFile.canRead() && pathAsFile.canWrite() && pathAsFile.canExecute(), "path is not readable, writable, and executable");
+ this.path = path;
+ return this;
+ }
+
+ public Builder rollCycle(String rollCycle)
+ {
+ Preconditions.checkNotNull(rollCycle, "rollCycle was null");
+ rollCycle = rollCycle.toUpperCase();
+ Preconditions.checkNotNull(RollCycles.valueOf(rollCycle), "unrecognized roll cycle");
+ this.rollCycle = rollCycle;
+ return this;
+ }
+
+ public Builder maxQueueWeight(int maxQueueWeight)
+ {
+ Preconditions.checkArgument(maxQueueWeight > 0, "maxQueueWeight must be > 0");
+ this.maxQueueWeight = maxQueueWeight;
+ return this;
+ }
+
+ public Builder maxLogSize(long maxLogSize)
+ {
+ Preconditions.checkArgument(maxLogSize > 0, "maxLogSize must be > 0");
+ this.maxLogSize = maxLogSize;
+ return this;
+ }
+
+ public Builder archiveCommand(String archiveCommand)
+ {
+ this.archiveCommand = archiveCommand;
+ return this;
+ }
+
+ public Builder maxArchiveRetries(int maxArchiveRetries)
+ {
+ this.maxArchiveRetries = maxArchiveRetries;
+ return this;
+ }
+
+ public Builder blocking(boolean blocking)
+ {
+ this.blocking = blocking;
+ return this;
+ }
+
+
+ public BinLog build(boolean cleanDirectory)
+ {
+ logger.info("Attempting to configure bin log: Path: {} Roll cycle: {} Blocking: {} Max queue weight: {} Max log size:{} Archive command: {}", path, rollCycle, blocking, maxQueueWeight, maxLogSize, archiveCommand);
+ synchronized (currentPaths)
+ {
+ if (currentPaths.contains(path))
+ throw new IllegalStateException("Already logging to " + path);
+ currentPaths.add(path);
+ }
+ try
+ {
+ // create the archiver before cleaning directories - ExternalArchiver will try to archive any existing file.
+ BinLogArchiver archiver = Strings.isNullOrEmpty(archiveCommand) ? new DeletingArchiver(maxLogSize) : new ExternalArchiver(archiveCommand, path, maxArchiveRetries);
+ if (cleanDirectory)
+ {
+ logger.info("Cleaning directory: {} as requested", path);
+ if (path.toFile().exists())
+ {
+ Throwable error = cleanDirectory(path.toFile(), null);
+ if (error != null)
+ {
+ throw new RuntimeException(error);
+ }
+ }
+ }
+ BinLog binlog = new BinLog(path, RollCycles.valueOf(rollCycle), maxQueueWeight, archiver, blocking);
+ binlog.start();
+ return binlog;
+ }
+ catch (Exception e)
+ {
+ currentPaths.remove(path);
+ throw e;
+ }
+ }
+ }
+
+ public static Throwable cleanDirectory(File directory, Throwable accumulate)
+ {
+ if (!directory.exists())
+ {
+ return Throwables.merge(accumulate, new RuntimeException(String.format("%s does not exists", directory)));
+ }
+ if (!directory.isDirectory())
+ {
+ return Throwables.merge(accumulate, new RuntimeException(String.format("%s is not a directory", directory)));
+ }
+ for (File f : directory.listFiles())
+ {
+ accumulate = deleteRecursively(f, accumulate);
+ }
+ if (accumulate instanceof FSError)
+ {
- FileUtils.handleFSError((FSError)accumulate);
++ JVMStabilityInspector.inspectThrowable(accumulate);
+ }
+ return accumulate;
+ }
+
+ private static Throwable deleteRecursively(File fileOrDirectory, Throwable accumulate)
+ {
+ if (fileOrDirectory.isDirectory())
+ {
+ for (File f : fileOrDirectory.listFiles())
+ {
+ accumulate = FileUtils.deleteWithConfirm(f, accumulate);
+ }
+ }
+ return FileUtils.deleteWithConfirm(fileOrDirectory, accumulate);
+ }
+}
diff --cc test/distributed/org/apache/cassandra/distributed/test/JVMStabilityInspectorCorruptSSTableExceptionTest.java
index 0000000,b9e8d67..c7abfc5
mode 000000,100644..100644
--- a/test/distributed/org/apache/cassandra/distributed/test/JVMStabilityInspectorCorruptSSTableExceptionTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/JVMStabilityInspectorCorruptSSTableExceptionTest.java
@@@ -1,0 -1,204 +1,206 @@@
+ /*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+ package org.apache.cassandra.distributed.test;
+
+ import java.io.IOException;
+ import java.util.HashSet;
+ import java.util.Set;
+
+ import org.junit.Assert;
+ import org.junit.Test;
+
+ import org.apache.cassandra.config.Config.DiskFailurePolicy;
+ import org.apache.cassandra.db.ColumnFamilyStore;
+ import org.apache.cassandra.db.DecoratedKey;
+ import org.apache.cassandra.db.Keyspace;
+ import org.apache.cassandra.db.RowIndexEntry;
+ import org.apache.cassandra.db.Slices;
+ import org.apache.cassandra.db.filter.ColumnFilter;
+ import org.apache.cassandra.db.rows.UnfilteredRowIterator;
+ import org.apache.cassandra.distributed.Cluster;
+ import org.apache.cassandra.distributed.api.ConsistencyLevel;
+ import org.apache.cassandra.distributed.api.IInvokableInstance;
+ import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableCallable;
+ import org.apache.cassandra.distributed.shared.AbstractBuilder;
+ import org.apache.cassandra.distributed.shared.NetworkTopology;
+ import org.apache.cassandra.gms.Gossiper;
+ import org.apache.cassandra.io.sstable.CorruptSSTableException;
+ import org.apache.cassandra.io.sstable.format.ForwardingSSTableReader;
+ import org.apache.cassandra.io.sstable.format.SSTableReader;
+ import org.apache.cassandra.io.sstable.format.SSTableReadsListener;
+ import org.apache.cassandra.io.util.FileDataInput;
+ import org.apache.cassandra.service.CassandraDaemon;
+ import org.apache.cassandra.service.StorageService;
+
+ import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+ import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL;
+ import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+
+ public class JVMStabilityInspectorCorruptSSTableExceptionTest extends TestBaseImpl
+ {
+ @Test
+ public void testAbstractLocalAwareExecutorServiceOnIgnoredDiskFailurePolicy() throws Exception
+ {
+ test(DiskFailurePolicy.ignore, true, true);
+ }
+
+ @Test
+ public void testAbstractLocalAwareExecutorServiceOnStopParanoidDiskFailurePolicy() throws Exception
+ {
+ test(DiskFailurePolicy.stop_paranoid, false, false);
+ }
+
+ private static void test(DiskFailurePolicy policy, boolean expectNativeTransportRunning, boolean expectGossiperEnabled) throws Exception
+ {
+ String table = policy.name();
+ try (final Cluster cluster = init(getCluster(policy).start()))
+ {
+ IInvokableInstance node = cluster.get(1);
+ boolean[] setup = node.callOnInstance(() -> {
+ CassandraDaemon instanceForTesting = CassandraDaemon.getInstanceForTesting();
+ instanceForTesting.completeSetup();
+ StorageService.instance.registerDaemon(instanceForTesting);
+ return new boolean[]{ StorageService.instance.isNativeTransportRunning(), Gossiper.instance.isEnabled() };
+ });
+
+ // make sure environment is setup propertly
+ Assert.assertTrue("Native support is not running, test is not ready!", setup[0]);
+ Assert.assertTrue("Gossiper is not running, test is not ready!", setup[1]);
+
+ cluster.schemaChange("CREATE TABLE " + KEYSPACE + "." + table + " (id bigint PRIMARY KEY)");
+ node.executeInternal("INSERT INTO " + KEYSPACE + "." + table + " (id) VALUES (?)", 0L);
+ corruptTable(node, KEYSPACE, table);
+
+ try
+ {
+ cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + '.' + table + " WHERE id=?", ConsistencyLevel.ONE, 0L);
+ Assert.fail("Select should fail as we corrupted SSTable on purpose.");
+ }
+ catch (final Exception ex)
+ {
+ // we expect that above query fails as we corrupted an sstable
+ }
+
+ waitForStop(!expectGossiperEnabled, node, new SerializableCallable<Boolean>()
+ {
+ public Boolean call()
+ {
+ return Gossiper.instance.isEnabled();
+ }
+ });
+
+ waitForStop(!expectNativeTransportRunning, node, new SerializableCallable<Boolean>()
+ {
+ public Boolean call()
+ {
+ return StorageService.instance.isNativeTransportRunning();
+ }
+ });
+ }
+ }
+
+ private static void waitForStop(boolean shouldWaitForStop,
+ IInvokableInstance node,
+ SerializableCallable<Boolean> serializableCallable) throws Exception
+ {
+ int attempts = 3;
+ boolean running = true;
+
+ while (attempts > 0 && running)
+ {
+ try
+ {
+ running = node.callOnInstance(serializableCallable);
+ attempts--;
+ }
+ catch (final NoClassDefFoundError ex)
+ {
+ // gossiper throws this
+ Assert.assertEquals("Could not initialize class org.apache.cassandra.service.StorageService", ex.getMessage());
+ running = false;
+ }
+ catch (final ExceptionInInitializerError ex)
+ {
+ // native thows this, ignore on purpose, this means that native transport is closed.
+ running = false;
+ }
+
+ Thread.sleep(5000);
+ }
+
+ if (shouldWaitForStop && running)
+ {
+ Assert.fail("we did want a service to stop, but it did not.");
+ }
+
+ if (!shouldWaitForStop && !running)
+ {
+ Assert.fail("we did not want a service to stop, but it did.");
+ }
+ }
+
+ private static void corruptTable(IInvokableInstance node, String keyspace, String table)
+ {
+ node.runOnInstance(() -> {
+ ColumnFamilyStore cf = Keyspace.open(keyspace).getColumnFamilyStore(table);
+ cf.forceBlockingFlush();
+
+ Set<SSTableReader> remove = cf.getLiveSSTables();
+ Set<SSTableReader> replace = new HashSet<>();
+ for (SSTableReader r : remove)
+ replace.add(new CorruptedSSTableReader(r));
+
+ cf.getTracker().removeUnsafe(remove);
+ cf.addSSTables(replace);
+ });
+ }
+
+ private static AbstractBuilder<IInvokableInstance, Cluster, Cluster.Builder> getCluster(DiskFailurePolicy diskFailurePolicy)
+ {
+ return Cluster.build()
+ .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(1, "dc0", "rack0"))
+ .withConfig(config -> config.with(NETWORK, GOSSIP, NATIVE_PROTOCOL)
+ .set("disk_failure_policy", diskFailurePolicy.name()));
+ }
+
+ private static final class CorruptedSSTableReader extends ForwardingSSTableReader
+ {
+ public CorruptedSSTableReader(SSTableReader delegate)
+ {
+ super(delegate);
+ }
+
- public UnfilteredRowIterator iterator(DecoratedKey key, Slices slices, ColumnFilter selectedColumns, boolean reversed, boolean isForThrift, SSTableReadsListener listener)
++ @Override
++ public UnfilteredRowIterator iterator(DecoratedKey key, Slices slices, ColumnFilter selectedColumns, boolean reversed, SSTableReadsListener listener)
+ {
+ throw throwCorrupted();
+ }
+
- public UnfilteredRowIterator iterator(FileDataInput file, DecoratedKey key, RowIndexEntry indexEntry, Slices slices, ColumnFilter selectedColumns, boolean reversed, boolean isForThrift)
++ @Override
++ public UnfilteredRowIterator iterator(FileDataInput file, DecoratedKey key, RowIndexEntry indexEntry, Slices slices, ColumnFilter selectedColumns, boolean reversed)
+ {
+ throw throwCorrupted();
+ }
+
+ private CorruptSSTableException throwCorrupted()
+ {
+ throw new CorruptSSTableException(new IOException("failed to get position"), descriptor.baseFilename());
+ }
+ }
-}
++}
diff --cc test/distributed/org/apache/cassandra/io/sstable/format/ForwardingSSTableReader.java
index 40adc9c,ddf9487..7e21253
--- a/test/distributed/org/apache/cassandra/io/sstable/format/ForwardingSSTableReader.java
+++ b/test/distributed/org/apache/cassandra/io/sstable/format/ForwardingSSTableReader.java
@@@ -137,16 -108,19 +110,13 @@@ public abstract class ForwardingSSTable
return delegate.getFilename();
}
- public boolean equals(Object that)
- {
- return delegate.equals(that);
- }
-
- public int hashCode()
+ @Override
+ public void setupOnline()
{
- return delegate.hashCode();
+ delegate.setupOnline();
}
+ @Override
- public boolean isKeyCacheSetup()
- {
- return delegate.isKeyCacheSetup();
- }
-
- @Override
public boolean loadSummary()
{
return delegate.loadSummary();
@@@ -287,7 -288,8 +284,8 @@@
return delegate.getKeySamples(range);
}
+ @Override
- public List<Pair<Long, Long>> getPositionsForRanges(Collection<Range<Token>> ranges)
+ public List<PartitionPositionBounds> getPositionsForRanges(Collection<Range<Token>> ranges)
{
return delegate.getPositionsForRanges(ranges);
}
@@@ -312,11 -318,31 +314,37 @@@
return delegate.getCachedPosition(unifiedKey, updateStats);
}
+ @Override
+ public boolean isKeyCacheEnabled()
+ {
+ return delegate.isKeyCacheEnabled();
+ }
+
++ @Override
+ protected RowIndexEntry getPosition(PartitionPosition key, Operator op, boolean updateCacheAndStats, boolean permitMatchPastLast, SSTableReadsListener listener)
+ {
+ return delegate.getPosition(key, op, updateCacheAndStats, permitMatchPastLast, listener);
+ }
+
+ @Override
- public UnfilteredRowIterator iterator(DecoratedKey key, Slices slices, ColumnFilter selectedColumns, boolean reversed, boolean isForThrift, SSTableReadsListener listener)
++ public UnfilteredRowIterator iterator(DecoratedKey key, Slices slices, ColumnFilter selectedColumns, boolean reversed, SSTableReadsListener listener)
+ {
- return delegate.iterator(key, slices, selectedColumns, reversed, isForThrift, listener);
++ return delegate.iterator(key, slices, selectedColumns, reversed, listener);
+ }
+
+ @Override
- public UnfilteredRowIterator iterator(FileDataInput file, DecoratedKey key, RowIndexEntry indexEntry, Slices slices, ColumnFilter selectedColumns, boolean reversed, boolean isForThrift)
++ public UnfilteredRowIterator iterator(FileDataInput file, DecoratedKey key, RowIndexEntry indexEntry, Slices slices, ColumnFilter selectedColumns, boolean reversed)
+ {
- return delegate.iterator(file, key, indexEntry, slices, selectedColumns, reversed, isForThrift);
++ return delegate.iterator(file, key, indexEntry, slices, selectedColumns, reversed);
+ }
+
+ @Override
+ public UnfilteredRowIterator simpleIterator(FileDataInput file, DecoratedKey key, RowIndexEntry indexEntry, boolean tombstoneOnly)
+ {
+ return delegate.simpleIterator(file, key, indexEntry, tombstoneOnly);
+ }
+
+ @Override
public DecoratedKey firstKeyBeyond(PartitionPosition token)
{
return delegate.firstKeyBeyond(token);
@@@ -357,21 -390,55 +392,49 @@@
delegate.markSuspect();
}
+ @Override
- public boolean isMarkedSuspect()
- {
- return delegate.isMarkedSuspect();
- }
-
- @Override
- public ISSTableScanner getScanner()
+ public void unmarkSuspect()
{
- return delegate.getScanner();
+ delegate.unmarkSuspect();
}
+ @Override
- public ISSTableScanner getScanner(ColumnFilter columns, DataRange dataRange, boolean isForThrift, SSTableReadsListener listener)
+ public boolean isMarkedSuspect()
{
- return delegate.getScanner(columns, dataRange, isForThrift, listener);
+ return delegate.isMarkedSuspect();
}
+ @Override
- public ISSTableScanner getScanner(Range<Token> range, RateLimiter limiter)
+ public ISSTableScanner getScanner(Range<Token> range)
{
- return delegate.getScanner(range, limiter);
+ return delegate.getScanner(range);
}
+ @Override
- public ISSTableScanner getScanner(RateLimiter limiter)
++ public ISSTableScanner getScanner()
+ {
- return delegate.getScanner(limiter);
++ return delegate.getScanner();
+ }
+
+ @Override
- public ISSTableScanner getScanner(Collection<Range<Token>> ranges, RateLimiter limiter)
++ public ISSTableScanner getScanner(Collection<Range<Token>> ranges)
+ {
- return delegate.getScanner(ranges, limiter);
++ return delegate.getScanner(ranges);
+ }
+
+ @Override
+ public ISSTableScanner getScanner(Iterator<AbstractBounds<PartitionPosition>> rangeIterator)
+ {
+ return delegate.getScanner(rangeIterator);
+ }
+
+ @Override
- public ISSTableScanner getScanner(ColumnFilter columns, DataRange dataRange, RateLimiter limiter, boolean isForThrift, SSTableReadsListener listener)
++ public ISSTableScanner getScanner(ColumnFilter columns, DataRange dataRange, SSTableReadsListener listener)
+ {
- return delegate.getScanner(columns, dataRange, limiter, isForThrift, listener);
++ return delegate.getScanner(columns, dataRange, listener);
+ }
+
+ @Override
public FileDataInput getFileDataInput(long position)
{
return delegate.getFileDataInput(position);
@@@ -397,31 -468,7 +464,37 @@@
return delegate.keyAt(indexPosition);
}
+ @Override
+ public boolean isPendingRepair()
+ {
+ return delegate.isPendingRepair();
+ }
+
++ @Override
+ public UUID getPendingRepair()
+ {
+ return delegate.getPendingRepair();
+ }
+
++ @Override
+ public long getRepairedAt()
+ {
+ return delegate.getRepairedAt();
+ }
+
++ @Override
+ public boolean isTransient()
+ {
+ return delegate.isTransient();
+ }
+
++ @Override
+ public boolean intersects(Collection<Range<Token>> ranges)
+ {
+ return delegate.intersects(ranges);
+ }
+
++ @Override
public long getBloomFilterFalsePositiveCount()
{
return delegate.getBloomFilterFalsePositiveCount();
@@@ -452,11 -504,13 +530,13 @@@
return delegate.getEstimatedPartitionSize();
}
+ @Override
- public EstimatedHistogram getEstimatedColumnCount()
+ public EstimatedHistogram getEstimatedCellPerPartitionCount()
{
- return delegate.getEstimatedColumnCount();
+ return delegate.getEstimatedCellPerPartitionCount();
}
+ @Override
public double getEstimatedDroppableTombstoneRatio(int gcBefore)
{
return delegate.getEstimatedDroppableTombstoneRatio(gcBefore);
@@@ -617,11 -708,7 +734,13 @@@
delegate.addTo(identities);
}
+ @Override
+ public TableMetadata metadata()
+ {
+ return delegate.metadata();
+ }
+
++ @Override
public IPartitioner getPartitioner()
{
return delegate.getPartitioner();
@@@ -667,13 -780,9 +812,15 @@@
return delegate.toString();
}
- public AbstractBounds<Token> getBounds()
+ @Override
+ public void addComponents(Collection<Component> newComponents)
{
- return delegate.getBounds();
+ delegate.addComponents(newComponents);
}
-}
+
- public ChannelProxy getDataChannel()
++ @Override
++ public AbstractBounds<Token> getBounds()
+ {
- return delegate.getDataChannel();
++ return delegate.getBounds();
+ }
+}
diff --cc test/unit/org/apache/cassandra/db/DirectoriesTest.java
index dd4e51f,9afe492..eb2016f
--- a/test/unit/org/apache/cassandra/db/DirectoriesTest.java
+++ b/test/unit/org/apache/cassandra/db/DirectoriesTest.java
@@@ -51,6 -50,8 +51,7 @@@ import org.apache.cassandra.io.util.Fil
import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.service.DefaultFSErrorHandler;
import org.apache.cassandra.utils.ByteBufferUtil;
+ import org.apache.cassandra.utils.JVMStabilityInspector;
-import org.apache.cassandra.utils.Pair;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org