---

pongad · web-flow · commit 3c23da1572ca · 2017-01-24T12:31:34.000+11:00
yaml --- r: 7797 b: refs/heads/tswast-patch-1 c: 2c858ee h: refs/heads/master i: 7795: 93f2edf
diff --git a/[refs] b/[refs]
@@ -57,5 +57,5 @@ refs/tags/v0.18.0: 9d193c4c4b9d1c6f21515dd8e50836b9194ec9bb
 refs/tags/v0.19.0: e67b56e4d8dad5f9a7b38c9b2107c23c828f2ed5
 refs/tags/v0.20.0: 839f7fb7156535146aa1cb2c5aadd8d375d854e8
 refs/tags/v0.20.1: 370471f437f1f4f68a11e068df5cd6bf39edb1fa
-refs/heads/tswast-patch-1: 4e45df5efe48002700227f98fa13ea76ce3cf174
+refs/heads/tswast-patch-1: 2c858ee0383f7b9063758ce9e401aab7fba1b94f
 refs/heads/pubsub-streaming-pull: 19262b752ee874eb2ca3b950eb2aef44d5a5267b
diff --git a/branches/tswast-patch-1/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageDispatcher.java b/branches/tswast-patch-1/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);
diff --git a/branches/tswast-patch-1/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/FakeScheduledExecutorService.java b/branches/tswast-patch-1/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();
   }
 
