Reimplement session-processor to remove the pattern using merge operator

[anuchandy]

Rework session-processor to -

1. Remove the pattern of using merge operator that contributes to "out-of-order" message delivery.
2. Remove the pattern of using parallel operator that result in undesired prefetch.
3. Ensure this credit-placement in ServiceBusSessionReceiver is good enough and not do not contribute to undesired prefetch.
4. Reduce the allocation by sharing one BoundedElastic Pool, rather than creating N (N==max-concurrency) number of Bounded Elastic pools .

Related issue: #33232

[clemlesne]

Any resolution ETA?

Customer end-user quality of service is strongly impacted due to the performance impact.

[clemlesne]

Behavior seems to have been fixed in the release azure-messaging-servicebus_7.15.0-beta.3.

To update, please override the transitive package version (beta.4 is the latest):

<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-messaging-servicebus</artifactId>
    <version>7.15.0-beta.4</version>
</dependency>

[anuchandy]

Hello Clemence, thanks for following up. You’re right the beta.4 addresses this ordering issue. The steps to use the new session implementation is –

Add 7.15.0-beta.4 dependency

<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-messaging-servicebus</artifactId>
    <version>7.15.0-beta.4</version>
</dependency>

Opt-in for new session (V2) functionality

private static final String V2_SESSION_PROCESSOR_CONFIGURATION_KEY = "com.azure.messaging.servicebus.session.processor.asyncReceive.v2";
ServiceBusProcessorClient processor = new ServiceBusClientBuilder()
        .fullyQualifiedNamespace(Config.FULLY_QUALIFIED_NAMESPACE)
        .credential(Config.CREDENTIAL)
        .configuration(new com.azure.core.util.ConfigurationBuilder()
          .putProperty(V2_SESSION_PROCESSOR_CONFIGURATION_KEY, "true") // 'false' by default, so opt-in to enable V2.
          .build())
        .sessionProcessor()
        .topicName(Config.TOPIC)
        .subscriptionName(Config.SUBSCRIPTION)
        .disableAutoComplete()
        .processMessage(processMessage)
        .processError(processError)
        .maxAutoLockRenewDuration(Duration.ofMinutes(5))
        .receiveMode(ServiceBusReceiveMode.PEEK_LOCK)
        .subQueue(SubQueue.NONE)
        .maxConcurrentSessions(2)
        .maxConcurrentCalls(1)
        .prefetchCount(0)
        .sessionIdleTimeout(Duration.ofMinutes(1))
        .buildProcessorClient();

Also, make sure the app developer understands concurrency and capacity aspects, to avoid stalling/hang/interruption. Refer this comment.

[brunomace]

Hello @anuchandy
Is it possible set this property under Spring Cloud ?
I don't see any property in https://learn.microsoft.com/en-us/azure/developer/java/spring-framework/configuration-properties-azure-service-bus or https://microsoft.github.io/spring-cloud-azure/current/reference/html/appendix.html#azure_service_bus_proeprties
Or maybe we need to override a Spring bean ?

Edit: it works by settings an environment variable "com.azure.messaging.servicebus.session.processor.asyncReceive.v2" at "true"

[anuchandy]

Hi @brunomace,

Opt-in can be also done using env-var (as you found) or via system property when Configuration route is not accessible. JFYI, 7.15.0 can be used instead of beta, opt-in is still needed.

Also, please have a look at this comment regarding the session concurrency.

[anuchandy]

Resolving this. Refer https://learn.microsoft.com/en-us/azure/developer/java/sdk/troubleshooting-messaging-service-bus-overview#upgrade-to-715x for details about version 7.15.x and how to opting-in new session rework.