diff --git a/CHANGES.txt b/CHANGES.txt index b405fdf3f0..3baa63b0ea 100644 --- a/CHANGES.txt +++ b/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) diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 7251244774..183176c99f 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/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 task = ListenableFutureTask.create(flush.postFlush); - postFlushExecutor.submit(task); + postFlushExecutor.execute(task); return task; } } diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index e0fe4ecb42..44839603a4 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/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 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) 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 0000000000..9b07da91ee --- /dev/null +++ b/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"); + } +}