pubsub: fix testStartStopReset (#1453)

The test verifies that calling stop() actually stops the emulator.
The previous implementation stops the emulator by
- sending HTTP POST to /shutdown
- terminating the emulator process
The second step follows directly after the first.
It then queries the server it just shut down to make sure it does not
respond.

This behavior is incorrect for two reasons.
- The server might continue serving request after responding to
  /shutdown.
- The "emulator process" might not be the emulator at all.
  It might be a script that in-turn runs the emulator.
  On UNIX systems, killing a parent process does not kill any of its
  children; the actual emulator might run indefinitely after
  we kill the process we spawned.

This commit changes this behavior.
After POSTing to /shutdown, we simply wait for the spawned process
to terminate on its own.

Technically, there is still a race:
Another emulator could have been started on the same port.
When we send a request to verify that the old emulator has terminated,
we might get a response from the new one.
Since we execute tests sequentially though,
I believe this should not be a problem in practice.

Author: pongad

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);
   }