cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From marc...@apache.org
Subject [04/10] cassandra git commit: Fix compaction and flush exception not captured issue
Date Mon, 04 Sep 2017 13:05:16 GMT
Fix compaction and flush exception not captured issue

patch by Jay Zhuang; reviewed by marcuse for CASSANDRA-13833


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

Branch: refs/heads/trunk
Commit: e80ede6d393460f22ee2b313d4bac7e3fbbfe893
Parents: 4d90573
Author: Jay Zhuang <jay.zhuang@yahoo.com>
Authored: Thu Aug 31 11:07:07 2017 -0700
Committer: Marcus Eriksson <marcuse@apache.org>
Committed: Mon Sep 4 15:01:02 2017 +0200

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../apache/cassandra/db/ColumnFamilyStore.java  |   4 +-
 .../db/compaction/CompactionManager.java        |   4 +-
 .../db/compaction/CompactionExecutorTest.java   | 131 +++++++++++++++++++
 4 files changed, 136 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/e80ede6d/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 4e68ddc..03a78fd 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 2.2.11
+ * Fix compaction and flush exception not captured (CASSANDRA-13833)
  * Make BatchlogManagerMBean.forceBatchlogReplay() blocking (CASSANDRA-13809)
  * Uncaught exceptions in Netty pipeline (CASSANDRA-13649)
  * Prevent integer overflow on exabyte filesystems (CASSANDRA-13067) 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e80ede6d/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 2e52eb2..7e36e11 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -906,9 +906,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
             logFlush();
             Flush flush = new Flush(false);
             ListenableFutureTask<Void> flushTask = ListenableFutureTask.create(flush,
null);
-            flushExecutor.submit(flushTask);
+            flushExecutor.execute(flushTask);
             ListenableFutureTask<ReplayPosition> task = ListenableFutureTask.create(flush.postFlush);
-            postFlushExecutor.submit(task);
+            postFlushExecutor.execute(task);
 
             @SuppressWarnings("unchecked")
             ListenableFuture<ReplayPosition> future = 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e80ede6d/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index d21f1e8..cd50646 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -1457,7 +1457,7 @@ public class CompactionManager implements CompactionManagerMBean
         return CompactionMetrics.getCompactions().size();
     }
 
-    private static class CompactionExecutor extends JMXEnabledThreadPoolExecutor
+    static class CompactionExecutor extends JMXEnabledThreadPoolExecutor
     {
         protected CompactionExecutor(int minThreads, int maxThreads, String name, BlockingQueue<Runnable>
queue)
         {
@@ -1537,7 +1537,7 @@ public class CompactionManager implements CompactionManagerMBean
             try
             {
                 ListenableFutureTask ret = ListenableFutureTask.create(task);
-                submit(ret);
+                execute(ret);
                 return ret;
             }
             catch (RejectedExecutionException ex)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e80ede6d/test/unit/org/apache/cassandra/db/compaction/CompactionExecutorTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionExecutorTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionExecutorTest.java
new file mode 100644
index 0000000..c6feb3f
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/compaction/CompactionExecutorTest.java
@@ -0,0 +1,131 @@
+/*
+ * 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.db.compaction;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+public class CompactionExecutorTest
+{
+    static Throwable testTaskThrowable = null;
+    private static class TestTaskExecutor extends CompactionManager.CompactionExecutor
+    {
+        @Override
+        public void afterExecute(Runnable r, Throwable t)
+        {
+            if (t == null)
+            {
+                t = DebuggableThreadPoolExecutor.extractThrowable(r);
+            }
+            testTaskThrowable = t;
+        }
+        @Override
+        protected void beforeExecute(Thread t, Runnable r)
+        {
+        }
+    }
+    private CompactionManager.CompactionExecutor executor;
+
+    @Before
+    public void setup()
+    {
+        executor = new TestTaskExecutor();
+    }
+
+    @After
+    public void destroy() throws Exception
+    {
+        executor.shutdown();
+        executor.awaitTermination(1, TimeUnit.MINUTES);
+    }
+
+    @Test
+    public void testFailedRunnable() throws Exception
+    {
+        testTaskThrowable = null;
+
+        Future<?> tt = executor.submitIfRunning(
+            new Runnable()
+            {
+                @Override
+                public void run()
+                {
+                    assert false : "testFailedRunnable";
+                }
+            }, "compactionExecutorTest"
+        );
+
+        while (!tt.isDone())
+            Thread.sleep(10);
+        assertNotNull(testTaskThrowable);
+        assertEquals(testTaskThrowable.getMessage(), "testFailedRunnable");
+    }
+
+    @Test
+    public void testFailedCallable() throws Exception
+    {
+        testTaskThrowable = null;
+        Future<?> tt = executor.submitIfRunning(
+            new Callable<Integer>()
+            {
+                @Override
+                public Integer call() throws Exception
+                {
+                    assert false : "testFailedCallable";
+                    return 1;
+                }
+            }
+            , "compactionExecutorTest");
+
+        while (!tt.isDone())
+            Thread.sleep(10);
+        assertNotNull(testTaskThrowable);
+        assertEquals(testTaskThrowable.getMessage(), "testFailedCallable");
+    }
+
+    @Test
+    public void testExceptionRunnable() throws Exception
+    {
+        testTaskThrowable = null;
+        Future<?> tt = executor.submitIfRunning(
+        new Runnable()
+        {
+            @Override
+            public void run()
+            {
+                throw new RuntimeException("testExceptionRunnable");
+            }
+        }
+        , "compactionExecutorTest");
+
+        while (!tt.isDone())
+            Thread.sleep(10);
+        assertNotNull(testTaskThrowable);
+        assertEquals(testTaskThrowable.getMessage(), "testExceptionRunnable");
+    }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org


Mime
View raw message