---

sduskis · web-flow · commit 86bbaeac5e85 · 2019-04-25T16:39:04.000-04:00
yaml --- r: 30719 b: refs/heads/autosynth-bigquerydatatransfer c: abf92a3 h: refs/heads/master i: 30717: 1b01112 30715: d761553 30711: 55b6038 30703: 97365c2 30687: e8773e7 30655: 768abac 30591: c46245d 30463: 61d5885 30207: 405f7b5 29695: 0bbc4da 28671: 0ab66f1
diff --git a/[refs] b/[refs]
@@ -122,7 +122,7 @@ refs/tags/v0.68.0: 9cc799fcf68c82ab431d425fefa58ef615ce8e5b
 refs/tags/v0.69.0: 78f67a29e8b9c46ba01de566a2eae0fd1c03edea
 refs/heads/autosynth-asset: bdb45634a0fe8f7a510692b56b31f5312e25f453
 refs/heads/autosynth-automl: 22f9dd5b6f5df8dbfa7da0126864d565229519b2
-refs/heads/autosynth-bigquerydatatransfer: 264643df7bcc715efbd648c3ab40389ae2253d19
+refs/heads/autosynth-bigquerydatatransfer: abf92a3ea30a7375790f6447532f6fd45bed5744
 refs/heads/autosynth-bigquerystorage: d2c53da3b012e38c662e4df0738042435f19365f
 refs/heads/autosynth-bigtable: 9e5429f45cf9face9fed585d0233534993e36b58
 refs/heads/autosynth-bigtable-admin: 6379a2bc712f2736c83de0e009b4d26da4fa82ca
diff --git a/branches/autosynth-bigquerydatatransfer/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java b/branches/autosynth-bigquerydatatransfer/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java
@@ -40,13 +40,13 @@
 import com.google.cloud.pubsub.v1.stub.PublisherStub;
 import com.google.cloud.pubsub.v1.stub.PublisherStubSettings;
 import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
 import com.google.pubsub.v1.PublishRequest;
 import com.google.pubsub.v1.PublishResponse;
 import com.google.pubsub.v1.PubsubMessage;
 import com.google.pubsub.v1.TopicName;
 import com.google.pubsub.v1.TopicNames;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.LinkedList;
@@ -197,7 +197,7 @@ public ApiFuture<String> publish(PubsubMessage message) {
     }
 
     message = messageTransform.apply(message);
-    OutstandingBatch batchToSend = null;
+    List<OutstandingBatch> batchesToSend = new ArrayList<>();
     final OutstandingPublish outstandingPublish = new OutstandingPublish(message);
     messagesBatchLock.lock();
     try {
@@ -206,19 +206,18 @@ public ApiFuture<String> publish(PubsubMessage message) {
           && hasBatchingBytes()
           && messagesBatch.getBatchedBytes() + outstandingPublish.messageSize
               >= getMaxBatchBytes()) {
-        batchToSend = messagesBatch.popOutstandingBatch();
+        batchesToSend.add(messagesBatch.popOutstandingBatch());
       }
 
-      // Border case if the message to send is greater or equals to the max batch size then can't
-      // be included in the current batch and instead sent immediately.
-      if (!hasBatchingBytes() || outstandingPublish.messageSize < getMaxBatchBytes()) {
-        messagesBatch.addMessage(outstandingPublish, outstandingPublish.messageSize);
+      messagesBatch.addMessage(outstandingPublish, outstandingPublish.messageSize);
 
-        // If after adding the message we have reached the batch max messages then we have a batch
-        // to send.
-        if (messagesBatch.getMessagesCount() == getBatchingSettings().getElementCountThreshold()) {
-          batchToSend = messagesBatch.popOutstandingBatch();
-        }
+      // Border case: If the message to send is greater or equals to the max batch size then send it
+      // immediately.
+      // Alternatively if after adding the message we have reached the batch max messages then we
+      // have a batch to send.
+      if ((hasBatchingBytes() && outstandingPublish.messageSize >= getMaxBatchBytes())
+          || messagesBatch.getMessagesCount() == getBatchingSettings().getElementCountThreshold()) {
+        batchesToSend.add(messagesBatch.popOutstandingBatch());
       }
       // Setup the next duration based delivery alarm if there are messages batched.
       setupAlarm();
@@ -228,32 +227,17 @@ && hasBatchingBytes()
 
     messagesWaiter.incrementPendingMessages(1);
 
-    if (batchToSend != null) {
-      logger.log(Level.FINER, "Scheduling a batch for immediate sending.");
-      final OutstandingBatch finalBatchToSend = batchToSend;
-      executor.execute(
-          new Runnable() {
-            @Override
-            public void run() {
-              publishOutstandingBatch(finalBatchToSend);
-            }
-          });
-    }
-
-    // If the message is over the size limit, it was not added to the pending messages and it will
-    // be sent in its own batch immediately.
-    if (hasBatchingBytes() && outstandingPublish.messageSize >= getMaxBatchBytes()) {
-      logger.log(
-          Level.FINER, "Message exceeds the max batch bytes, scheduling it for immediate send.");
-      executor.execute(
-          new Runnable() {
-            @Override
-            public void run() {
-              publishOutstandingBatch(
-                  new OutstandingBatch(
-                      ImmutableList.of(outstandingPublish), outstandingPublish.messageSize));
-            }
-          });
+    if (!batchesToSend.isEmpty()) {
+      for (final OutstandingBatch batch : batchesToSend) {
+        logger.log(Level.FINER, "Scheduling a batch for immediate sending.");
+        executor.execute(
+            new Runnable() {
+              @Override
+              public void run() {
+                publishOutstandingBatch(batch);
+              }
+            });
+      }
     }
 
     return outstandingPublish.publishResult;
