cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jmcken...@apache.org
Subject [01/15] cassandra git commit: Fix out-of-space error treatment in memtable flushing
Date Fri, 01 Apr 2016 15:55:38 GMT
Repository: cassandra
Updated Branches:
  refs/heads/cassandra-2.1 105fbb35d -> f3b3c410a
  refs/heads/cassandra-2.2 106750c78 -> caaa9fc82
  refs/heads/cassandra-3.0 be3808140 -> bd4cab24b
  refs/heads/cassandra-3.5 4111e4fdf -> af9b9cd5a
  refs/heads/trunk bdae4520c -> 6d1254ac2


Fix out-of-space error treatment in memtable flushing

Patch by Branimir Lambov; reviewed by Joshua McKenzie for
CASSANDRA-11448


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/f3b3c410
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/f3b3c410
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/f3b3c410

Branch: refs/heads/cassandra-2.1
Commit: f3b3c410a0d84a4348cf05954b38df6b087762a7
Parents: 105fbb3
Author: Branimir Lambov <branimir.lambov@datastax.com>
Authored: Wed Mar 30 16:29:55 2016 +0300
Committer: Josh McKenzie <josh.mckenzie@datastax.com>
Committed: Fri Apr 1 11:45:18 2016 -0400

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../apache/cassandra/db/ColumnFamilyStore.java  |  24 ++-
 .../org/apache/cassandra/db/Directories.java    |   2 +-
 .../db/commitlog/CommitLogSegmentManager.java   |   4 +-
 .../cassandra/io/util/DiskAwareRunnable.java    |   5 +-
 .../apache/cassandra/cql3/OutOfSpaceTest.java   | 157 +++++++++++++++++++
 6 files changed, 185 insertions(+), 8 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/f3b3c410/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index af2518c..50bc894 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 2.1.14
+ * Fix out-of-space error treatment in memtable flushing (CASSANDRA-11448).
  * Backport CASSANDRA-10859 (CASSANDRA-11415)
  * COPY FROM fails when importing blob (CASSANDRA-11375)
  * Backport CASSANDRA-10679 (CASSANDRA-9598)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/f3b3c410/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 3d66d3a..edc3564 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -967,6 +967,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         final OpOrder.Barrier writeBarrier;
         final CountDownLatch latch = new CountDownLatch(1);
         final ReplayPosition lastReplayPosition;
+        volatile FSWriteError flushFailure = null;
 
         private PostFlush(boolean flushSecondaryIndexes, OpOrder.Barrier writeBarrier, ReplayPosition
lastReplayPosition)
         {
@@ -1009,12 +1010,16 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
             // must check lastReplayPosition != null because Flush may find that all memtables
are clean
             // and so not set a lastReplayPosition
-            if (lastReplayPosition != null)
+            // If a flush errored out but the error was ignored, make sure we don't discard
the commit log.
+            if (lastReplayPosition != null && flushFailure == null)
             {
                 CommitLog.instance.discardCompletedSegments(metadata.cfId, lastReplayPosition);
             }
 
             metric.pendingFlushes.dec();
+
+            if (flushFailure != null)
+                throw flushFailure;
         }
     }
 
@@ -1114,11 +1119,20 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
             metric.memtableSwitchCount.inc();
 
-            for (Memtable memtable : memtables)
+            try
+            {
+                for (Memtable memtable : memtables)
+                {
+                    // flush the memtable
+                    MoreExecutors.sameThreadExecutor().execute(memtable.flushRunnable());
+                    reclaim(memtable);
+                }
+            }
+            catch (FSWriteError e)
             {
-                // flush the memtable
-                MoreExecutors.sameThreadExecutor().execute(memtable.flushRunnable());
-                reclaim(memtable);
+                JVMStabilityInspector.inspectThrowable(e);
+                // If we weren't killed, try to continue work but do not allow CommitLog
to be discarded.
+                postFlush.flushFailure = e;
             }
 
             // signal the post-flush we've done our work

http://git-wip-us.apache.org/repos/asf/cassandra/blob/f3b3c410/src/java/org/apache/cassandra/db/Directories.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Directories.java b/src/java/org/apache/cassandra/db/Directories.java
index 810c336..35aa447 100644
--- a/src/java/org/apache/cassandra/db/Directories.java
+++ b/src/java/org/apache/cassandra/db/Directories.java
@@ -310,7 +310,7 @@ public class Directories
             if (tooBig)
                 return null;
             else
-                throw new IOError(new IOException("All configured data directories have been
blacklisted as unwritable for erroring out"));
+                throw new FSWriteError(new IOException("All configured data directories have
been blacklisted as unwritable for erroring out"), "");
 
         // shortcut for single data directory systems
         if (candidates.size() == 1)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/f3b3c410/src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManager.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManager.java
index 0ea54ab..6838de6 100644
--- a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManager.java
+++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManager.java
@@ -35,6 +35,7 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.*;
 import org.slf4j.Logger;
@@ -554,7 +555,8 @@ public class CommitLogSegmentManager
     /**
      * @return a read-only collection of the active commit log segments
      */
-    Collection<CommitLogSegment> getActiveSegments()
+    @VisibleForTesting
+    public Collection<CommitLogSegment> getActiveSegments()
     {
         return Collections.unmodifiableCollection(activeSegments);
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/f3b3c410/src/java/org/apache/cassandra/io/util/DiskAwareRunnable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/util/DiskAwareRunnable.java b/src/java/org/apache/cassandra/io/util/DiskAwareRunnable.java
index 925efd6..1a15d6f 100644
--- a/src/java/org/apache/cassandra/io/util/DiskAwareRunnable.java
+++ b/src/java/org/apache/cassandra/io/util/DiskAwareRunnable.java
@@ -17,7 +17,10 @@
  */
 package org.apache.cassandra.io.util;
 
+import java.io.IOException;
+
 import org.apache.cassandra.db.Directories;
+import org.apache.cassandra.io.FSWriteError;
 import org.apache.cassandra.utils.WrappedRunnable;
 
 public abstract class DiskAwareRunnable extends WrappedRunnable
@@ -26,7 +29,7 @@ public abstract class DiskAwareRunnable extends WrappedRunnable
     {
         Directories.DataDirectory directory = getDirectories().getWriteableLocation(writeSize);
         if (directory == null)
-            throw new RuntimeException("Insufficient disk space to write " + writeSize +
" bytes");
+            throw new FSWriteError(new IOException("Insufficient disk space to write " +
writeSize + " bytes"), "");
 
         return directory;
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/f3b3c410/test/unit/org/apache/cassandra/cql3/OutOfSpaceTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/cql3/OutOfSpaceTest.java b/test/unit/org/apache/cassandra/cql3/OutOfSpaceTest.java
new file mode 100644
index 0000000..8304aff
--- /dev/null
+++ b/test/unit/org/apache/cassandra/cql3/OutOfSpaceTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.cql3;
+
+import static junit.framework.Assert.fail;
+
+import java.io.IOError;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.cassandra.config.Config.DiskFailurePolicy;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.BlacklistedDirectories;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Directories.DataDirectory;
+import org.apache.cassandra.db.commitlog.CommitLog;
+import org.apache.cassandra.db.commitlog.CommitLogSegment;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.io.FSWriteError;
+import org.apache.cassandra.utils.JVMStabilityInspector;
+import org.apache.cassandra.utils.KillerForTests;
+
+/**
+ * Test that TombstoneOverwhelmingException gets thrown when it should be and doesn't when
it shouldn't be.
+ */
+public class OutOfSpaceTest extends CQLTester
+{
+    @Test
+    public void testFlushUnwriteableDie() throws Throwable
+    {
+        makeTable();
+        markDirectoriesUnwriteable();
+
+        KillerForTests killerForTests = new KillerForTests();
+        JVMStabilityInspector.Killer originalKiller = JVMStabilityInspector.replaceKiller(killerForTests);
+        DiskFailurePolicy oldPolicy = DatabaseDescriptor.getDiskFailurePolicy();
+        try
+        {
+            DatabaseDescriptor.setDiskFailurePolicy(DiskFailurePolicy.die);
+            flushAndExpectError();
+            Assert.assertTrue(killerForTests.wasKilled());
+            Assert.assertFalse(killerForTests.wasKilledQuietly()); //only killed quietly
on startup failure
+        }
+        finally
+        {
+            DatabaseDescriptor.setDiskFailurePolicy(oldPolicy);
+            JVMStabilityInspector.replaceKiller(originalKiller);
+        }
+    }
+
+    @Test
+    public void testFlushUnwriteableStop() throws Throwable
+    {
+        makeTable();
+        markDirectoriesUnwriteable();
+
+        DiskFailurePolicy oldPolicy = DatabaseDescriptor.getDiskFailurePolicy();
+        try
+        {
+            DatabaseDescriptor.setDiskFailurePolicy(DiskFailurePolicy.stop);
+            flushAndExpectError();
+            Assert.assertFalse(Gossiper.instance.isEnabled());
+        }
+        finally
+        {
+            DatabaseDescriptor.setDiskFailurePolicy(oldPolicy);
+        }
+    }
+
+    @Test
+    public void testFlushUnwriteableIgnore() throws Throwable
+    {
+        makeTable();
+        markDirectoriesUnwriteable();
+
+        DiskFailurePolicy oldPolicy = DatabaseDescriptor.getDiskFailurePolicy();
+        try
+        {
+            DatabaseDescriptor.setDiskFailurePolicy(DiskFailurePolicy.ignore);
+            flushAndExpectError();
+        }
+        finally
+        {
+            DatabaseDescriptor.setDiskFailurePolicy(oldPolicy);
+        }
+
+        // Next flush should succeed.
+        makeTable();
+        flush();
+    }
+
+    public void makeTable() throws Throwable
+    {
+        createTable("CREATE TABLE %s (a text, b text, c text, PRIMARY KEY (a, b));");
+
+        // insert exactly the amount of tombstones that shouldn't trigger an exception
+        for (int i = 0; i < 10; i++)
+            execute("INSERT INTO %s (a, b, c) VALUES ('key', 'column" + i + "', null);");
+    }
+
+    public void markDirectoriesUnwriteable()
+    {
+        ColumnFamilyStore cfs = Keyspace.open(keyspace()).getColumnFamilyStore(currentTable());
+        try
+        {
+            for ( ; ; )
+            {
+                DataDirectory dir = cfs.directories.getWriteableLocation(1);
+                BlacklistedDirectories.maybeMarkUnwritable(cfs.directories.getLocationForDisk(dir));
+            }
+        }
+        catch (IOError e)
+        {
+            // Expected -- marked all directories as unwritable
+        }
+    }
+
+    public void flushAndExpectError() throws InterruptedException, ExecutionException
+    {
+        try
+        {
+            Keyspace.open(KEYSPACE).getColumnFamilyStore(currentTable()).forceFlush().get();
+            fail("FSWriteError expected.");
+        }
+        catch (ExecutionException e)
+        {
+            // Correct path.
+            Assert.assertTrue(e.getCause() instanceof FSWriteError);
+        }
+
+        // Make sure commit log wasn't discarded.
+        UUID cfid = currentTableMetadata().cfId;
+        for (CommitLogSegment segment : CommitLog.instance.allocator.getActiveSegments())
+            if (segment.getDirtyCFIDs().contains(cfid))
+                return;
+        fail("Expected commit log to remain dirty for the affected table.");
+    }
+}


Mime
View raw message