---

sduskis · web-flow · commit 1432b0e15fe7 · 2019-06-14T10:37:03.000-04:00
yaml --- r: 28883 b: refs/heads/autosynth-dlp c: a560bb5 h: refs/heads/master i: 28881: 793eeb7 28879: 64fb085
diff --git a/[refs] b/[refs]
@@ -103,7 +103,7 @@ refs/tags/v0.60.0: 4cd518d0612329f8a8e53484eef4cd1651e32855
 refs/tags/v0.61.0: e4b526656bb1bf5eefd0ee578b7405147821225e
 refs/tags/v0.62.0: bbede7385d48ba08f487bdd29ec10668ace96396
 refs/heads/0.60.0-alpha: 10939381ffe0b8da32db4fe3087c86e3aa7f3e55
-refs/heads/autosynth-dlp: 288cc4b2df9737de805ed224cd08e723b735e2eb
+refs/heads/autosynth-dlp: a560bb578f1f75af30281421403aee7f35799fe5
 refs/heads/autosynth-logging: eca54b98c8cf82050bbdfc5c19139673dff9e5b8
 refs/heads/dupes: 3478c5d81fd242d0e985656645a679420a2060c2
 refs/tags/v0.63.0: 94f19b71d40f46b36120e7b9d78a1a3d41bfcbd6
diff --git a/branches/autosynth-dlp/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/branches/autosynth-dlp/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java
@@ -86,6 +86,7 @@ class MessageDispatcher {
   private final AtomicBoolean extendDeadline = new AtomicBoolean(true);
   private final Lock jobLock;
   private ScheduledFuture<?> backgroundJob;
+  private ScheduledFuture<?> setExtendedDeadlineFuture;
 
   // To keep track of number of seconds the receiver takes to process messages.
   private final Distribution ackLatencyDistribution;
@@ -239,11 +240,15 @@ public void run() {
                       int newDeadlineSec = computeDeadlineSeconds();
                       messageDeadlineSeconds.set(newDeadlineSec);
                       extendDeadlines();
-                      // Don't bother cancelling this when we stop. It'd just set an atomic boolean.
-                      systemExecutor.schedule(
-                          setExtendDeadline,
-                          newDeadlineSec - ackExpirationPadding.getSeconds(),
-                          TimeUnit.SECONDS);
+                      if (setExtendedDeadlineFuture != null && !backgroundJob.isDone()) {
+                        setExtendedDeadlineFuture.cancel(true);
+                      }
+
+                      setExtendedDeadlineFuture =
+                          systemExecutor.schedule(
+                              setExtendDeadline,
+                              newDeadlineSec - ackExpirationPadding.getSeconds(),
+                              TimeUnit.SECONDS);
                     }
                     processOutstandingAckOperations();
                   } catch (Throwable t) {
@@ -266,8 +271,12 @@ void stop() {
     try {
       if (backgroundJob != null) {
         backgroundJob.cancel(false);
-        backgroundJob = null;
       }
+      if (setExtendedDeadlineFuture != null) {
+        setExtendedDeadlineFuture.cancel(true);
+      }
+      backgroundJob = null;
+      setExtendedDeadlineFuture = null;
     } finally {
       jobLock.unlock();
     }
diff --git a/branches/autosynth-dlp/google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/SubscriberTest.java b/branches/autosynth-dlp/google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/SubscriberTest.java
@@ -19,6 +19,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
+import com.google.api.gax.core.ExecutorProvider;
 import com.google.api.gax.core.FixedExecutorProvider;
 import com.google.api.gax.core.InstantiatingExecutorProvider;
 import com.google.api.gax.core.NoCredentialsProvider;
@@ -36,6 +37,9 @@
 import io.grpc.StatusException;
 import io.grpc.inprocess.InProcessChannelBuilder;
 import io.grpc.inprocess.InProcessServerBuilder;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
@@ -112,6 +116,38 @@ public void testFailedChannel_recoverableError_channelReopened() throws Exceptio
     subscriber.stopAsync().awaitTerminated();
   }
 
+  @Test
+  public void testFailedChannel_testTerminated() throws Exception {
+    final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);
+    ExecutorProvider provider =
+        new ExecutorProvider() {
+          @Override
+          public boolean shouldAutoClose() {
+            return true;
+          }
+
+          @Override
+          public ScheduledExecutorService getExecutor() {
+            return scheduledExecutorService;
+          }
+        };
+
+    try {
+      Subscriber subscriber =
+          startSubscriber(
+              getTestSubscriberBuilder(testReceiver).setSystemExecutorProvider(provider));
+
+      // wait long enough for the MessageDispatcher to set up, which at one point
+      // caused shutdown problems.
+      Thread.sleep(100);
+      subscriber.stopAsync().awaitTerminated();
+
+      assertTrue(scheduledExecutorService.awaitTermination(10, TimeUnit.SECONDS));
+    } finally {
+      scheduledExecutorService.shutdownNow();
+    }
+  }
+
   @Test(expected = IllegalStateException.class)
   public void testFailedChannel_fatalError_subscriberFails() throws Exception {
     Subscriber subscriber =
