pubsub: reverse the padding direction

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

@@ -58,7 +58,7 @@ class MessageDispatcher {
 
   private static final int INITIAL_ACK_DEADLINE_EXTENSION_SECONDS = 2;
   @VisibleForTesting static final Duration PENDING_ACKS_SEND_DELAY = Duration.ofMillis(100);
-  private static final int MAX_ACK_DEADLINE_EXTENSION_SECS = 10 * 60; // 10m
+  static final Duration MAX_ACK_DEADLINE = Duration.ofSeconds(600); // 10m
 
   private final ScheduledExecutorService executor;
   private final ScheduledExecutorService systemExecutor;
@@ -117,7 +117,8 @@ void extendExpiration(Instant now) {
       Instant possibleExtension = now.plus(Duration.ofSeconds(nextExtensionSeconds));
       Instant maxExtension = creation.plus(maxAckExtensionPeriod);
       expiration = possibleExtension.isBefore(maxExtension) ? possibleExtension : maxExtension;
-      nextExtensionSeconds = Math.min(2 * nextExtensionSeconds, MAX_ACK_DEADLINE_EXTENSION_SECS);
+      nextExtensionSeconds =
+          Math.min(2 * nextExtensionSeconds, (int) (MAX_ACK_DEADLINE.getSeconds()));
     }
 
     @Override
@@ -256,7 +257,7 @@ void sendAckOperations(
     outstandingAckHandlers = new PriorityQueue<>();
     pendingAcks = new HashSet<>();
     pendingNacks = new HashSet<>();
-    // 601 buckets of 1s resolution from 0s to MAX_ACK_DEADLINE_SECONDS
+    // 601 buckets of 1s resolution from 0s to MAX_ACK_DEADLINE.getSeconds()
     this.ackLatencyDistribution = ackLatencyDistribution;
     alarmsLock = new ReentrantLock();
     nextAckDeadlineExtensionAlarmTime = Instant.ofEpochMilli(Long.MAX_VALUE);
@@ -278,8 +279,8 @@ public void stop() {
     processOutstandingAckOperations();
   }
 
-  public void setMessageDeadlineSeconds(int messageDeadlineSeconds) {
-    this.messageDeadlineSeconds.set(messageDeadlineSeconds);
+  public void setMessageDeadline(Duration dur) {
+    this.messageDeadlineSeconds.set((int) (dur.getSeconds()));
   }
 
   public int getMessageDeadlineSeconds() {

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/PollingSubscriberConnection.java

@@ -90,7 +90,7 @@ public PollingSubscriberConnection(
             executor,
             systemExecutor,
             clock);
-    messageDispatcher.setMessageDeadlineSeconds(subscription.getAckDeadlineSeconds());
+    messageDispatcher.setMessageDeadline(Duration.ofSeconds(subscription.getAckDeadlineSeconds()));
     this.maxDesiredPulledMessages =
         maxDesiredPulledMessages != null
             ? Ints.saturatedCast(maxDesiredPulledMessages)

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java

@@ -24,6 +24,7 @@
 import com.google.cloud.pubsub.v1.MessageDispatcher.PendingModifyAckDeadline;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
+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;
@@ -59,6 +60,7 @@ final class StreamingSubscriberConnection extends AbstractApiService implements
   private final String subscription;
   private final ScheduledExecutorService executor;
   private final MessageDispatcher messageDispatcher;
+  private final Duration ackExpirationPadding;
 
   private final AtomicLong channelReconnectBackoffMillis =
       new AtomicLong(INITIAL_CHANNEL_RECONNECT_BACKOFF.toMillis());
@@ -82,19 +84,22 @@ public StreamingSubscriberConnection(
     this.subscription = subscription;
     this.executor = executor;
     this.asyncStub = asyncStub;
+    this.ackExpirationPadding = ackExpirationPadding;
     this.messageDispatcher =
         new MessageDispatcher(
             receiver,
             this,
-            ackExpirationPadding,
+            Duration.ZERO,
             maxAckExtensionPeriod,
             ackLatencyDistribution,
             flowController,
             outstandingMessageBatches,
             executor,
             alarmsExecutor,
             clock);
-    messageDispatcher.setMessageDeadlineSeconds(streamAckDeadlineSeconds);
+    messageDispatcher.setMessageDeadline(
+        intendedToExtensionDeadline(
+            Duration.ofSeconds(streamAckDeadlineSeconds), ackExpirationPadding));
   }
 
   @Override
@@ -185,14 +190,19 @@ private void initialize() {
     final ClientCallStreamObserver<StreamingPullRequest> requestObserver =
         (ClientCallStreamObserver<StreamingPullRequest>)
             (asyncStub.streamingPull(responseObserver));
+
+    Duration deadline =
+        extensionToServerDeadline(
+            Duration.ofSeconds(messageDispatcher.getMessageDeadlineSeconds()),
+            ackExpirationPadding);
     logger.log(
         Level.FINER,
         "Initializing stream to subscription {0} with deadline {1}",
-        new Object[] {subscription, messageDispatcher.getMessageDeadlineSeconds()});
+        new Object[] {subscription, deadline});
     requestObserver.onNext(
         StreamingPullRequest.newBuilder()
             .setSubscription(subscription)
-            .setStreamAckDeadlineSeconds(messageDispatcher.getMessageDeadlineSeconds())
+            .setStreamAckDeadlineSeconds(Ints.saturatedCast(deadline.getSeconds()))
             .build());
     requestObserver.request(1);
 
@@ -261,7 +271,8 @@ private boolean isAlive() {
   public void sendAckOperations(
       List<String> acksToSend, List<PendingModifyAckDeadline> ackDeadlineExtensions) {
     List<StreamingPullRequest> requests =
-        partitionAckOperations(acksToSend, ackDeadlineExtensions, MAX_PER_REQUEST_CHANGES);
+        partitionAckOperations(
+            acksToSend, ackDeadlineExtensions, MAX_PER_REQUEST_CHANGES, ackExpirationPadding);
     lock.lock();
     try {
       for (StreamingPullRequest request : requests) {
@@ -276,7 +287,10 @@ public void sendAckOperations(
 
   @VisibleForTesting
   static List<StreamingPullRequest> partitionAckOperations(
-      List<String> acksToSend, List<PendingModifyAckDeadline> ackDeadlineExtensions, int size) {
+      List<String> acksToSend,
+      List<PendingModifyAckDeadline> ackDeadlineExtensions,
+      int size,
+      Duration padding) {
     int numExtensions = 0;
     for (PendingModifyAckDeadline modify : ackDeadlineExtensions) {
       numExtensions += modify.ackIds.size();
@@ -298,10 +312,18 @@ static List<StreamingPullRequest> partitionAckOperations(
     reqCount = 0;
     int ackCount = 0;
     for (PendingModifyAckDeadline modify : ackDeadlineExtensions) {
+      // If the extension is zero, it's a nack, don't do anything.
+      // Otherwise, add the padding.
+      int serverExtensionSeconds = 0;
+      if (modify.deadlineExtensionSeconds != 0) {
+        Duration intendedDeadlineExtension = Duration.ofSeconds(modify.deadlineExtensionSeconds);
+        Duration serverExtension = intendedToServerDeadline(intendedDeadlineExtension, padding);
+        serverExtensionSeconds = (int) (serverExtension.getSeconds());
+      }
       for (String ackId : modify.ackIds) {
         requests
             .get(reqCount)
-            .addModifyDeadlineSeconds(modify.deadlineExtensionSeconds)
+            .addModifyDeadlineSeconds(serverExtensionSeconds)
             .addModifyDeadlineAckIds(ackId);
         ackCount++;
         if (ackCount == size) {
@@ -318,18 +340,66 @@ static List<StreamingPullRequest> partitionAckOperations(
     return ret;
   }
 
-  public void updateStreamAckDeadline(int newAckDeadlineSeconds) {
-    messageDispatcher.setMessageDeadlineSeconds(newAckDeadlineSeconds);
+  public void updateStreamAckDeadline(int intendedDeadlineSeconds) {
+    Duration intendedDeadline = Duration.ofSeconds(intendedDeadlineSeconds);
+    messageDispatcher.setMessageDeadline(
+        intendedToExtensionDeadline(intendedDeadline, ackExpirationPadding));
     lock.lock();
     try {
       requestObserver.onNext(
           StreamingPullRequest.newBuilder()
-              .setStreamAckDeadlineSeconds(newAckDeadlineSeconds)
+              .setStreamAckDeadlineSeconds(
+                  (int)
+                      (intendedToServerDeadline(intendedDeadline, ackExpirationPadding)
+                          .getSeconds()))
               .build());
     } catch (Exception e) {
       logger.log(Level.WARNING, "failed to set deadline", e);
     } finally {
       lock.unlock();
     }
   }
+
+  /*
+    We define 3 forms of deadline:
+    1. intendedDeadline: The amount of time we take to process a message.
+    2. extensionDeadline: After we hold on to the message for this amount of time,
+      we ask the server for extension.
+    3. serverDeadline: The deadline we configure the stream to.
+
+    Consider the following scenarios:
+    Scenario 1.
+    We take about 10 seconds to process a message (intendedDeadline = 10s).
+    However, we expect 5s communication latency, so we have to ask the server
+    for a little more time (serverDeadline = intendedDeadline + padding = 10s + 5s = 15s)
+    so that the message doesn't expire before our ACK reaches the server.
+    The latency also affect modify deadline requests, so we need to send modify request
+    a little before the serverDeadline (extensionDeadline = serverDeadline - padding = 15s - 5s = 10s).
+
+    Scenario 2.
+    We think we take 600 seconds to process a message (intendedDeadline = 600s).
+    We also expect 5s latency, but then we can't set server deadline to 605s,
+    since pubsub only supports deadline up to 600s.
+    Our best course of action is to set serverDeadline as high as possible (serverDeadline = 600s).
+    Like scenario 1, we need to send modify deadline requests a little before
+    (extensionDeadline = serverDeadline - padding = 600s - 5s = 595s)
+
+    The helper methods below convert between these deadlines.
+  */
+
+  private static Duration intendedToServerDeadline(Duration intendedDeadline, Duration padding) {
+    Duration deadline = intendedDeadline.plus(padding);
+    if (deadline.compareTo(MessageDispatcher.MAX_ACK_DEADLINE) > 0) {
+      deadline = MessageDispatcher.MAX_ACK_DEADLINE;
+    }
+    return deadline;
+  }
+
+  private static Duration intendedToExtensionDeadline(Duration intendedDeadline, Duration padding) {
+    return intendedToServerDeadline(intendedDeadline, padding).minus(padding);
+  }
+
+  private static Duration extensionToServerDeadline(Duration extensionDeadline, Duration padding) {
+    return extensionDeadline.plus(padding);
+  }
 }

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java

@@ -90,8 +90,8 @@ public class Subscriber extends AbstractApiService {
   private static final int MAX_INBOUND_MESSAGE_SIZE =
       20 * 1024 * 1024; // 20MB API maximum message size.
   private static final int INITIAL_ACK_DEADLINE_SECONDS = 10;
-  private static final int MAX_ACK_DEADLINE_SECONDS = 600;
-  static final int MIN_ACK_DEADLINE_SECONDS = 10;
+  static final Duration MAX_ACK_DEADLINE = Duration.ofSeconds(600);
+  static final Duration MIN_ACK_DEADLINE = Duration.ofSeconds(10);
   private static final Duration ACK_DEADLINE_UPDATE_PERIOD = Duration.ofMinutes(1);
   private static final double PERCENTILE_FOR_ACK_DEADLINE_UPDATES = 99.9;
 
@@ -108,7 +108,7 @@ public class Subscriber extends AbstractApiService {
   private final ScheduledExecutorService executor;
   @Nullable private final ScheduledExecutorService alarmsExecutor;
   private final Distribution ackLatencyDistribution =
-      new Distribution(MAX_ACK_DEADLINE_SECONDS + 1);
+      new Distribution((int) (MAX_ACK_DEADLINE.getSeconds()) + 1);
   private final int numChannels;
   private final FlowController flowController;
   private final ChannelProvider channelProvider;
@@ -132,11 +132,7 @@ private Subscriber(Builder builder) {
     cachedSubscriptionNameString = subscriptionName.toString();
     ackExpirationPadding = builder.ackExpirationPadding;
     maxAckExtensionPeriod = builder.maxAckExtensionPeriod;
-    long streamAckDeadlineMillis = ackExpirationPadding.toMillis();
-    streamAckDeadlineSeconds =
-        Math.max(
-            INITIAL_ACK_DEADLINE_SECONDS,
-            Ints.saturatedCast(TimeUnit.MILLISECONDS.toSeconds(streamAckDeadlineMillis)));
+    streamAckDeadlineSeconds = INITIAL_ACK_DEADLINE_SECONDS;
     clock = builder.clock.isPresent() ? builder.clock.get() : CurrentMillisClock.getDefaultClock();
 
     flowController =
@@ -418,13 +414,11 @@ public void run() {
   }
 
   private void updateAckDeadline() {
-    // It is guaranteed this will be <= MAX_ACK_DEADLINE_SECONDS, the max of the API.
+    // It is guaranteed this will be <= MAX_ACK_DEADLINE.getSeconds(), the max of the API.
     long ackLatency = ackLatencyDistribution.getNthPercentile(PERCENTILE_FOR_ACK_DEADLINE_UPDATES);
     if (ackLatency > 0) {
       int possibleStreamAckDeadlineSeconds =
-          Math.max(
-              MIN_ACK_DEADLINE_SECONDS,
-              Ints.saturatedCast(Math.max(ackLatency, ackExpirationPadding.getSeconds())));
+          Math.max((int) (MIN_ACK_DEADLINE.getSeconds()), Ints.saturatedCast(ackLatency));
       if (streamAckDeadlineSeconds != possibleStreamAckDeadlineSeconds) {
         streamAckDeadlineSeconds = possibleStreamAckDeadlineSeconds;
         logger.log(
@@ -481,7 +475,7 @@ private void stopConnections(List<? extends ApiService> connections) {
   /** Builder of {@link Subscriber Subscribers}. */
   public static final class Builder {
     private static final Duration MIN_ACK_EXPIRATION_PADDING = Duration.ofMillis(100);
-    private static final Duration DEFAULT_ACK_EXPIRATION_PADDING = Duration.ofMillis(500);
+    private static final Duration DEFAULT_ACK_EXPIRATION_PADDING = Duration.ofSeconds(5);
     private static final Duration DEFAULT_MAX_ACK_EXTENSION_PERIOD = Duration.ofMinutes(60);
     private static final long DEFAULT_MEMORY_PERCENTAGE = 20;
 

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

@@ -93,11 +93,11 @@ public void setupScheduleExpectation(Duration delay) {
       expectedWorkQueue.add(delay);
     }
   }
-  
+
   /**
-   * Blocks the current thread until all the work 
-   * {@link FakeScheduledExecutorService#setupScheduleExpectation(Duration) expected} has been
-   * scheduled in the executor. 
+   * Blocks the current thread until all the work {@link
+   * FakeScheduledExecutorService#setupScheduleExpectation(Duration) expected} has been scheduled in
+   * the executor.
    */
   public void waitForExpectedWork() {
     synchronized (expectedWorkQueue) {
@@ -224,7 +224,7 @@ <V> ScheduledFuture<V> schedulePendingCallable(PendingCallable<V> callable) {
       }
       expectedWorkQueue.notifyAll();
     }
-    
+
     return callable.getScheduledFuture();
   }
 

google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/FakeSubscriberServiceImpl.java

@@ -51,7 +51,7 @@ class FakeSubscriberServiceImpl extends SubscriberImplBase {
   private final AtomicBoolean subscriptionInitialized = new AtomicBoolean(false);
   private String subscription = "";
   private final AtomicInteger messageAckDeadline =
-      new AtomicInteger(Subscriber.MIN_ACK_DEADLINE_SECONDS);
+      new AtomicInteger((int) (Subscriber.MIN_ACK_DEADLINE.getSeconds()));
   private final AtomicInteger getSubscriptionCalled = new AtomicInteger();
   private final List<Stream> openedStreams = new ArrayList<>();
   private final List<Stream> closedStreams = new ArrayList<>();