---

kimkyung-goog · kolea2 · commit bfe6124e0f82 · 2019-03-04T07:49:34.000-07:00
yaml --- r: 20657 b: refs/heads/pubsub-ordering-keys c: 858d4e9 h: refs/heads/master i: 20655: 7eca050
diff --git a/[refs] b/[refs]
@@ -157,7 +157,7 @@ refs/tags/v0.72.0: a7703f2593ba312c0b2dde6fdfd4f5c764bb55ac
 refs/tags/v0.73.0: 21241ea8be9439cc5764c4944cdce21d34ce4f9e
 refs/tags/v0.74.0: 9d1f733dbbf790de7b494418523b69c4a9a57638
 refs/heads/ignoretest: 23c412ae07af3d0ab1caa2d44d5bc5c0ccb8b31d
-refs/heads/pubsub-ordering-keys: fab0188fc02390be1c7a12051b37e722e82fcd27
+refs/heads/pubsub-ordering-keys: 858d4e986a0ba48e08f00d42f51cbdecb175f5d6
 "refs/heads/update_mvn_badge": ae2d773814db0f71197ccf5a8612ee1d8056f8de
 refs/tags/v0.75.0: c3673089ae09a897c1b4cf7dfe167fe4f8ab32fb
 refs/tags/v0.76.0: 395b016826d3ddf9cb8b34919636df15a4dbd032
diff --git a/branches/pubsub-ordering-keys/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java b/branches/pubsub-ordering-keys/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;
diff --git a/branches/pubsub-ordering-keys/google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/PublisherImplTest.java b/branches/pubsub-ordering-keys/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.
