---

pongad · web-flow · commit 18c30224e554 · 2018-01-04T07:30:27.000Z
yaml --- r: 8605 b: refs/heads/master c: f6ed7fc h: refs/heads/master i: 8603: 70e3d0b
diff --git a/[refs] b/[refs]
@@ -1,5 +1,5 @@
 ---
-refs/heads/master: 931fb9a808d9c6f8c83d2effcae778f55aedfc7f
+refs/heads/master: f6ed7fc6ea10b05fc6a4013fd99bc9d56177e5f0
 refs/heads/travis: 47e4fee4fd5af9b2a8ce46f23c72ec95f9b195b2
 refs/heads/gh-pages: 6daca92127d91b7c2c99490080ecf8a13fa94cde
 refs/tags/0.0.9: 22f1839238f66c39e67ed4dfdcd273b1ae2e8444
diff --git a/trunk/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/trunk/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java
@@ -22,13 +22,16 @@
 import com.google.api.gax.batching.FlowController.FlowControlException;
 import com.google.api.gax.core.Distribution;
 import com.google.cloud.pubsub.v1.MessageDispatcher.OutstandingMessageBatch.OutstandingMessage;
+import com.google.common.collect.ArrayListMultimap;
 import com.google.common.primitives.Ints;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.pubsub.v1.PubsubMessage;
 import com.google.pubsub.v1.ReceivedMessage;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Deque;
 import java.util.Iterator;
@@ -50,6 +53,7 @@
 import java.util.logging.Logger;
 import org.threeten.bp.Duration;
 import org.threeten.bp.Instant;
+import org.threeten.bp.temporal.ChronoUnit;
 
 /**
  * Dispatches messages to a message receiver while handling the messages acking and lease
@@ -98,11 +102,12 @@ static class PendingModifyAckDeadline {
     final int deadlineExtensionSeconds;
 
     PendingModifyAckDeadline(int deadlineExtensionSeconds, String... ackIds) {
-      this.ackIds = new ArrayList<String>();
+      this(deadlineExtensionSeconds, Arrays.asList(ackIds));
+    }
+
+    private PendingModifyAckDeadline(int deadlineExtensionSeconds, Collection<String> ackIds) {
+      this.ackIds = new ArrayList<String>(ackIds);
       this.deadlineExtensionSeconds = deadlineExtensionSeconds;
-      for (String ackId : ackIds) {
-        addAckId(ackId);
-      }
     }
 
     public void addAckId(String ackId) {
@@ -426,21 +431,39 @@ int computeDeadlineSeconds() {
 
   @InternalApi
   void extendDeadlines() {
-    List<String> acksToSend = Collections.<String>emptyList();
-    PendingModifyAckDeadline modack = new PendingModifyAckDeadline(getMessageDeadlineSeconds());
+    int extendSeconds = getMessageDeadlineSeconds();
+    List<PendingModifyAckDeadline> modacks = new ArrayList<>();
+    PendingModifyAckDeadline modack = new PendingModifyAckDeadline(extendSeconds);
     Instant now = now();
+    Instant extendTo = now.plusSeconds(extendSeconds);
 
+    int count = 0;
     Iterator<Map.Entry<String, Instant>> it = pendingMessages.entrySet().iterator();
     while (it.hasNext()) {
       Map.Entry<String, Instant> entry = it.next();
-      if (entry.getValue().isBefore(now)) {
-        it.remove();
-      } else {
-        modack.ackIds.add(entry.getKey());
+      String ackId = entry.getKey();
+      Instant totalExpiration = entry.getValue();
+      // TODO(pongad): PendingModifyAckDeadline is created to dance around polling pull,
+      // since one modack RPC only takes one expiration.
+      // Whenever we delete polling pull, we should also delete PendingModifyAckDeadline,
+      // and just construct StreamingPullRequest directly.
+      if (totalExpiration.isAfter(extendTo)) {
+        modack.ackIds.add(ackId);
+        count++;
+        continue;
+      }
+      it.remove();
+      if (totalExpiration.isAfter(now)) {
+        int sec = Math.max(1, (int) now.until(totalExpiration, ChronoUnit.SECONDS));
+        modacks.add(new PendingModifyAckDeadline(sec, ackId));
+        count++;
       }
     }
+    modacks.add(modack);
+    logger.log(Level.FINER, "Sending {0} modacks", count);
 
-    ackProcessor.sendAckOperations(acksToSend, Collections.singletonList(modack));
+    List<String> acksToSend = Collections.<String>emptyList();
+    ackProcessor.sendAckOperations(acksToSend, modacks);
   }
 
   @InternalApi
diff --git a/trunk/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageDispatcherTest.java b/trunk/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageDispatcherTest.java
@@ -155,13 +155,30 @@ public void testExtension() throws Exception {
   }
 
   @Test
-  public void testExtension_GiveUp() throws Exception {
+  public void testExtension_Close() throws Exception {
     dispatcher.processReceivedMessages(Collections.singletonList(TEST_MESSAGE), NOOP_RUNNABLE);
     dispatcher.extendDeadlines();
     assertThat(sentModAcks)
         .contains(ModAckItem.of(TEST_MESSAGE.getAckId(), Subscriber.MIN_ACK_DEADLINE_SECONDS));
+    sentModAcks.clear();
+
+    // Default total expiration is an hour (60*60 seconds). We normally would extend by 10s.
+    // However, only extend by 5s here, since there's only 5s left before total expiration.
+    clock.advance(60 * 60 - 5, TimeUnit.SECONDS);
+    dispatcher.extendDeadlines();
+    assertThat(sentModAcks).contains(ModAckItem.of(TEST_MESSAGE.getAckId(), 5));
+  }
 
+  @Test
+  public void testExtension_GiveUp() throws Exception {
+    dispatcher.processReceivedMessages(Collections.singletonList(TEST_MESSAGE), NOOP_RUNNABLE);
+    dispatcher.extendDeadlines();
+    assertThat(sentModAcks)
+        .contains(ModAckItem.of(TEST_MESSAGE.getAckId(), Subscriber.MIN_ACK_DEADLINE_SECONDS));
     sentModAcks.clear();
+
+    // If we run extendDeadlines after totalExpiration, we shouldn't send anything.
+    // In particular, don't send negative modacks.
     clock.advance(1, TimeUnit.DAYS);
     dispatcher.extendDeadlines();
     assertThat(sentModAcks).isEmpty();
