Simplify the logic reading session-id by making it final (#35822)

Author: anuchandy

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

@@ -338,9 +338,10 @@ private Flux<ServiceBusMessageContext> getSession(Scheduler scheduler, boolean d
                     return existing;
                 }
 
-                return new ServiceBusSessionReceiver(link, messageSerializer, connectionProcessor.getRetryOptions(),
+                final Duration idleTimeout = disposeOnIdle ? sessionIdleTimeout : null;
+                return new ServiceBusSessionReceiver(sessionId, link, messageSerializer, connectionProcessor.getRetryOptions(),
                     receiverOptions.getPrefetchCount(), scheduler, this::renewSessionLock,
-                    maxSessionLockRenewDuration, disposeOnIdle ? sessionIdleTimeout : null);
+                    maxSessionLockRenewDuration, idleTimeout);
             })))
             .flatMapMany(sessionReceiver -> sessionReceiver.receive().doFinally(signalType -> {
                 LOGGER.atVerbose()

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

@@ -29,7 +29,6 @@
 
 import static com.azure.core.amqp.implementation.ClientConstants.ENTITY_PATH_KEY;
 import static com.azure.core.amqp.implementation.ClientConstants.LINK_NAME_KEY;
-import static com.azure.core.amqp.implementation.ClientConstants.NOT_APPLICABLE;
 import static com.azure.messaging.servicebus.implementation.ServiceBusConstants.MESSAGE_ID_LOGGING_KEY;
 import static com.azure.messaging.servicebus.implementation.ServiceBusConstants.SESSION_ID_KEY;
 
@@ -47,8 +46,8 @@ class ServiceBusSessionReceiver implements AsyncCloseable, AutoCloseable {
     // also, the lock is removed after the completion of the message disposition.
     private final LockContainer<OffsetDateTime> lockContainer;
     private final AtomicReference<OffsetDateTime> sessionLockedUntil = new AtomicReference<>();
-    private final AtomicReference<String> sessionId = new AtomicReference<>();
     private final AtomicReference<LockRenewalOperation> renewalOperation = new AtomicReference<>();
+    private final String sessionId;
     private final ServiceBusReceiveLink receiveLink;
     private final Disposable.Composite subscriptions;
     private final Flux<ServiceBusMessageContext> receivedMessages;
@@ -61,6 +60,7 @@ class ServiceBusSessionReceiver implements AsyncCloseable, AutoCloseable {
      * Creates a receiver for the first available session.
      *
      * @param receiveLink Service Bus receive link for available session.
+     * @param sessionId Identifier of the Service Bus Session that the receiver is associated with.
      * @param messageSerializer Serializes and deserializes messages from Service Bus.
      * @param retryOptions Retry options for the receiver.
      * @param prefetch Number of messages to prefetch from session.
@@ -71,10 +71,11 @@ class ServiceBusSessionReceiver implements AsyncCloseable, AutoCloseable {
      * @param sessionIdleTimeout Timeout after which session receiver will be disposed if there are no more messages
      *                           and the receiver is idle. Set it to {@code null} to not dispose receiver.
      */
-    ServiceBusSessionReceiver(ServiceBusReceiveLink receiveLink, MessageSerializer messageSerializer,
+    ServiceBusSessionReceiver(String sessionId, ServiceBusReceiveLink receiveLink, MessageSerializer messageSerializer,
         AmqpRetryOptions retryOptions, int prefetch, Scheduler scheduler,
         Function<String, Mono<OffsetDateTime>> renewSessionLock, Duration maxSessionLockRenewDuration, Duration sessionIdleTimeout) {
 
+        this.sessionId = sessionId;
         this.receiveLink = receiveLink;
         this.lockContainer = new LockContainer<>(ServiceBusConstants.OPERATION_TIMEOUT);
         this.retryOptions = retryOptions;
@@ -157,14 +158,6 @@ class ServiceBusSessionReceiver implements AsyncCloseable, AutoCloseable {
                 }));
         }
 
-        this.subscriptions.add(receiveLink.getSessionId().subscribe(id -> {
-            if (!sessionId.compareAndSet(null, id)) {
-                LOGGER.atWarning()
-                    .addKeyValue("existingSessionId", sessionId.get())
-                    .addKeyValue("returnedSessionId", id)
-                    .log("Another method set sessionId.");
-            }
-        }));
         this.subscriptions.add(receiveLink.getSessionLockedUntil().subscribe(lockedUntil -> {
             if (!sessionLockedUntil.compareAndSet(null, lockedUntil)) {
                 withReceiveLinkInformation(LOGGER.atInfo())
@@ -174,7 +167,7 @@ class ServiceBusSessionReceiver implements AsyncCloseable, AutoCloseable {
 
                 return;
             }
-            this.renewalOperation.compareAndSet(null, new LockRenewalOperation(sessionId.get(),
+            this.renewalOperation.compareAndSet(null, new LockRenewalOperation(sessionId,
                 maxSessionLockRenewDuration, true, renewSessionLock, lockedUntil));
         }));
     }
@@ -204,7 +197,7 @@ String getLinkName() {
     }
 
     String getSessionId() {
-        return sessionId.get();
+        return sessionId;
     }
 
     /**
@@ -254,9 +247,7 @@ public void close() {
     }
 
     private LoggingEventBuilder withReceiveLinkInformation(LoggingEventBuilder builder) {
-        final String current = sessionId.get();
-
-        return builder.addKeyValue(SESSION_ID_KEY, current != null ? current : NOT_APPLICABLE)
+        return builder.addKeyValue(SESSION_ID_KEY, sessionId)
             .addKeyValue(ENTITY_PATH_KEY, receiveLink.getEntityPath())
             .addKeyValue(LINK_NAME_KEY, receiveLink.getLinkName());
     }

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

@@ -104,7 +104,7 @@ public void getsProperties() {
         final Duration maxSessionRenewalDuration = Duration.ofMinutes(5);
 
         // Act
-        final ServiceBusSessionReceiver sessionReceiver = new ServiceBusSessionReceiver(amqpReceiveLink,
+        final ServiceBusSessionReceiver sessionReceiver = new ServiceBusSessionReceiver(sessionId, amqpReceiveLink,
             messageSerializer, retryOptions, 1, scheduler,
             unused -> renewSessionLock(Duration.ofMinutes(1)), maxSessionRenewalDuration, NO_SESSION_IDLE_TIMEOUT);
 
@@ -152,7 +152,7 @@ public void receivesMessages() {
         final AmqpRetryOptions retryOptions = new AmqpRetryOptions();
         final Scheduler scheduler = Schedulers.boundedElastic();
         final Duration maxSessionRenewalDuration = Duration.ofMinutes(5);
-        final ServiceBusSessionReceiver sessionReceiver = new ServiceBusSessionReceiver(amqpReceiveLink,
+        final ServiceBusSessionReceiver sessionReceiver = new ServiceBusSessionReceiver(sessionId, amqpReceiveLink,
             messageSerializer, retryOptions, 1, scheduler,
             unused -> renewSessionLock(Duration.ofMinutes(1)), maxSessionRenewalDuration, NO_SESSION_IDLE_TIMEOUT);
 
@@ -222,7 +222,7 @@ public void disposesOnIdle() {
         final AmqpRetryOptions retryOptions = new AmqpRetryOptions().setTryTimeout(Duration.ofMinutes(10));
         final Scheduler scheduler = Schedulers.boundedElastic();
         final Duration maxSessionRenewalDuration = Duration.ofMinutes(5);
-        final ServiceBusSessionReceiver sessionReceiver = new ServiceBusSessionReceiver(amqpReceiveLink,
+        final ServiceBusSessionReceiver sessionReceiver = new ServiceBusSessionReceiver(sessionId, amqpReceiveLink,
             messageSerializer, retryOptions, 1, scheduler,
             unused -> renewSessionLock(Duration.ofMinutes(1)), maxSessionRenewalDuration, waitTime);
 
@@ -281,7 +281,7 @@ public void removesLockOnUpdateDisposition() {
         final AmqpRetryOptions retryOptions = new AmqpRetryOptions();
         final Scheduler scheduler = Schedulers.boundedElastic();
         final Duration maxSessionRenewalDuration = Duration.ofMinutes(5);
-        final ServiceBusSessionReceiver sessionReceiver = new ServiceBusSessionReceiver(amqpReceiveLink,
+        final ServiceBusSessionReceiver sessionReceiver = new ServiceBusSessionReceiver(sessionId, amqpReceiveLink,
             messageSerializer, retryOptions, 1, scheduler,
             unused -> renewSessionLock(Duration.ofMinutes(1)), maxSessionRenewalDuration, NO_SESSION_IDLE_TIMEOUT);