maxConcurentCallsPerSession replacement when upgrading to Service Bus SDK v7?

[danilobanjac]

Query/Question
What is the equivalent of maxConcurrentCallsPerSession for the new Azure Service Bus SDK? I've tried using maxConcurrentCalls, but it seems like it is not working. Thanks!

Why is this not a Bug or a feature Request?
It might be a feature request after my question get's answered.

Setup (please complete the following information if applicable):

• OS: [e.g. iOS]
• IDE: [e.g. IntelliJ]
• Library/Libraries: com.azure:azure-messaging-servicebus:7.13.0

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

• [ x ] Query Added
• [ x ] Setup information Added

[joshfree]

@ZejiaJiang could you please follow up with @danilobanjac on this question?

[ZejiaJiang]

Hi @danilobanjac ,

we do use maxConcurrentCalls in latest SDK(which is V7), this parameter is used to receive message from all session like its naming. So the behavior is different from the lagacy SDK. I'd like to know what's your scenario and help you to use SDK.

[danilobanjac]

Hi @ZejiaJiang,

I'm trying to achieve the following: I want to receive and process sessions concurrently, but the messages within the session are to be sequential. I've tried different combinations and could not come up with the solution. Therefore I've introduced locking, which I would like to avoid.

[ZejiaJiang]

Hi @danilobanjac ,

It seems like a very basic scenario, could you provide some code snippet? And how many sessions do you have?

[danilobanjac]

This is the session processor I've been using:

new ServiceBusClientBuilder()
      .connectionString(
          azureServiceBusConfig.getConnectionString()
      )
      // Give priority to the library's retry mechanism
      .retryOptions(new AmqpRetryOptions().setMaxRetries(0))
      .sessionProcessor()
      .disableAutoComplete()
      .maxConcurrentCalls(1)
      .maxConcurrentSessions(100)
      .maxAutoLockRenewDuration(
          Duration.ofSeconds(30)
      )
      .prefetchCount(0)
      .receiveMode(ServiceBusReceiveMode.PEEK_LOCK)
      .topicName(azureServiceBusConfig.getAlertsTopicName())
      .subscriptionName(azureServiceBusConfig.getAlertsTopicSubscriptionName())
      .processMessage(alertMessageProcessor::processMessage)
      .processError(alertMessageProcessor::processError)
      .buildProcessorClient();

[danilobanjac]

I've tried also different combinations for maxConcurrentCalls and maxConcurrentSessions but I always get messages within the session also in a concurrent mode. This is causing a race condition in my code.

[ZejiaJiang]

Thank you @danilobanjac , I got your point, you mean that even under a same session, the message are not sequential, thats cause a race condition in your code.

I'll try the code. BTW, I see that you set 100 sessions in the code, do you have 100 sessions? just want to ensure this to repro this issue.

[danilobanjac]

Correct, I use client ID for the sessionId and I have a lot of clients, that is why I set it to 100. Hope that is fine.

[ZejiaJiang]

Thank you. I'll try to repro the sequential issue in one session first.

[ZejiaJiang]

Hi @danilobanjac ,

After testing and discussing with my coworker, the session processor process message with maxConcurrentSessions > 1 has ordering issue is a known issue. My coworker did this before, and found that code need to be changed a lot and add more error handlings. Thus, we haven't fixed this yet. I'm trying to fix this base on my coworker's effort. It need some time to refactor related code and test new code.

I'm sorry for this issue and here is a workaround, you can create a ServiceBusSessionReceiverAsyncClient for each sessionId and you need to manage all receiver clients by yourself. or use ServiceBusProcessorClient with maxConcurrentSessions is 1.

[brunomace]

Thanks for your effort @ZejiaJiang .
I've the same problem as @danilobanjac . I use the sessions to have a FIFO Subscription, but if I set the maxConcurrentCalls > 1, my messages listener gets called by multiple threads for the same session. I must set the maxConcurrentCalls=1 and maxConcurrentSessions=1 but it is very inefficient since I may have a lot of sessions at the same time.

[conniey]

This is a known issue. This is part of our work in: #33706

[purijatin]

We are also stuck on this issue.

We want to concurrently process multiple sessions. But within each session, process message one after the other (we dont care whether sequence is maintained or not. Just that it should not be concurrent).

            ServiceBusProcessorClient processorClient = new ServiceBusClientBuilder()
                    .credential(new DefaultAzureCredentialBuilder().build())
                    .fullyQualifiedNamespace("xxxxx")
                    .sessionProcessor()
                    .queueName("xxxx")
                    .maxConcurrentSessions(5)
                    .maxConcurrentCalls(1)
                    .maxAutoLockRenewDuration(Duration.ofMinutes(100))
                    .sessionIdleTimeout(Duration.ofMinutes(100))

I expected 5 sessions to be concurrently processed. But do not process concurrently within each session.

It appears to me maxConcurrentCallsPerSession from legacy SDK would have solved the issue.

[purijatin]

Reaching out on this :)

[brunomace]

Seems to be fixed in the azure-messaging-servicebus 7.15.0 release. At least it works for my application.
See https://learn.microsoft.com/en-us/azure/developer/java/sdk/troubleshooting-messaging-service-bus-overview#upgrade-to-715x

[anuchandy]

Yes, this is addressed in the 7.15.x version, closing.