You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by st...@apache.org on 2016/10/07 08:57:49 UTC
[07/10] cassandra git commit: Merge branch 'cassandra-2.2' into
cassandra-3.0
Merge branch 'cassandra-2.2' into cassandra-3.0
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/695065e2
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/695065e2
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/695065e2
Branch: refs/heads/trunk
Commit: 695065e27a16c30019f34fc4c626a1841616d037
Parents: 45d0176 be6e6ea
Author: Stefania Alborghetti <st...@datastax.com>
Authored: Fri Oct 7 16:51:10 2016 +0800
Committer: Stefania Alborghetti <st...@datastax.com>
Committed: Fri Oct 7 16:52:01 2016 +0800
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../DebuggableScheduledThreadPoolExecutor.java | 2 +-
.../concurrent/ScheduledExecutors.java | 4 -
.../db/compaction/CompactionManager.java | 127 +++++++++++--------
.../db/lifecycle/LifecycleTransaction.java | 14 +-
.../io/sstable/format/SSTableReader.java | 15 ++-
.../apache/cassandra/net/MessagingService.java | 3 +-
.../cassandra/service/StorageService.java | 12 +-
.../org/apache/cassandra/utils/ExpiringMap.java | 4 +-
9 files changed, 116 insertions(+), 66 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/695065e2/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 827a208,54425fa..894113a
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,19 -1,10 +1,20 @@@
-2.2.9
+3.0.10
+ * Make stress use client mode to avoid checking commit log size on startup (CASSANDRA-12478)
+ * Fix exceptions with new vnode allocation (CASSANDRA-12715)
+ * Unify drain and shutdown processes (CASSANDRA-12509)
+ * Fix NPE in ComponentOfSlice.isEQ() (CASSANDRA-12706)
+ * Fix failure in LogTransactionTest (CASSANDRA-12632)
+ * Fix potentially incomplete non-frozen UDT values when querying with the
+ full primary key specified (CASSANDRA-12605)
+ * Skip writing MV mutations to commitlog on mutation.applyUnsafe() (CASSANDRA-11670)
+ * Establish consistent distinction between non-existing partition and NULL value for LWTs on static columns (CASSANDRA-12060)
+ * Extend ColumnIdentifier.internedInstances key to include the type that generated the byte buffer (CASSANDRA-12516)
+ * Backport CASSANDRA-10756 (race condition in NativeTransportService shutdown) (CASSANDRA-12472)
+ * If CF has no clustering columns, any row cache is full partition cache (CASSANDRA-12499)
+Merged from 2.2:
+ * Fix leak errors and execution rejected exceptions when draining (CASSANDRA-12457)
* Fix merkle tree depth calculation (CASSANDRA-12580)
* Make Collections deserialization more robust (CASSANDRA-12618)
-
-
-2.2.8
* Fix exceptions when enabling gossip on nodes that haven't joined the ring (CASSANDRA-12253)
* Fix authentication problem when invoking clqsh copy from a SOURCE command (CASSANDRA-12642)
* Decrement pending range calculator jobs counter in finally block
http://git-wip-us.apache.org/repos/asf/cassandra/blob/695065e2/src/java/org/apache/cassandra/concurrent/DebuggableScheduledThreadPoolExecutor.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/695065e2/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index 4d1757e,626bd27..478b896
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@@ -171,17 -164,15 +171,15 @@@ public class CompactionManager implemen
logger.trace("Scheduling a background task check for {}.{} with {}",
cfs.keyspace.getName(),
cfs.name,
- cfs.getCompactionStrategy().getName());
+ cfs.getCompactionStrategyManager().getName());
- List<Future<?>> futures = new ArrayList<>();
- // we must schedule it at least once, otherwise compaction will stop for a CF until next flush
- if (executor.isShutdown())
+
+ List<Future<?>> futures = new ArrayList<>(1);
+ Future<?> fut = executor.submitIfRunning(new BackgroundCompactionCandidate(cfs), "background task");
+ if (!fut.isCancelled())
{
- logger.info("Executor has shut down, not submitting background task");
- return Collections.emptyList();
+ compactingCF.add(cfs);
+ futures.add(fut);
}
- compactingCF.add(cfs);
- futures.add(executor.submit(new BackgroundCompactionCandidate(cfs)));
-
return futures;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/695065e2/src/java/org/apache/cassandra/db/lifecycle/LifecycleTransaction.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/lifecycle/LifecycleTransaction.java
index 91515aa,a95c4a8..582c9d8
--- a/src/java/org/apache/cassandra/db/lifecycle/LifecycleTransaction.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/LifecycleTransaction.java
@@@ -200,14 -155,10 +206,16 @@@ public class LifecycleTransaction exten
public Throwable doCommit(Throwable accumulate)
{
assert staged.isEmpty() : "must be no actions introduced between prepareToCommit and a commit";
- logger.trace("Committing update:{}, obsolete:{}", staged.update, staged.obsolete);
+
+ if (logger.isTraceEnabled())
+ logger.trace("Committing transaction over {} staged: {}, logged: {}", originals, staged, logged);
+ // accumulate must be null if we have been used correctly, so fail immediately if it is not
+ maybeFail(accumulate);
+
+ // transaction log commit failure means we must abort; safe commit is not possible
+ maybeFail(log.commit(null));
+
// this is now the point of no return; we cannot safely rollback, so we ignore exceptions until we're done
// we restore state by obsoleting our obsolete files, releasing our references to them, and updating our size
// and notification status for the obsolete and new files
@@@ -226,12 -175,10 +234,12 @@@
public Throwable doAbort(Throwable accumulate)
{
if (logger.isTraceEnabled())
- logger.trace("Aborting transaction over {}, with ({},{}) logged and ({},{}) staged", originals, logged.update, logged.obsolete, staged.update, staged.obsolete);
+ logger.trace("Aborting transaction over {} staged: {}, logged: {}", originals, staged, logged);
+ accumulate = abortObsoletion(obsoletions, accumulate);
+
if (logged.isEmpty() && staged.isEmpty())
- return accumulate;
+ return log.abort(accumulate);
// mark obsolete all readers that are not versions of those present in the original set
Iterable<SSTableReader> obsolete = filterOut(concatUniq(staged.update, logged.update), originals);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/695065e2/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
index 9f2663e,fddf058..9f31af1
--- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
@@@ -2049,12 -2028,16 +2049,12 @@@ public abstract class SSTableReader ext
private Runnable runOnClose;
private boolean isReplaced = false;
- // a reference to our shared per-Descriptor.Type tidy instance, that
+ // a reference to our shared tidy instance, that
// we will release when we are ourselves released
- private Ref<DescriptorTypeTidy> typeRef;
-
- // a convenience stashing of the shared per-descriptor-type tidy instance itself
- // and the per-logical-sstable globally shared state that it is linked to
- private DescriptorTypeTidy type;
+ private Ref<GlobalTidy> globalRef;
private GlobalTidy global;
- private boolean setup;
+ private volatile boolean setup;
void setup(SSTableReader reader, boolean trackHotness)
{
@@@ -2108,7 -2102,10 +2118,10 @@@
dfile.close();
if (ifile != null)
ifile.close();
- typeRef.release();
+ globalRef.release();
+
+ if (logger.isTraceEnabled())
+ logger.trace("Async instance tidier for {}, completed", descriptor);
}
});
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/695065e2/src/java/org/apache/cassandra/net/MessagingService.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/695065e2/src/java/org/apache/cassandra/service/StorageService.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/service/StorageService.java
index 0be5d92,db86294..24ad73d
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@@ -4013,143 -3933,88 +4013,149 @@@ public class StorageService extends Not
*/
public synchronized void drain() throws IOException, InterruptedException, ExecutionException
{
- inShutdownHook = true;
+ drain(false);
+ }
+ protected synchronized void drain(boolean isFinalShutdown) throws IOException, InterruptedException, ExecutionException
+ {
ExecutorService counterMutationStage = StageManager.getStage(Stage.COUNTER_MUTATION);
+ ExecutorService viewMutationStage = StageManager.getStage(Stage.VIEW_MUTATION);
ExecutorService mutationStage = StageManager.getStage(Stage.MUTATION);
- if (mutationStage.isTerminated() && counterMutationStage.isTerminated())
+
+ if (mutationStage.isTerminated()
+ && counterMutationStage.isTerminated()
+ && viewMutationStage.isTerminated())
{
- logger.warn("Cannot drain node (did it already happen?)");
+ if (!isFinalShutdown)
+ logger.warn("Cannot drain node (did it already happen?)");
return;
}
- setMode(Mode.DRAINING, "starting drain process", true);
- shutdownClientServers();
- ScheduledExecutors.optionalTasks.shutdown();
- Gossiper.instance.stop();
- setMode(Mode.DRAINING, "shutting down MessageService", false);
- MessagingService.instance().shutdown();
+ assert !isShutdown;
+ isShutdown = true;
- setMode(Mode.DRAINING, "clearing mutation stage", false);
- counterMutationStage.shutdown();
- mutationStage.shutdown();
- counterMutationStage.awaitTermination(3600, TimeUnit.SECONDS);
- mutationStage.awaitTermination(3600, TimeUnit.SECONDS);
+ try
+ {
+ setMode(Mode.DRAINING, "starting drain process", !isFinalShutdown);
- StorageProxy.instance.verifyNoHintsInProgress();
+ BatchlogManager.instance.shutdown();
+ HintsService.instance.pauseDispatch();
- setMode(Mode.DRAINING, "flushing column families", false);
- // count CFs first, since forceFlush could block for the flushWriter to get a queue slot empty
- totalCFs = 0;
- for (Keyspace keyspace : Keyspace.nonSystem())
- totalCFs += keyspace.getColumnFamilyStores().size();
- remainingCFs = totalCFs;
- // flush
- List<Future<?>> flushes = new ArrayList<>();
- for (Keyspace keyspace : Keyspace.nonSystem())
- {
- for (ColumnFamilyStore cfs : keyspace.getColumnFamilyStores())
- flushes.add(cfs.forceFlush());
- }
- // wait for the flushes.
- // TODO this is a godawful way to track progress, since they flush in parallel. a long one could
- // thus make several short ones "instant" if we wait for them later.
- for (Future f : flushes)
- {
- FBUtilities.waitOnFuture(f);
- remainingCFs--;
- }
+ if (daemon != null)
+ shutdownClientServers();
+ ScheduledExecutors.optionalTasks.shutdown();
+ Gossiper.instance.stop();
- BatchlogManager.shutdown();
+ if (!isFinalShutdown)
+ setMode(Mode.DRAINING, "shutting down MessageService", false);
+
+ // In-progress writes originating here could generate hints to be written, so shut down MessagingService
+ // before mutation stage, so we can get all the hints saved before shutting down
+ MessagingService.instance().shutdown();
+
+ if (!isFinalShutdown)
+ setMode(Mode.DRAINING, "clearing mutation stage", false);
+ viewMutationStage.shutdown();
+ counterMutationStage.shutdown();
+ mutationStage.shutdown();
+ viewMutationStage.awaitTermination(3600, TimeUnit.SECONDS);
+ counterMutationStage.awaitTermination(3600, TimeUnit.SECONDS);
+ mutationStage.awaitTermination(3600, TimeUnit.SECONDS);
+
+ StorageProxy.instance.verifyNoHintsInProgress();
+
+ if (!isFinalShutdown)
+ setMode(Mode.DRAINING, "flushing column families", false);
+
+ // disable autocompaction - we don't want to start any new compactions while we are draining
+ for (Keyspace keyspace : Keyspace.all())
+ for (ColumnFamilyStore cfs : keyspace.getColumnFamilyStores())
+ cfs.disableAutoCompaction();
+
+ // count CFs first, since forceFlush could block for the flushWriter to get a queue slot empty
+ totalCFs = 0;
+ for (Keyspace keyspace : Keyspace.nonSystem())
+ totalCFs += keyspace.getColumnFamilyStores().size();
+ remainingCFs = totalCFs;
+ // flush
+ List<Future<?>> flushes = new ArrayList<>();
+ for (Keyspace keyspace : Keyspace.nonSystem())
+ {
+ for (ColumnFamilyStore cfs : keyspace.getColumnFamilyStores())
+ flushes.add(cfs.forceFlush());
+ }
+ // wait for the flushes.
+ // TODO this is a godawful way to track progress, since they flush in parallel. a long one could
+ // thus make several short ones "instant" if we wait for them later.
+ for (Future f : flushes)
+ {
+ try
+ {
+ FBUtilities.waitOnFuture(f);
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ // don't let this stop us from shutting down the commitlog and other thread pools
+ logger.warn("Caught exception while waiting for memtable flushes during shutdown hook", t);
+ }
- // Interrupt on going compaction and shutdown to prevent further compaction
- CompactionManager.instance.forceShutdown();
+ remainingCFs--;
+ }
- // flush the system ones after all the rest are done, just in case flushing modifies any system state
- // like CASSANDRA-5151. don't bother with progress tracking since system data is tiny.
+
- // Flush the system tables after all other tables are flushed, just in case flushing modifies any system state
- // like CASSANDRA-5151. Don't bother with progress tracking since system data is tiny.
- // Flush system tables after stopping the batchlog manager and compactions since they both modify
- // system tables (for example compactions can obsolete sstables and the tidiers in SSTableReader update
- // system tables, see SSTableReader.GlobalTidy)
- flushes.clear();
- for (Keyspace keyspace : Keyspace.system())
- {
- for (ColumnFamilyStore cfs : keyspace.getColumnFamilyStores())
- flushes.add(cfs.forceFlush());
- }
- FBUtilities.waitOnFutures(flushes);
++ // Interrupt ongoing compactions and shutdown CM to prevent further compactions.
++ CompactionManager.instance.forceShutdown();
++ // Flush the system tables after all other tables are flushed, just in case flushing modifies any system state
++ // like CASSANDRA-5151. Don't bother with progress tracking since system data is tiny.
++ // Flush system tables after stopping compactions since they modify
++ // system tables (for example compactions can obsolete sstables and the tidiers in SSTableReader update
++ // system tables, see SSTableReader.GlobalTidy)
+ flushes.clear();
+ for (Keyspace keyspace : Keyspace.system())
+ {
+ for (ColumnFamilyStore cfs : keyspace.getColumnFamilyStores())
+ flushes.add(cfs.forceFlush());
+ }
+ FBUtilities.waitOnFutures(flushes);
+
+ HintsService.instance.shutdownBlocking();
- // whilst we've flushed all the CFs, which will have recycled all completed segments, we want to ensure
- // there are no segments to replay, so we force the recycling of any remaining (should be at most one)
- CommitLog.instance.forceRecycleAllSegments();
+ // Interrupt ongoing compactions and shutdown CM to prevent further compactions.
+ CompactionManager.instance.forceShutdown();
- CommitLog.instance.shutdownBlocking();
+ // whilst we've flushed all the CFs, which will have recycled all completed segments, we want to ensure
+ // there are no segments to replay, so we force the recycling of any remaining (should be at most one)
+ CommitLog.instance.forceRecycleAllSegments();
- // wait for miscellaneous tasks like sstable and commitlog segment deletion
- ScheduledExecutors.nonPeriodicTasks.shutdown();
- if (!ScheduledExecutors.nonPeriodicTasks.awaitTermination(1, TimeUnit.MINUTES))
- logger.warn("Failed to wait for non periodic tasks to shutdown");
+ CommitLog.instance.shutdownBlocking();
- ColumnFamilyStore.shutdownPostFlushExecutor();
+ // wait for miscellaneous tasks like sstable and commitlog segment deletion
+ ScheduledExecutors.nonPeriodicTasks.shutdown();
+ if (!ScheduledExecutors.nonPeriodicTasks.awaitTermination(1, TimeUnit.MINUTES))
- logger.warn("Miscellaneous task executor still busy after one minute; proceeding with shutdown");
++ logger.warn("Failed to wait for non periodic tasks to shutdown");
- setMode(Mode.DRAINED, true);
+ ColumnFamilyStore.shutdownPostFlushExecutor();
+ setMode(Mode.DRAINED, !isFinalShutdown);
+ }
+ catch (Throwable t)
+ {
+ logger.error("Caught an exception while draining ", t);
+ }
+ }
+
+ /**
+ * Some services are shutdown during draining and we should not attempt to start them again.
+ *
+ * @param service - the name of the service we are trying to start.
+ * @throws IllegalStateException - an exception that nodetool is able to convert into a message to display to the user
+ */
+ synchronized void checkServiceAllowedToStart(String service)
+ {
+ if (isDraining()) // when draining isShutdown is also true, so we check first to return a more accurate message
+ throw new IllegalStateException(String.format("Unable to start %s because the node is draining.", service));
+
+ if (isShutdown()) // do not rely on operationMode in case it gets changed to decomissioned or other
+ throw new IllegalStateException(String.format("Unable to start %s because the node was drained.", service));
}
// Never ever do this at home. Used by tests.