diff --git a/build.xml b/build.xml
index 3973689511..60b66fc89b 100644
--- a/build.xml
+++ b/build.xml
@@ -1300,6 +1300,13 @@
+
+
+
diff --git a/ide/idea/workspace.xml b/ide/idea/workspace.xml
index a2dea2a44e..150f1a08bd 100644
--- a/ide/idea/workspace.xml
+++ b/ide/idea/workspace.xml
@@ -167,7 +167,7 @@
-
+
diff --git a/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java b/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java
index e51e4c2bc9..5e3e5cf396 100644
--- a/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java
+++ b/src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java
@@ -52,7 +52,7 @@ public class ScheduledExecutors
{
ExecutorService[] executors = new ExecutorService[] { scheduledFastTasks, scheduledTasks, nonPeriodicTasks, optionalTasks };
for (ExecutorService executor : executors)
- executor.shutdown();
+ executor.shutdownNow();
for (ExecutorService executor : executors)
executor.awaitTermination(60, TimeUnit.SECONDS);
}
diff --git a/src/java/org/apache/cassandra/hints/HintsBufferPool.java b/src/java/org/apache/cassandra/hints/HintsBufferPool.java
index 25f9bc11ff..f705de1ffb 100644
--- a/src/java/org/apache/cassandra/hints/HintsBufferPool.java
+++ b/src/java/org/apache/cassandra/hints/HintsBufferPool.java
@@ -17,18 +17,21 @@
*/
package org.apache.cassandra.hints;
+import java.io.Closeable;
+import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.net.MessagingService;
+import sun.nio.ch.DirectBuffer;
/**
* A primitive pool of {@link HintsBuffer} buffers. Under normal conditions should only hold two buffers - the currently
* written to one, and a reserve buffer to switch to when the first one is beyond capacity.
*/
-final class HintsBufferPool
+final class HintsBufferPool implements Closeable
{
interface FlushCallback
{
@@ -129,4 +132,9 @@ final class HintsBufferPool
allocatedBuffers++;
return HintsBuffer.create(bufferSize);
}
+
+ public void close()
+ {
+ currentBuffer.free();
+ }
}
diff --git a/src/java/org/apache/cassandra/hints/HintsService.java b/src/java/org/apache/cassandra/hints/HintsService.java
index 1a352c246c..1fd2d1a632 100644
--- a/src/java/org/apache/cassandra/hints/HintsService.java
+++ b/src/java/org/apache/cassandra/hints/HintsService.java
@@ -256,6 +256,7 @@ public final class HintsService implements HintsServiceMBean
writeExecutor.shutdownBlocking();
HintsServiceDiagnostics.dispatchingShutdown(this);
+ bufferPool.close();
}
/**
diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
index cf14c3d44f..01e30d6ee5 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
@@ -2475,4 +2475,13 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted implements RefCounted
{
EXEC.shutdown();
EXEC.awaitTermination(60, TimeUnit.SECONDS);
+ if (STRONG_LEAK_DETECTOR != null)
+ {
+ STRONG_LEAK_DETECTOR.shutdownNow();
+ STRONG_LEAK_DETECTOR.awaitTermination(60, TimeUnit.SECONDS);
+ }
}
}
diff --git a/test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java b/test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java
index d03ef4feea..04ea8b010a 100644
--- a/test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/DistributedReadWritePathTest.java
@@ -28,7 +28,16 @@ import static org.apache.cassandra.net.MessagingService.Verb.READ_REPAIR;
public class DistributedReadWritePathTest extends DistributedTestBase
{
@Test
- public void coordinatorRead() throws Throwable
+ public void coordinatorReadTest() throws Throwable
+ {
+ for (int i = 0; i < 10; i++)
+ {
+ System.out.println(i);
+ coordinatorRead();
+ }
+ }
+
+ private void coordinatorRead() throws Throwable
{
try (TestCluster cluster = createCluster(3))
{
diff --git a/test/distributed/org/apache/cassandra/distributed/Instance.java b/test/distributed/org/apache/cassandra/distributed/Instance.java
index f344411549..c68b96119c 100644
--- a/test/distributed/org/apache/cassandra/distributed/Instance.java
+++ b/test/distributed/org/apache/cassandra/distributed/Instance.java
@@ -29,6 +29,9 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
+import org.slf4j.LoggerFactory;
+
+import ch.qos.logback.classic.LoggerContext;
import org.apache.cassandra.batchlog.BatchlogManager;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.concurrent.SharedExecutorPool;
@@ -53,6 +56,7 @@ import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.VersionedValue;
import org.apache.cassandra.hints.HintsService;
import org.apache.cassandra.index.SecondaryIndexManager;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.locator.InetAddressAndPort;
@@ -64,6 +68,7 @@ import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.async.MessageInHandler;
+import org.apache.cassandra.net.async.NettyFactory;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.ActiveRepairService;
@@ -229,6 +234,11 @@ public class Instance extends InvokableInstance
DatabaseDescriptor.createAllDirectories();
Keyspace.setInitialized();
SystemKeyspace.persistLocalMetadata();
+ // Even though we don't use MessagingService, access the static NettyFactory
+ // instance here so that we start the static event loop state
+ // (e.g. acceptGroup, inboundGroup, outboundGroup, etc ...). We can remove this
+ // once we actually use the MessagingService to communicate between nodes
+ NettyFactory.instance.getClass();
}).accept(config);
}
@@ -330,10 +340,10 @@ public class Instance extends InvokableInstance
runOnInstance(() -> {
Throwable error = null;
error = runAndMergeThrowable(error,
+ CompactionManager.instance::forceShutdown,
BatchlogManager.instance::shutdown,
HintsService.instance::shutdownBlocking,
CommitLog.instance::shutdownBlocking,
- CompactionManager.instance::forceShutdown,
Gossiper.instance::stop,
SecondaryIndexManager::shutdownExecutors,
MessagingService.instance()::shutdown,
@@ -347,8 +357,12 @@ public class Instance extends InvokableInstance
StageManager::shutdownAndWait,
SharedExecutorPool.SHARED::shutdown,
Memtable.MEMORY_POOL::shutdown,
- ScheduledExecutors::shutdownAndWait);
+ ScheduledExecutors::shutdownAndWait,
+ SSTableReader::shutdownBlocking);
+
error = shutdownAndWait(error, ActiveRepairService.repairCommandExecutor);
+ LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
+ loggerContext.stop();
Throwables.maybeFail(error);
});
}
@@ -357,25 +371,11 @@ public class Instance extends InvokableInstance
{
return runAndMergeThrowable(existing, () -> {
executor.shutdownNow();
- executor.awaitTermination(5, TimeUnit.SECONDS);
+ executor.awaitTermination(20, TimeUnit.SECONDS);
assert executor.isTerminated() && executor.isShutdown() : executor;
});
}
- private static Throwable runAndMergeThrowable(Throwable existing, ThrowingRunnable runnable)
- {
- try
- {
- runnable.run();
- }
- catch (Throwable t)
- {
- return Throwables.merge(existing, t);
- }
-
- return existing;
- }
-
private static Throwable runAndMergeThrowable(Throwable existing, ThrowingRunnable ... runnables)
{
for (ThrowingRunnable runnable : runnables)
diff --git a/test/distributed/org/apache/cassandra/distributed/TestCluster.java b/test/distributed/org/apache/cassandra/distributed/TestCluster.java
index 2b979ee445..cc6cf81667 100644
--- a/test/distributed/org/apache/cassandra/distributed/TestCluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/TestCluster.java
@@ -31,16 +31,21 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import java.util.function.IntFunction;
import java.util.stream.Collectors;
import com.google.common.collect.Sets;
+import io.netty.util.concurrent.FastThreadLocal;
+import io.netty.util.concurrent.FastThreadLocalThread;
+import io.netty.util.internal.InternalThreadLocalMap;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ConsistencyLevel;
@@ -79,7 +84,7 @@ import org.apache.cassandra.utils.concurrent.SimpleCondition;
*/
public class TestCluster implements AutoCloseable
{
- private static ExecutorService exec = Executors.newCachedThreadPool(new NamedThreadFactory("cluster-async-tasks"));
+ private final ExecutorService exec = Executors.newCachedThreadPool(new NamedThreadFactory("cluster-async-tasks"));
private final File root;
private final List instances;
@@ -271,19 +276,37 @@ public class TestCluster implements AutoCloseable
}
@Override
- public void close()
+ public void close() throws InterruptedException, TimeoutException, ExecutionException
{
List> futures = instances.stream()
.map(i -> exec.submit(i::shutdown))
.collect(Collectors.toList());
-// withThreadLeakCheck(futures);
-
// Make sure to only delete directory when threads are stopped
- exec.submit(() -> {
+ Future combined = exec.submit(() -> {
FBUtilities.waitOnFutures(futures);
FileUtils.deleteRecursive(root);
});
+
+ combined.get(60, TimeUnit.SECONDS);
+
+ exec.shutdownNow();
+ exec.awaitTermination(10, TimeUnit.SECONDS);
+
+ //withThreadLeakCheck(futures);
+ Set threadSet = Thread.getAllStackTraces().keySet();
+ for (Thread thread : threadSet)
+ {
+ if (thread instanceof FastThreadLocalThread)
+ ((FastThreadLocalThread)thread).setThreadLocalMap(null);
+ }
+
+ InternalThreadLocalMap.remove();
+ InternalThreadLocalMap.destroy();
+
+ FastThreadLocal.removeAll();
+ FastThreadLocal.destroy();
+ System.gc();
}
// We do not want this check to run every time until we fix problems with tread stops