pubsub: fix race in messageDeadlineSeconds (#2470)

pongad · web-flow · commit d2419d40e9a2 · 2017-09-29T23:04:57.000+10:00
The variable is concurrently accessed,
so it should be made atomic.
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/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(
