Change the order of publishing batches when a large message is requested; Add unit tests (#4578)

Author: kimkyung-goog

google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java

@@ -269,18 +269,18 @@ && hasBatchingBytes()
 
     if (batchToSend != null) {
       logger.log(Level.FINER, "Scheduling a batch for immediate sending.");
-      publishOutstandingBatch(batchToSend);
       publishAllOutstanding();
+      publishOutstandingBatch(batchToSend);
     }
 
     // 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() && messageSize >= getMaxBatchBytes()) {
       logger.log(
           Level.FINER, "Message exceeds the max batch bytes, scheduling it for immediate send.");
+      publishAllOutstanding();
       publishOutstandingBatch(
           new OutstandingBatch(ImmutableList.of(outstandingPublish), messageSize, orderingKey));
-      publishAllOutstanding();
     }
 
     return publishResult;

google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java

@@ -338,6 +338,39 @@ public void testBatchedMessagesWithOrderingKeyByDuration() throws Exception {
     publisher.shutdown();
   }
 
+  @Test
+  public void testLargeMessagesDoNotReorderBatches() throws Exception {
+    // Set the maximum batching size to 20 bytes.
+    Publisher publisher =
+        getTestPublisherBuilder()
+            .setBatchingSettings(
+                Publisher.Builder.DEFAULT_BATCHING_SETTINGS
+                    .toBuilder()
+                    .setElementCountThreshold(10L)
+                    .setRequestByteThreshold(20L)
+                    .setDelayThreshold(Duration.ofSeconds(100))
+                    .build())
+            .setEnableMessageOrdering(true)
+            .build();
+    testPublisherServiceImpl.setAutoPublishResponse(true);
+    ApiFuture<String> publishFuture1 = sendTestMessageWithOrderingKey(publisher, "m1", "OrderA");
+    ApiFuture<String> publishFuture2 = sendTestMessageWithOrderingKey(publisher, "m2", "OrderB");
+
+    assertFalse(publishFuture1.isDone());
+    assertFalse(publishFuture2.isDone());
+
+    ApiFuture<String> publishFuture3 =
+        sendTestMessageWithOrderingKey(publisher, "VeryLargeMessage", "OrderB");
+    // Verify that messages with "OrderB" were delivered in order.
+    assertTrue(Integer.parseInt(publishFuture2.get()) < Integer.parseInt(publishFuture3.get()));
+
+    assertTrue(publishFuture1.isDone());
+    assertTrue(publishFuture2.isDone());
+    assertTrue(publishFuture3.isDone());
+
+    publisher.shutdown();
+  }
+
   @Test
   public void testOrderingKeyWhenDisabled_throwsException() throws Exception {
     // Message ordering is disabled by default.