Merge branch 'cassandra-2.2' into cassandra-3.0

CHANGES.txt

@@ -30,6 +30,7 @@
  * Nodetool listsnapshots output is missing a newline, if there are no snapshots (CASSANDRA-13568)
  * sstabledump reports incorrect usage for argument order (CASSANDRA-13532)
 Merged from 2.2:
+ * Fix compaction and flush exception not captured (CASSANDRA-13833)
  * Uncaught exceptions in Netty pipeline (CASSANDRA-13649)
  * Prevent integer overflow on exabyte filesystems (CASSANDRA-13067)
  * Fix queries with LIMIT and filtering on clustering columns (CASSANDRA-11223)

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -841,7 +841,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
             Flush flush = new Flush(false);
             flushExecutor.execute(flush);
             ListenableFutureTask<ReplayPosition> task = ListenableFutureTask.create(flush.postFlush);
-            postFlushExecutor.submit(task);
+            postFlushExecutor.execute(task);
             return task;
         }
     }

src/java/org/apache/cassandra/db/compaction/CompactionManager.java

@@ -1487,7 +1487,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)
         {
@@ -1567,7 +1567,7 @@ public class CompactionManager implements CompactionManagerMBean
             try
             {
                 ListenableFutureTask ret = ListenableFutureTask.create(task);
-                submit(ret);
+                execute(ret);
                 return ret;
             }
             catch (RejectedExecutionException ex)

test/unit/org/apache/cassandra/db/compaction/CompactionExecutorTest.java

@@ -0,0 +1,107 @@
+/*
+ * 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.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(
+            () -> { 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(
+            () -> { 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(
+        () -> { throw new RuntimeException("testExceptionRunnable"); }
+        , "compactionExecutorTest");
+
+        while (!tt.isDone())
+            Thread.sleep(10);
+        assertNotNull(testTaskThrowable);
+        assertEquals(testTaskThrowable.getMessage(), "testExceptionRunnable");
+    }
+}