Cleanup execution changes

presto-main/src/main/java/com/facebook/presto/execution/SqlStageExecution.java

@@ -81,17 +81,23 @@ public final class SqlStageExecution
     private final Map<PlanFragmentId, RemoteSourceNode> exchangeSources;
 
     private final Map<Node, Set<RemoteTask>> tasks = new ConcurrentHashMap<>();
+
+    @GuardedBy("this")
     private final AtomicInteger nextTaskId = new AtomicInteger();
     @GuardedBy("this")
     private final Set<TaskId> allTasks = newConcurrentHashSet();
     @GuardedBy("this")
     private final Set<TaskId> finishedTasks = newConcurrentHashSet();
     @GuardedBy("this")
     private final Set<TaskId> tasksWithFinalInfo = newConcurrentHashSet();
+    @GuardedBy("this")
     private final AtomicBoolean splitsScheduled = new AtomicBoolean();
 
+    @GuardedBy("this")
     private final Multimap<PlanNodeId, RemoteTask> sourceTasks = HashMultimap.create();
+    @GuardedBy("this")
     private final Set<PlanNodeId> completeSources = newConcurrentHashSet();
+    @GuardedBy("this")
     private final Set<PlanFragmentId> completeSourceFragments = newConcurrentHashSet();
 
     private final AtomicReference<OutputBuffers> outputBuffers = new AtomicReference<>();

presto-main/src/main/java/com/facebook/presto/util/FinalizerService.java

@@ -39,7 +39,8 @@ public class FinalizerService
 
     private final Set<FinalizerReference> finalizers = Sets.newConcurrentHashSet();
     private final ReferenceQueue<Object> finalizerQueue = new ReferenceQueue<>();
-    private final ExecutorService executor = newSingleThreadExecutor(daemonThreadsNamed("FinalizerService"));
+    @GuardedBy("this")
+    private ExecutorService executor;
 
     @GuardedBy("this")
     private Future<?> finalizerTask;
@@ -50,6 +51,9 @@ public synchronized void start()
         if (finalizerTask != null) {
             return;
         }
+        if (executor == null) {
+            executor = newSingleThreadExecutor(daemonThreadsNamed("FinalizerService"));
+        }
         if (executor.isShutdown()) {
             throw new IllegalStateException("Finalizer service has been destroyed");
         }
@@ -63,7 +67,10 @@ public synchronized void destroy()
             finalizerTask.cancel(true);
             finalizerTask = null;
         }
-        executor.shutdownNow();
+        if (executor != null) {
+            executor.shutdownNow();
+            executor = null;
+        }
     }
 
     /**

presto-main/src/test/java/com/facebook/presto/execution/TestSqlStageExecution.java

@@ -77,7 +77,9 @@ public void setUp()
     public void tearDown()
     {
         executor.shutdownNow();
+        executor = null;
         scheduledExecutor.shutdownNow();
+        scheduledExecutor = null;
     }
 
     @Test(timeOut = 2 * 60 * 1000)

presto-main/src/test/java/com/facebook/presto/execution/TestStateMachine.java

@@ -240,8 +240,8 @@ private static void assertStateChange(StateMachine<State> stateMachine, StateCha
 
         assertEquals(stateMachine.get(), expectedState);
 
-        assertEquals(futureChange.get(1, SECONDS), expectedState);
-        assertEquals(listenerChange.get(1, SECONDS), expectedState);
+        assertEquals(futureChange.get(10, SECONDS), expectedState);
+        assertEquals(listenerChange.get(10, SECONDS), expectedState);
 
         // listeners should not be retained if we are in a terminal state
         boolean isTerminalState = stateMachine.isTerminalState(expectedState);
@@ -272,7 +272,7 @@ private static void assertNoStateChange(StateMachine<State> stateMachine, StateC
         boolean isTerminalState = stateMachine.isTerminalState(initialState);
         if (isTerminalState) {
             assertEquals(stateMachine.getStateChangeListeners(), ImmutableSet.of());
-            assertEquals(listenerChange.get(1, SECONDS), initialState);
+            assertEquals(listenerChange.get(10, SECONDS), initialState);
         }
 
         stateChange.run();

presto-main/src/test/java/com/facebook/presto/operator/TestDriver.java

@@ -118,7 +118,7 @@ public void testNormalFinish()
     }
 
     // The race can be reproduced somewhat reliably when the invocationCount is 10K, but we use 1K iterations to cap the test runtime.
-    @Test(invocationCount = 1_000, timeOut = 1_000)
+    @Test(invocationCount = 1_000, timeOut = 10_000)
     public void testConcurrentClose()
     {
         List<Type> types = ImmutableList.of(VARCHAR, BIGINT, BIGINT);