wip

Author: pongad

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

@@ -268,7 +268,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,6 +287,7 @@ public void run() {
       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

@@ -91,20 +91,29 @@ public void advanceTime(Duration toAdvance) {
     clock.advance(toAdvance.getMillis(), TimeUnit.MILLISECONDS);
     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);
     }
+    tick(0, TimeUnit.MILLISECONDS);
     return callable.getScheduledFuture();
   }