[BUG] ServiceBusProcessorClient not obeying maxConcurrentSessions

[ankitrajsingh]

Describe the bug
MaxConcurrentSessions is not being obeyed by session based ServiceBusProcessorClient. It does appears to lock onto multiple sessions but processes only 1 message from any one session at a given time (Single Threaded). Expectation is to process 1 message from each of the locked session concurrently as per the SDK Doc and Github Readme.

Exception or Stack Trace
NA

To Reproduce
Steps to reproduce the behavior:

1. Push messages corresponding to multiple sessions (more than 1 session) in the ASB session queue via any sender. Ensure that each session have at least 10 messages.
2. Create message callback for ServiceBusProcessorClient containing logs describing the current session.
3. Create a Session based ServiceBusProcessorClient with maxConcurrentSession higher than 1 along with the above callback.
4. Start the client.

Code Snippet
Add the code snippet that causes the issue.

Consumer<ServiceBusReceivedMessageContext> onMessage = context -> {
      ServiceBusReceivedMessage message = context.getMessage();
      log.info(message.getSessionId() + " message processed");
      Thread.sleep(1000);
      context.complete();
    };

Consumer<ServiceBusErrorContext> onError = context -> {
  System.out.printf("Error when receiving messages from namespace: '%s'. Entity: '%s'%n",
                    context.getFullyQualifiedNamespace(), context.getEntityPath());

  if (context.getException() instanceof ServiceBusException) {
    ServiceBusException exception = (ServiceBusException) context.getException();
    System.out.printf("Error source: %s, reason %s%n", context.getErrorSource(),
                      exception.getReason());
  } else {
    System.out.printf("Error occurred: %s%n", context.getException());
  }
};

ServiceBusProcessorClient sessionProcessor = new ServiceBusClientBuilder()
        .connectionString("<<conn_string>>")
        .sessionProcessor()
        .queueName("<<session_queue_name>>")
        .receiveMode(ServiceBusReceiveMode.PEEK_LOCK)
        .maxConcurrentSessions(10)
        .processMessage(onMessage)
        .processError(onError)
        .disableAutoComplete()
        .buildProcessorClient();

// Start the processor in the background
sessionProcessor.start();

Expected behavior
Expected behaviour is that the client should process 1 message from each of the sessions concurrently when supplied with maxConcurrentSessions and locked onto multiple sessions.

For example,
Scenario: Multiple sessions say session1, session2, session3, etc are present in ASB and maxConcurrentSessions is set to 3 in the client.

Expectation: Client will process 1 message from each of the session1, session2 and session3 concurrently. Throughput = at most 3 messages (each belonging to different sessions) at a given time

Actual: Client is processing only 1 message from session1 alone at a given time and afterwards repeating the same process with messages from session2 sequentially. Throughput = 1 message at a given time

Screenshots
Logs describing only 1 message from any one session is being processed at a given time although multiple sessions are available. Thread name under which all messages are being processed is same i.e [oundedElastic-2] which signifies single threaded sequential execution.

Setup (please complete the following information):

• OS: MacOS
• IDE: IntelliJ
• Library/Libraries: com.azure:azure-messaging-servicebus:7.4.0
• Java version: 11
• App Server/Environment: Tomcat
• Frameworks: Spring Boot

If you suspect a dependency version mismatch (e.g. you see NoClassDefFoundError, NoSuchMethodError or similar), please provide

• dependency tree (mvn dependency:tree)
  NA

Additional context
Client does appear to process messages from sessions concurrently when MaxConcurrentCalls are set along with MaxConcurrentSessions in the session based ServiceBusProcessorClient. But this also causes parallel processing of messages from the same session thereby disrupting the ordered nature of messages within a session.

Information Checklist
Kindly make sure that you have added all the following information above and checkoff the required fields otherwise we will treat the issuer as an incomplete report

• Bug Description Added
• Repro Steps Added
• Setup information Added

[joshfree]

@anuchandy could you please follow up with @ankitrajsingh

[ankitrajsingh]

Hi @anuchandy, please let me know if something else is needed on the issue. Would be great if you can help us out. Thanks!

[anuchandy]

Hi @ankitrajsingh, thanks for sharing the details. I was able to create a repro and can see that the messages across sessions are delivered serially. I'll be looking into enabling the concurrency feature; it seems quite a bit of code needs to be changed and requires some discussion in behavior change. Given that the callbacks are not invoked serially, existing consumers need to prepare/change their handler to be thread-safe.

[ankitrajsingh]

Sure thanks 👍

[lidkowiak]

Hi @anuchandy, is there any progress on this issue?

[jwyseu]

@anuchandy An update would be great
@ramya-rao-a I see the march milestone is now closed so this should move to the April milestone?

[anuchandy]

Hi @jwyseu, I've updated the other issue 27336. Closing this issue as this is a duplicate.