ACCUMULO-2985 Make sure the ES used to stop processes with a timeout is stopped itself.
To use a Callable to apply a timeout when trying to stop the accumulo processes in MiniAccumuloCluster, an ExecutorService is required. A single-thread ES was added as a part of ACCUMULO-2764, however, it was neglected to be stopped itself which added a non-daemon'ized thread to the JVM. This non-daemonized thread prevents the JVM from exiting cleanly. After using the ES to stop the Accumulo processes with a timeout, we need to be sure to stop the ES as well. Test with a Mocked ES ensures that the ES is stopped before MAC.stop() returns. Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/609c7267 Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/609c7267 Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/609c7267 Branch: refs/heads/master Commit: 609c7267c33368ff9cfbb459153546d22bc5d2ce Parents: dec0384 Author: Josh Elser <els...@apache.org> Authored: Fri Jul 11 12:04:09 2014 -0400 Committer: Josh Elser <els...@apache.org> Committed: Fri Jul 11 12:15:18 2014 -0400 ---------------------------------------------------------------------- minicluster/pom.xml | 5 ++ .../minicluster/MiniAccumuloCluster.java | 26 ++++++++- .../minicluster/CleanShutdownMacTest.java | 61 ++++++++++++++++++++ 3 files changed, 90 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/accumulo/blob/609c7267/minicluster/pom.xml ---------------------------------------------------------------------- diff --git a/minicluster/pom.xml b/minicluster/pom.xml index 9d2b7cb..7284ca4 100644 --- a/minicluster/pom.xml +++ b/minicluster/pom.xml @@ -77,6 +77,11 @@ <scope>test</scope> </dependency> <dependency> + <groupId>org.easymock</groupId> + <artifactId>easymock</artifactId> + <scope>test</scope> + </dependency> + <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <scope>test</scope> http://git-wip-us.apache.org/repos/asf/accumulo/blob/609c7267/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java ---------------------------------------------------------------------- diff --git a/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java index 9b0e196..10fed73 100644 --- a/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java +++ b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java @@ -130,6 +130,9 @@ public class MiniAccumuloCluster { private MiniAccumuloConfig config; private Process[] tabletServerProcesses; + + // Non-final for testing purposes + private ExecutorService executor; private Process exec(Class<? extends Object> clazz, String... args) throws IOException { String javaHome = System.getProperty("java.home"); @@ -295,7 +298,8 @@ public class MiniAccumuloCluster { zooCfg.store(fileWriter, null); fileWriter.close(); - + + this.executor = Executors.newSingleThreadExecutor(); } /** @@ -407,9 +411,27 @@ public class MiniAccumuloCluster { log.warn("GarbageCollector did not fully stop after 30 seconds", e); } } + + // ACCUMULO-2985 stop the ExecutorService after we finished using it to stop accumulo procs + if (null != executor) { + List<Runnable> tasksRemaining = executor.shutdownNow(); + + // the single thread executor shouldn't have any pending tasks, but check anyways + if (!tasksRemaining.isEmpty()) { + log.warn("Unexpectedly had " + tasksRemaining.size() + " task(s) remaining in threadpool for execution when being stopped"); + } + } } - private final ExecutorService executor = Executors.newSingleThreadExecutor(); + // Visible for testing + protected void setShutdownExecutor(ExecutorService svc) { + this.executor = svc; + } + + // Visible for testing + protected ExecutorService getShutdownExecutor() { + return executor; + } private int stopProcessWithTimeout(final Process proc, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { FutureTask<Integer> future = new FutureTask<Integer>(new Callable<Integer>() { http://git-wip-us.apache.org/repos/asf/accumulo/blob/609c7267/minicluster/src/test/java/org/apache/accumulo/minicluster/CleanShutdownMacTest.java ---------------------------------------------------------------------- diff --git a/minicluster/src/test/java/org/apache/accumulo/minicluster/CleanShutdownMacTest.java b/minicluster/src/test/java/org/apache/accumulo/minicluster/CleanShutdownMacTest.java new file mode 100644 index 0000000..b22a45e --- /dev/null +++ b/minicluster/src/test/java/org/apache/accumulo/minicluster/CleanShutdownMacTest.java @@ -0,0 +1,61 @@ +/* + * 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.accumulo.minicluster; + +import java.io.File; +import java.util.Collections; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; + +import org.easymock.EasyMock; +import org.junit.Test; + +import com.google.common.io.Files; + +/** + * + */ +public class CleanShutdownMacTest { + + @SuppressWarnings("unchecked") + @Test + public void testExecutorServiceShutdown() throws Exception { + File tmp = Files.createTempDir(); + MiniAccumuloCluster cluster = new MiniAccumuloCluster(tmp, "foo"); + ExecutorService service = cluster.getShutdownExecutor(); + + // Don't leak the one that gets created + service.shutdownNow(); + + ExecutorService mockService = EasyMock.createMock(ExecutorService.class); + Future<Integer> future = EasyMock.createMock(Future.class); + + cluster.setShutdownExecutor(mockService); + + EasyMock.expect(future.get()).andReturn(0).anyTimes(); + EasyMock.expect(mockService.submit(EasyMock.anyObject(Callable.class))).andReturn(future).anyTimes(); + EasyMock.expect(mockService.shutdownNow()).andReturn(Collections.<Runnable> emptyList()).once(); + + EasyMock.replay(mockService); + + cluster.stop(); + + EasyMock.verify(mockService); + } + +}
