Fix race between executor insert and advanceTime (#1554)

This is a follow up to
#1552 .

The linked PR has a race condition:
If the ExtensionJobs is added to the queue after the alarm is set up,
it is possible that the alarm would fire before the jobs are added.
Fixing this is easy, just set up the alarm after.
However, doing this consistently deadlocks the tests. Why?

The tests uses a fake executor.
In tests, time does not flow naturally, but is forced to increment,
usually by the executor's `advanceTime` method.
There is a race between the test thread advancing the time and
the mock server thread inserting more tasks to the fake executor.
If the tasks get inserted first, all is well.
Otherwise, `advanceTime` doesn't see the tasks,
and they do not get executed.
The fix is to check the "current time" every time a task is inserted.
If the task is inserted "in the past", we run the task immediately.
Doing this still deadlocks the tests. Why?

The fake executor needs to lock the task queue when registering a task.
If the task is inserted in the past, it also needs to run the task.
Running the task might involve sending a requst to the mock server.
A GRPC thread on the mock server might handle the request by adding more
tasks to the executor.
The executor's queue is locked by the first thread,
resulting in a deadlock.
The fix is to lock the queue just long enough to retrieve a task,
then execute the task without the lock.

Author: pongad

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageDispatcher.java

@@ -272,7 +272,6 @@ public void processReceivedMessages(List<com.google.pubsub.v1.ReceivedMessage> r
     }
     Instant expiration = now.plus(messageDeadlineSeconds * 1000);
     logger.debug("Received {} messages at {}", responseMessages.size(), now);
-    setupNextAckDeadlineExtensionAlarm(expiration);
 
     messagesWaiter.incrementPendingMessages(responseMessages.size());
     Iterator<AckHandler> acksIterator = ackHandlers.iterator();
@@ -288,15 +287,11 @@ public void run() {
           });
     }
 
-    // There is a race condition. setupNextAckDeadlineExtensionAlarm might set
-    // an alarm that fires before this block can run.
-    // The fix is to move setup below this block, but doing so aggravates another
-    // race condition.
-    // TODO(pongad): Fix both races.
     synchronized (outstandingAckHandlers) {
       outstandingAckHandlers.add(
           new ExtensionJob(expiration, INITIAL_ACK_DEADLINE_EXTENSION_SECONDS, ackHandlers));
     }
+    setupNextAckDeadlineExtensionAlarm(expiration);
 
     try {
       flowController.reserve(receivedMessagesCount, totalByteCount);

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

@@ -79,32 +79,41 @@ public ScheduledFuture<?> scheduleWithFixedDelay(
             Duration.millis(unit.toMillis(initialDelay)), command, PendingCallableType.FIXED_DELAY));
   }
 
-  public void tick(long time, TimeUnit unit) {
-    advanceTime(Duration.millis(unit.toMillis(time)));
-  }
-
   /**
    * This will advance the reference time of the executor and execute (in the same thread) any
    * outstanding callable which execution time has passed.
    */
   public void advanceTime(Duration toAdvance) {
     clock.advance(toAdvance.getMillis(), TimeUnit.MILLISECONDS);
+    work();
+  }
+
+  private void work() {
     DateTime cmpTime = new DateTime(clock.millis());
 
-    synchronized (pendingCallables) {
-      while (!pendingCallables.isEmpty()
-          && pendingCallables.peek().getScheduledTime().compareTo(cmpTime) <= 0) {
-        try {
-          pendingCallables.poll().call();
-          if (shutdown.get() && pendingCallables.isEmpty()) {
-            pendingCallables.notifyAll();
-          }
+    for (;;) {
+      PendingCallable<?> callable = null;
+      synchronized (pendingCallables) {
+        if (pendingCallables.isEmpty() || pendingCallables.peek().getScheduledTime().isAfter(cmpTime)) {
+          break;
+        }
+        callable = pendingCallables.poll();
+      }
+      if (callable != null) {
+        try{
+          callable.call();
         } catch (Exception e) {
           // We ignore any callable exception, which should be set to the future but not relevant to
           // advanceTime.
         }
       }
     }
+
+    synchronized (pendingCallables) {
+      if (shutdown.get() && pendingCallables.isEmpty()) {
+        pendingCallables.notifyAll();
+      }
+    }
   }
 
   @Override
@@ -172,6 +181,7 @@ <V> ScheduledFuture<V> schedulePendingCallable(PendingCallable<V> callable) {
     synchronized (pendingCallables) {
       pendingCallables.add(callable);
     }
+    work();
     return callable.getScheduledFuture();
   }