pubsub: fix testStartStopReset

google-cloud-core/src/main/java/com/google/cloud/testing/BaseEmulatorHelper.java

@@ -18,6 +18,8 @@
 
 import com.google.cloud.ServiceOptions;
 import com.google.common.io.CharStreams;
+import com.google.common.util.concurrent.SettableFuture;
+import com.google.common.util.concurrent.UncheckedExecutionException;
 
 import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
@@ -43,10 +45,14 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.zip.ZipEntry;
 import java.util.zip.ZipInputStream;
+import org.joda.time.Duration;
 
 /**
  * Utility class to start and stop a local service which is used by unit testing.
@@ -104,16 +110,50 @@ protected final void startProcess(String blockUntilOutput)
   }
 
   /**
-   * Stops the local service's subprocess and any possible thread listening for its output.
+   * Waits for the local service's subprocess to terminate,
+   * and stop any possible thread listening for its output.
    */
-  protected final void stopProcess() throws IOException, InterruptedException {
+  protected final int waitForProcess(Duration timeout) throws IOException, InterruptedException, TimeoutException {
     if (blockingProcessReader != null) {
       blockingProcessReader.terminate();
       blockingProcessReader = null;
     }
     if (activeRunner != null) {
-      activeRunner.stop();
+      int exitCode = activeRunner.waitFor(timeout);
       activeRunner = null;
+      return exitCode;
+    }
+    return 0;
+  }
+
+  private static int waitForProcess(final Process process, Duration timeout) throws InterruptedException, TimeoutException {
+    if (process == null) {
+      return 0;
+    }
+
+    final SettableFuture<Integer> exitValue = SettableFuture.create();
+
+    Thread waiter = new Thread(new Runnable() {
+      @Override
+      public void run() {
+        try {
+          exitValue.set(process.waitFor());
+        } catch (InterruptedException e) {
+          exitValue.setException(e);
+        }
+      }
+    });
+    waiter.start();
+
+    try {
+      return exitValue.get(timeout.getMillis(), TimeUnit.MILLISECONDS);
+    } catch (ExecutionException e) {
+      if (e.getCause() instanceof InterruptedException) {
+        throw (InterruptedException) e.getCause();
+      }
+      throw new UncheckedExecutionException(e);
+    } finally {
+      waiter.interrupt();
     }
   }
 
@@ -144,7 +184,7 @@ public String getProjectId() {
   /**
    * Stops the local emulator.
    */
-  public abstract void stop() throws IOException, InterruptedException;
+  public abstract void stop(Duration timeout) throws IOException, InterruptedException, TimeoutException;
 
   /**
    * Resets the internal state of the emulator.
@@ -195,9 +235,10 @@ protected interface EmulatorRunner {
     void start() throws IOException;
 
     /**
-     * Stops the emulator associated to this runner.
+     * Wait for the emulator associated to this runner to terminate,
+     * returning the exit status.
      */
-    void stop() throws InterruptedException;
+    int waitFor(Duration timeout) throws InterruptedException, TimeoutException;
 
     /**
      * Returns the process associated to the emulator, if any.
@@ -239,11 +280,8 @@ public void start() throws IOException {
     }
 
     @Override
-    public void stop() throws InterruptedException {
-      if (process != null) {
-        process.destroy();
-        process.waitFor();
-      }
+    public int waitFor(Duration timeout) throws InterruptedException, TimeoutException {
+      return waitForProcess(process, timeout);
     }
 
     @Override
@@ -337,11 +375,8 @@ public void start() throws IOException {
     }
 
     @Override
-    public void stop() throws InterruptedException {
-      if (process != null) {
-        process.destroy();
-        process.waitFor();
-      }
+    public int waitFor(Duration timeout) throws InterruptedException, TimeoutException {
+      return waitForProcess(process, timeout);
     }
 
     @Override

google-cloud-core/src/test/java/com/google/cloud/testing/BaseEmulatorHelperTest.java

@@ -22,12 +22,14 @@
 import com.google.common.collect.ImmutableList;
 
 import org.easymock.EasyMock;
+import org.joda.time.Duration;
 import org.junit.Test;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.List;
+import java.util.concurrent.TimeoutException;
 import java.util.logging.Logger;
 
 public class BaseEmulatorHelperTest {
@@ -66,8 +68,8 @@ public void start() throws IOException, InterruptedException {
     }
 
     @Override
-    public void stop() throws IOException, InterruptedException {
-      stopProcess();
+    public void stop(Duration timeout) throws IOException, InterruptedException, TimeoutException {
+      waitForProcess(timeout);
     }
 
     @Override
@@ -77,7 +79,7 @@ public void reset() throws IOException {
   }
 
   @Test
-  public void testEmulatorHelper() throws IOException, InterruptedException {
+  public void testEmulatorHelper() throws IOException, InterruptedException, TimeoutException {
     Process process = EasyMock.createStrictMock(Process.class);
     InputStream stream = new ByteArrayInputStream(BLOCK_UNTIL.getBytes(Charsets.UTF_8));
     EmulatorRunner emulatorRunner = EasyMock.createStrictMock(EmulatorRunner.class);
@@ -86,18 +88,18 @@ public void testEmulatorHelper() throws IOException, InterruptedException {
     emulatorRunner.start();
     EasyMock.expectLastCall();
     EasyMock.expect(emulatorRunner.getProcess()).andReturn(process);
-    emulatorRunner.stop();
-    EasyMock.expectLastCall();
+    emulatorRunner.waitFor(Duration.standardMinutes(1));
+    EasyMock.expectLastCall().andReturn(0);
     EasyMock.replay(process, emulatorRunner);
     TestEmulatorHelper helper =
         new TestEmulatorHelper(ImmutableList.of(emulatorRunner), BLOCK_UNTIL);
     helper.start();
-    helper.stop();
+    helper.stop(Duration.standardMinutes(1));
     EasyMock.verify();
   }
 
   @Test
-  public void testEmulatorHelperMultipleRunners() throws IOException, InterruptedException {
+  public void testEmulatorHelperMultipleRunners() throws IOException, InterruptedException, TimeoutException {
     Process process = EasyMock.createStrictMock(Process.class);
     InputStream stream = new ByteArrayInputStream(BLOCK_UNTIL.getBytes(Charsets.UTF_8));
     EmulatorRunner firstRunner = EasyMock.createStrictMock(EmulatorRunner.class);
@@ -108,13 +110,13 @@ public void testEmulatorHelperMultipleRunners() throws IOException, InterruptedE
     secondRunner.start();
     EasyMock.expectLastCall();
     EasyMock.expect(secondRunner.getProcess()).andReturn(process);
-    secondRunner.stop();
-    EasyMock.expectLastCall();
+    secondRunner.waitFor(Duration.standardMinutes(1));
+    EasyMock.expectLastCall().andReturn(0);
     EasyMock.replay(process, secondRunner);
     TestEmulatorHelper helper =
         new TestEmulatorHelper(ImmutableList.of(firstRunner, secondRunner), BLOCK_UNTIL);
     helper.start();
-    helper.stop();
+    helper.stop(Duration.standardMinutes(1));
     EasyMock.verify();
   }
 }

google-cloud-datastore/src/main/java/com/google/cloud/datastore/testing/LocalDatastoreHelper.java

@@ -22,6 +22,8 @@
 import com.google.cloud.testing.BaseEmulatorHelper;
 import com.google.common.collect.ImmutableList;
 
+import org.joda.time.Duration;
+
 import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
@@ -34,6 +36,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.TimeoutException;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -225,9 +228,9 @@ public void reset() throws IOException {
   /**
    * Stops the Datastore emulator.
    */
-  public void stop() throws IOException, InterruptedException {
+  public void stop(Duration timeout) throws IOException, InterruptedException, TimeoutException {
     sendPostRequest("/shutdown");
-    stopProcess();
+    waitForProcess(timeout);
     deleteRecursively(gcdPath);
   }
 

google-cloud-datastore/src/test/java/com/google/cloud/datastore/DatastoreTest.java

@@ -47,14 +47,15 @@
 import com.google.protobuf.ByteString;
 
 import org.easymock.EasyMock;
+import org.joda.time.Duration;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Rule;
-import org.junit.Test;
 import org.junit.rules.ExpectedException;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
+import org.junit.Test;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -63,6 +64,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.TimeoutException;
 
 @RunWith(JUnit4.class)
 public class DatastoreTest {
@@ -150,8 +152,8 @@ public void setUp() {
   }
 
   @AfterClass
-  public static void afterClass() throws IOException, InterruptedException {
-    helper.stop();
+  public static void afterClass() throws IOException, InterruptedException, TimeoutException {
+    helper.stop(Duration.standardMinutes(1));
   }
 
   @Test

google-cloud-datastore/src/test/java/com/google/cloud/datastore/testing/LocalDatastoreHelperTest.java

@@ -29,13 +29,15 @@
 import com.google.cloud.datastore.Entity;
 import com.google.cloud.datastore.Key;
 
+import org.joda.time.Duration;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
 import java.io.IOException;
+import java.util.concurrent.TimeoutException;
 
 @RunWith(JUnit4.class)
 public class LocalDatastoreHelperTest {
@@ -82,7 +84,7 @@ public void testOptions() {
   }
 
   @Test
-  public void testStartStopReset() throws IOException, InterruptedException {
+  public void testStartStopReset() throws IOException, InterruptedException, TimeoutException {
     LocalDatastoreHelper helper = LocalDatastoreHelper.create();
     helper.start();
     Datastore datastore = helper.getOptions().getService();
@@ -91,7 +93,7 @@ public void testStartStopReset() throws IOException, InterruptedException {
     assertNotNull(datastore.get(key));
     helper.reset();
     assertNull(datastore.get(key));
-    helper.stop();
+    helper.stop(Duration.standardMinutes(1));
     thrown.expect(DatastoreException.class);
     datastore.get(key);
   }

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/testing/LocalPubSubHelper.java

@@ -33,8 +33,11 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.TimeoutException;
 import java.util.logging.Logger;
 
+import org.joda.time.Duration;
+
 /**
  * A class that runs a Pubsub emulator instance for use in tests.
  */
@@ -144,8 +147,8 @@ public void reset() throws IOException {
    * Stops the PubSub emulator and related local service.
    */
   @Override
-  public void stop() throws IOException, InterruptedException {
+  public void stop(Duration timeout) throws IOException, InterruptedException, TimeoutException {
     sendPostRequest("/shutdown");
-    stopProcess();
+    waitForProcess(timeout);
   }
 }

google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/LocalSystemTest.java

@@ -18,6 +18,7 @@
 
 import com.google.cloud.pubsub.testing.LocalPubSubHelper;
 
+import org.joda.time.Duration;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 
@@ -49,6 +50,6 @@ public static void startServer() throws IOException, InterruptedException {
   public static void stopServer() throws Exception {
     pubsub.close();
     pubsubHelper.reset();
-    pubsubHelper.stop();
+    pubsubHelper.stop(Duration.standardMinutes(1));
   }
 }

google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/PublisherClientTest.java

@@ -40,6 +40,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.TimeoutException;
 
 public class PublisherClientTest {
   private static LocalPubSubHelper pubsubHelper;
@@ -55,8 +56,8 @@ public static void startServer() throws IOException, InterruptedException {
   }
 
   @AfterClass
-  public static void stopServer() throws IOException, InterruptedException {
-    pubsubHelper.stop();
+  public static void stopServer() throws IOException, InterruptedException, TimeoutException {
+    pubsubHelper.stop(Duration.standardMinutes(1));
   }
 
   @Before

google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/testing/LocalPubSubHelperTest.java

@@ -27,11 +27,13 @@
 import com.google.cloud.pubsub.PubSubOptions;
 import com.google.cloud.pubsub.TopicInfo;
 
+import org.joda.time.Duration;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
 
 import java.io.IOException;
+import java.util.concurrent.TimeoutException;
 
 public class LocalPubSubHelperTest {
 
@@ -57,15 +59,15 @@ public void testOptions() {
   }
 
   @Test
-  public void testStartStopReset() throws IOException, InterruptedException {
+  public void testStartStopReset() throws IOException, InterruptedException, TimeoutException {
     LocalPubSubHelper helper = LocalPubSubHelper.create();
     helper.start();
     PubSub pubsub = helper.getOptions().getService();
     pubsub.create(TopicInfo.of(TOPIC));
     assertNotNull(pubsub.getTopic(TOPIC));
     helper.reset();
     assertNull(pubsub.getTopic(TOPIC));
-    helper.stop();
+    helper.stop(Duration.standardMinutes(1));
     thrown.expect(PubSubException.class);
     pubsub.getTopic(TOPIC);
   }