---

pongad · web-flow · commit faadac0f54e6 · 2017-09-29T23:04:57.000+10:00
yaml --- r: 8777 b: refs/heads/lesv-patch-1 c: d2419d4 h: refs/heads/master i: 8775: ab296a3
diff --git a/[refs] b/[refs]
@@ -66,7 +66,7 @@ refs/tags/v0.22.0: 18b298fe4bfe8ec2f20b0e0bf7ffdcce5cc3c5fe
 refs/heads/vam-google-patch-1: d0c8fee3a4074d0bf7360ce8c4f7f7223d0ee7b9
 refs/heads/vam-google-patch-CODEOWNERS: 2ac1616e25229e51d08a984708ef1918f91a35ee
 refs/heads/danoscarmike-patch-1: 7342a9916bce4ed00002c7202e2a16c5d46afaea
-refs/heads/lesv-patch-1: 8f7792580e9e3de9d3b8e1fb5f61f9dba413a660
+refs/heads/lesv-patch-1: d2419d40e9a28f1a79d17b793b740c3c6c140fc9
 refs/heads/ml-update-branch: 079dd6610017f5c51b9d1938c12d6d55b61513cf
 refs/heads/vkedia-patch-2: 7d8241388a9769a5c069334761b06c7012c878e7
 refs/heads/vkedia-patch-3: 4d128043acaa7db9160faf439d2ca6104e8a88cb
diff --git a/branches/lesv-patch-1/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/branches/lesv-patch-1/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java
@@ -41,6 +41,7 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.logging.Level;
@@ -76,7 +77,9 @@ class MessageDispatcher {
   private final Set<String> pendingNacks;
 
   private final Lock alarmsLock;
-  private int messageDeadlineSeconds;
+  // The deadline should be set by the subscriber connection before use,
+  // but set it to some reasonable value just in case.
+  private final AtomicInteger messageDeadlineSeconds = new AtomicInteger(10);
   private ScheduledFuture<?> ackDeadlineExtensionAlarm;
   private Instant nextAckDeadlineExtensionAlarmTime;
   private ScheduledFuture<?> pendingAcksAlarm;
@@ -276,11 +279,11 @@ public void stop() {
   }
 
   public void setMessageDeadlineSeconds(int messageDeadlineSeconds) {
-    this.messageDeadlineSeconds = messageDeadlineSeconds;
+    this.messageDeadlineSeconds.set(messageDeadlineSeconds);
   }
 
   public int getMessageDeadlineSeconds() {
-    return messageDeadlineSeconds;
+    return messageDeadlineSeconds.get();
   }
 
   static class OutstandingMessageBatch {
@@ -336,7 +339,7 @@ public void processReceivedMessages(List<ReceivedMessage> messages, Runnable don
     }
 
     Instant expiration = Instant.ofEpochMilli(clock.millisTime())
-        .plusSeconds(messageDeadlineSeconds);
+        .plusSeconds(messageDeadlineSeconds.get());
     synchronized (outstandingAckHandlers) {
       outstandingAckHandlers.add(
           new ExtensionJob(
