---

dpcollins-google · sduskis · commit cf18158e3a55 · 2019-03-04T14:55:56.000-05:00
yaml --- r: 19287 b: refs/heads/autosynth-bigquerystorage c: 7b9807d h: refs/heads/master i: 19285: 71aa4c7 19283: be8c26d 19279: 89dc801
diff --git a/[refs] b/[refs]
@@ -124,7 +124,7 @@ refs/tags/v0.69.0: 78f67a29e8b9c46ba01de566a2eae0fd1c03edea
 refs/heads/autosynth-asset: cd8251de8c40e239ad24dcf9ed93ea2708a3eed5
 refs/heads/autosynth-automl: 2a8b018cf05811fd472e5d1053e67a12ceec0b64
 refs/heads/autosynth-bigquerydatatransfer: 564833a85642d4194adc025f021e10e723154246
-refs/heads/autosynth-bigquerystorage: 7971d0b4e04147c77f1a118cbc5408f6eb90d500
+refs/heads/autosynth-bigquerystorage: 7b9807d5d0a400c757b8905fee768be4c85eba25
 refs/heads/autosynth-bigtable: 2fbcb15847e0e89e79d6dc07420e28d7dfcea894
 refs/heads/autosynth-bigtable-admin: 6379a2bc712f2736c83de0e009b4d26da4fa82ca
 refs/heads/autosynth-containeranalysis: 039ca5b8db725c76c16a965ff26b2774322b8ef8
diff --git a/branches/autosynth-bigquerystorage/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/branches/autosynth-bigquerystorage/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java
@@ -365,38 +365,32 @@ public void processReceivedMessages(List<ReceivedMessage> messages, Runnable don
     }
 
     messagesWaiter.incrementPendingMessages(outstandingBatch.messages.size());
-    synchronized (outstandingMessageBatches) {
-      outstandingMessageBatches.add(outstandingBatch);
-    }
+    outstandingMessageBatches.add(outstandingBatch);
     processOutstandingBatches();
   }
 
   private void processOutstandingBatches() {
-    synchronized (outstandingMessageBatches) {
-      for (OutstandingMessageBatch nextBatch = outstandingMessageBatches.poll();
-          nextBatch != null;
-          nextBatch = outstandingMessageBatches.poll()) {
-        for (OutstandingMessage nextMessage = nextBatch.messages.poll();
-            nextMessage != null;
-            nextMessage = nextBatch.messages.poll()) {
-          try {
-            // This is a non-blocking flow controller.
-            flowController.reserve(1, nextMessage.receivedMessage.getMessage().getSerializedSize());
-          } catch (FlowController.MaxOutstandingElementCountReachedException
-              | FlowController.MaxOutstandingRequestBytesReachedException flowControlException) {
-            // Unwind previous changes in the batches outstanding.
-            nextBatch.messages.addFirst(nextMessage);
-            outstandingMessageBatches.addFirst(nextBatch);
-            return;
-          } catch (FlowControlException unexpectedException) {
-            throw new IllegalStateException(
-                "Flow control unexpected exception", unexpectedException);
-          }
-          processOutstandingMessage(
-              nextMessage.receivedMessage.getMessage(), nextMessage.ackHandler);
+    for (OutstandingMessageBatch nextBatch = outstandingMessageBatches.poll();
+        nextBatch != null;
+        nextBatch = outstandingMessageBatches.poll()) {
+      for (OutstandingMessage nextMessage = nextBatch.messages.poll();
+          nextMessage != null;
+          nextMessage = nextBatch.messages.poll()) {
+        try {
+          // This is a non-blocking flow controller.
+          flowController.reserve(1, nextMessage.receivedMessage.getMessage().getSerializedSize());
+        } catch (FlowController.MaxOutstandingElementCountReachedException
+            | FlowController.MaxOutstandingRequestBytesReachedException flowControlException) {
+          // Unwind previous changes in the batches outstanding.
+          nextBatch.messages.addFirst(nextMessage);
+          outstandingMessageBatches.addFirst(nextBatch);
+          return;
+        } catch (FlowControlException unexpectedException) {
+          throw new IllegalStateException("Flow control unexpected exception", unexpectedException);
         }
-        nextBatch.doneCallback.run();
+        processOutstandingMessage(nextMessage.receivedMessage.getMessage(), nextMessage.ackHandler);
       }
+      nextBatch.doneCallback.run();
     }
   }
 
