Fixing ordering problem when sending messages in a batch (Azure#27366)

* batch ordering fix

Author: ki1729

sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md

@@ -8,6 +8,7 @@
 
 ### Bugs Fixed
 - Removed the incorrect use of lock primitives from `ServiceBusMessageBatch.tryAddMessage()` implementation and documented that this API is not thread-safe. ([#25910](https://github.com/Azure/azure-sdk-for-java/issues/25910))
+- Fixed incorrect ordering of message when sent as a batch. ([#25112](https://github.com/Azure/azure-sdk-for-java/issues/25112), [#25599](https://github.com/Azure/azure-sdk-for-java/issues/25599))
 
 ### Other Changes
 

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClient.java

@@ -657,7 +657,7 @@ private Mono<Void> sendIterable(Iterable<ServiceBusMessage> messages, ServiceBus
         }
 
         return createMessageBatch().flatMap(messageBatch -> {
-            StreamSupport.stream(messages.spliterator(), true)
+            StreamSupport.stream(messages.spliterator(), false)
                 .forEach(message -> messageBatch.tryAddMessage(message));
             return sendInternal(messageBatch, transaction);
         });
@@ -723,7 +723,7 @@ private Mono<Void> sendInternal(ServiceBusMessageBatch batch, ServiceBusTransact
 
         AtomicReference<Context> sharedContext = new AtomicReference<>(Context.NONE);
         final List<org.apache.qpid.proton.message.Message> messages = Collections.synchronizedList(new ArrayList<>());
-        batch.getMessages().parallelStream().forEach(serviceBusMessage -> {
+        batch.getMessages().forEach(serviceBusMessage -> {
             if (isTracingEnabled) {
                 parentContext.set(serviceBusMessage.getContext());
                 if (sharedContext.get().equals(Context.NONE)) {

sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSenderAsyncClientTest.java

@@ -56,9 +56,12 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import java.util.stream.IntStream;
 
 import static com.azure.core.util.tracing.Tracer.AZ_TRACING_NAMESPACE_KEY;
@@ -487,7 +490,6 @@ void sendMessagesListWithTransaction() {
     void sendMessagesList() {
         // Arrange
         final int count = 4;
-        final byte[] contents = TEST_CONTENTS.toBytes();
         final List<ServiceBusMessage> messages = TestUtils.getServiceBusMessages(count, UUID.randomUUID().toString());
 
         when(connection.createSendLink(eq(ENTITY_NAME), eq(ENTITY_NAME), eq(retryOptions), isNull()))
@@ -701,6 +703,48 @@ void cancelScheduleMessages() {
         Assertions.assertEquals(sequenceNumbers.size(), actualTotal.get());
     }
 
+    /**
+     * Verifies that sending multiple message will result in calling sender.send(Message...).
+     */
+    @Test
+    void verifyMessageOrdering() {
+        // Arrange
+        final ServiceBusMessage firstMessage = new ServiceBusMessage("First message " + UUID.randomUUID());
+        final ServiceBusMessage secondMessage = new ServiceBusMessage("Second message " + UUID.randomUUID());
+        final ServiceBusMessage thirdMessage = new ServiceBusMessage("Third message " + UUID.randomUUID());
+        final ServiceBusMessage fourthMessage = new ServiceBusMessage("Fourth message " + UUID.randomUUID());
+        final ServiceBusMessage fifthMessage = new ServiceBusMessage("Fifth message " + UUID.randomUUID());
+        final List<ServiceBusMessage> messages = new ArrayList<>();
+        messages.add(firstMessage);
+        messages.add(secondMessage);
+        messages.add(thirdMessage);
+        messages.add(fourthMessage);
+        messages.add(fifthMessage);
+
+        when(connection.createSendLink(eq(ENTITY_NAME), eq(ENTITY_NAME), eq(retryOptions), isNull()))
+            .thenReturn(Mono.just(sendLink));
+        when(sendLink.send(anyList())).thenReturn(Mono.empty());
+
+        // Act
+        StepVerifier.create(sender.sendMessages(messages))
+            .verifyComplete();
+
+        // Assert
+        verify(sendLink).send(messagesCaptor.capture());
+
+        final List<Message> messagesSent = messagesCaptor.getValue();
+        Assertions.assertEquals(messages.size(), messagesSent.size());
+
+        Iterator<ServiceBusMessage> iterator = messages.iterator();
+        Pattern regex = Pattern.compile("\\{(.*)\\}");
+        for (Message message : messagesSent) {
+            Matcher matcher = regex.matcher(message.getBody().toString());
+            String content = matcher.find() ? matcher.group(1) : "";
+            Assertions.assertEquals(content, iterator.next().getBody().toString());
+        }
+        messagesSent.forEach(message -> Assertions.assertEquals(Section.SectionType.Data, message.getBody().getType()));
+    }
+
     /**
      * Verifies that the onClientClose is called.
      */