---

pongad · web-flow · commit ef7c9e1b19b0 · 2017-09-29T23:04:57.000+10:00
yaml --- r: 8539 b: refs/heads/snehashah-bugfix c: d2419d4 h: refs/heads/master i: 8537: f12c351 8535: 6939f26
diff --git a/[refs] b/[refs]
@@ -73,4 +73,4 @@ refs/heads/vkedia-patch-3: 4d128043acaa7db9160faf439d2ca6104e8a88cb
 refs/tags/v0.23.0: e5405e1f6d144441b889acd3b6405fdcc3cdfd72
 refs/tags/v0.23.1: 30bcf8076ef9d71cc5a858d026cb907bb0954bec
 refs/tags/v0.24.0: b3cf61898d9c63d028fe088c14486721318d5fd5
-refs/heads/snehashah-bugfix: 8f7792580e9e3de9d3b8e1fb5f61f9dba413a660
+refs/heads/snehashah-bugfix: d2419d40e9a28f1a79d17b793b740c3c6c140fc9
diff --git a/branches/snehashah-bugfix/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/branches/snehashah-bugfix/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(
