Remove lock primitives from tryAddMessage() (Azure#27038)

* Change atomic to primitive types

* Remove unused lock

* Change LinkedList to ArrayList

* Revert import changes

* Update CHANGELOG

* Update CHANGELOG

Author: liukun-msft

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

@@ -7,6 +7,7 @@
 ### Breaking Changes
 
 ### 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))
 
 ### Other Changes
 

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

@@ -9,14 +9,13 @@
 import com.azure.core.amqp.implementation.MessageSerializer;
 import com.azure.core.amqp.implementation.TracerProvider;
 import com.azure.core.util.logging.ClientLogger;
+import org.apache.qpid.proton.message.Message;
 
 import java.nio.BufferOverflowException;
-import java.util.Collections;
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import static com.azure.messaging.servicebus.implementation.MessageUtils.traceMessageSpan;
 
@@ -26,13 +25,12 @@
  */
 public final class ServiceBusMessageBatch {
     private final ClientLogger logger = new ClientLogger(ServiceBusMessageBatch.class);
-    private final Object lock = new Object();
     private final int maxMessageSize;
     private final ErrorContextProvider contextProvider;
     private final MessageSerializer serializer;
     private final List<ServiceBusMessage> serviceBusMessageList;
     private final byte[] eventBytes;
-    private final AtomicInteger sizeInBytes;
+    private int sizeInBytes;
     private final TracerProvider tracerProvider;
     private final String entityPath;
     private final String hostname;
@@ -42,8 +40,8 @@ public final class ServiceBusMessageBatch {
         this.maxMessageSize = maxMessageSize;
         this.contextProvider = contextProvider;
         this.serializer = serializer;
-        this.serviceBusMessageList = Collections.synchronizedList(new LinkedList<>());
-        this.sizeInBytes = new AtomicInteger((maxMessageSize / 65536) * 1024); // reserve 1KB for every 64KB
+        this.serviceBusMessageList = new ArrayList<>();
+        this.sizeInBytes = (maxMessageSize / 65536) * 1024; // reserve 1KB for every 64KB
         this.eventBytes = new byte[maxMessageSize];
         this.tracerProvider = tracerProvider;
         this.entityPath = entityPath;
@@ -74,12 +72,14 @@ public int getMaxSizeInBytes() {
      * @return The size of the {@link ServiceBusMessageBatch batch} in bytes.
      */
     public int getSizeInBytes() {
-        return this.sizeInBytes.get();
+        return this.sizeInBytes;
     }
 
     /**
      * Tries to add an {@link ServiceBusMessage message} to the batch.
      *
+     * <p>This method is thread-unsafe.</p>
+     *
      * @param serviceBusMessage The {@link ServiceBusMessage} to add to the batch.
      *
      * @return {@code true} if the message could be added to the batch; {@code false} if the event was too large to fit
@@ -99,9 +99,9 @@ public boolean tryAddMessage(final ServiceBusMessage serviceBusMessage) {
                 tracerProvider)
                 : serviceBusMessage;
 
-        final AtomicInteger size = new AtomicInteger();
+        final int size;
         try {
-            size.set(getSize(serviceBusMessageUpdated, serviceBusMessageList.isEmpty()));
+            size = getSize(serviceBusMessageUpdated, serviceBusMessageList.isEmpty());
         } catch (BufferOverflowException exception) {
             final RuntimeException ex = new ServiceBusException(
                     new AmqpException(false, AmqpErrorCondition.LINK_PAYLOAD_SIZE_EXCEEDED,
@@ -111,11 +111,11 @@ public boolean tryAddMessage(final ServiceBusMessage serviceBusMessage) {
             throw logger.logExceptionAsWarning(ex);
         }
 
-        if (this.sizeInBytes.addAndGet(size.get()) > this.maxMessageSize) {
-            this.sizeInBytes.addAndGet(-1 * size.get());
+        if (this.sizeInBytes + size > this.maxMessageSize) {
             return false;
         }
 
+        this.sizeInBytes += size;
         this.serviceBusMessageList.add(serviceBusMessageUpdated);
         return true;
     }
@@ -132,7 +132,7 @@ List<ServiceBusMessage> getMessages() {
     private int getSize(final ServiceBusMessage serviceBusMessage, final boolean isFirst) {
         Objects.requireNonNull(serviceBusMessage, "'serviceBusMessage' cannot be null.");
 
-        final org.apache.qpid.proton.message.Message amqpMessage = serializer.serialize(serviceBusMessage);
+        final Message amqpMessage = serializer.serialize(serviceBusMessage);
         int eventSize = amqpMessage.encode(this.eventBytes, 0, maxMessageSize); // actual encoded bytes size
         eventSize += 16; // data section overhead