---

dpcollins-google · sduskis · commit 2377cab339a5 · 2019-02-28T12:24:49.000-05:00
yaml --- r: 19787 b: refs/heads/autosynth-kms c: 54aa55d h: refs/heads/master i: 19785: 5b1e9e8 19783: 3c05802
diff --git a/[refs] b/[refs]
@@ -133,7 +133,7 @@ refs/heads/autosynth-dialogflow: 7dbc2c1ea714328ccfa4f33645045f017ff080e7
 refs/heads/autosynth-errorreporting: 1101a04e8be074802c35332d5fcf8297f61cae32
 refs/heads/autosynth-firestore: d1a44f9acc302750e37b008ecb9c1aa535cc94df
 refs/heads/autosynth-iot: f03bdd338a9056ca3b7ea6d9ca901649ba9aab78
-refs/heads/autosynth-kms: 1e18a7d8f967c8b749eb99020dedf5b05506e764
+refs/heads/autosynth-kms: 54aa55dc867becf0b7dae15b5cd86cf76801ab47
 refs/heads/autosynth-language: 6262e2eb76944f01972c887b3e684aaf65ec999a
 refs/heads/autosynth-os-login: a88a337797996a205873040a63abe1d3116f5789
 refs/heads/autosynth-redis: 0cdb2e47359d51b73763bcea8af3de62aa99119b
diff --git a/branches/autosynth-kms/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/branches/autosynth-kms/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java
@@ -40,6 +40,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executor;
+import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
@@ -90,7 +91,8 @@ class MessageDispatcher {
   private final Lock jobLock;
   private ScheduledFuture<?> backgroundJob;
 
-  private final Deque<OutstandingMessageBatch> outstandingMessageBatches;
+  private final LinkedBlockingDeque<OutstandingMessageBatch> outstandingMessageBatches =
+      new LinkedBlockingDeque<>();
 
   // To keep track of number of seconds the receiver takes to process messages.
   private final Distribution ackLatencyDistribution;
@@ -200,7 +202,6 @@ void sendAckOperations(
       Duration maxAckExtensionPeriod,
       Distribution ackLatencyDistribution,
       FlowController flowController,
-      Deque<OutstandingMessageBatch> outstandingMessageBatches,
       Executor executor,
       ScheduledExecutorService systemExecutor,
       ApiClock clock) {
@@ -211,7 +212,6 @@ void sendAckOperations(
     this.receiver = receiver;
     this.ackProcessor = ackProcessor;
     this.flowController = flowController;
-    this.outstandingMessageBatches = outstandingMessageBatches;
     // 601 buckets of 1s resolution from 0s to MAX_ACK_DEADLINE_SECONDS
     this.ackLatencyDistribution = ackLatencyDistribution;
     jobLock = new ReentrantLock();
diff --git a/branches/autosynth-kms/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java b/branches/autosynth-kms/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
@@ -42,7 +42,6 @@
 import com.google.pubsub.v1.StreamingPullRequest;
 import com.google.pubsub.v1.StreamingPullResponse;
 import io.grpc.Status;
-import java.util.Deque;
 import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -84,7 +83,6 @@ public StreamingSubscriberConnection(
       SubscriberStub stub,
       int channelAffinity,
       FlowController flowController,
-      Deque<MessageDispatcher.OutstandingMessageBatch> outstandingMessageBatches,
       ScheduledExecutorService executor,
       ScheduledExecutorService systemExecutor,
       ApiClock clock) {
@@ -100,7 +98,6 @@ public StreamingSubscriberConnection(
             maxAckExtensionPeriod,
             ackLatencyDistribution,
             flowController,
-            outstandingMessageBatches,
             executor,
             systemExecutor,
             clock);
diff --git a/branches/autosynth-kms/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java b/branches/autosynth-kms/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java
@@ -44,8 +44,6 @@
 import com.google.pubsub.v1.ProjectSubscriptionName;
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Deque;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.logging.Level;
@@ -115,8 +113,6 @@ public class Subscriber extends AbstractApiService {
 
   private final MessageReceiver receiver;
   private final List<StreamingSubscriberConnection> streamingSubscriberConnections;
-  private final Deque<MessageDispatcher.OutstandingMessageBatch> outstandingMessageBatches =
-      new LinkedList<>();
   private final ApiClock clock;
   private final List<AutoCloseable> closeables = new ArrayList<>();
 
@@ -329,7 +325,6 @@ private void startStreamingConnections() {
                 subStub,
                 i,
                 flowController,
-                outstandingMessageBatches,
                 executor,
                 alarmsExecutor,
                 clock));
diff --git a/branches/autosynth-kms/google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageDispatcherTest.java b/branches/autosynth-kms/google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageDispatcherTest.java
@@ -28,7 +28,6 @@
 import com.google.pubsub.v1.ReceivedMessage;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -117,7 +116,6 @@ public void sendAckOperations(
             Duration.ofMinutes(60),
             new Distribution(Subscriber.MAX_ACK_DEADLINE_SECONDS + 1),
             flowController,
-            new LinkedList<MessageDispatcher.OutstandingMessageBatch>(),
             MoreExecutors.directExecutor(),
             systemExecutor,
             clock);
