Fix incorrect message ordering after session processor abandon message and when maxConcurrentSessions is 1 (Azure#29696)

liukun-msft · web-flow · commit c7880d313d1b · 2022-07-22T15:35:15.000+08:00
* Remove fuseable publishOn and set request=1 when doOnRequest

* Update CHANGELOG.md
diff --git a/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md b/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md
@@ -7,7 +7,7 @@
 ### Breaking Changes
 
 ### Bugs Fixed
-
+- Fixed incorrect message ordering after session processor abandon message and when `maxConcurrentSessions` is 1. ([#24064](https://github.com/Azure/azure-sdk-for-java/issues/24064), [#30027](https://github.com/Azure/azure-sdk-for-java/issues/30027))
 ### Other Changes
 
 ## 7.10.0 (2022-07-12)
diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionManager.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionManager.java
@@ -324,8 +324,7 @@ private Flux<ServiceBusMessageContext> getSession(Scheduler scheduler, boolean d
                 if (receiverOptions.isRollingSessionReceiver()) {
                     onSessionRequest(1L);
                 }
-            }))
-            .publishOn(scheduler, 1);
+            }));
     }
 
     private Mono<ServiceBusManagementNode> getManagementNode() {
diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiver.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiver.java
@@ -94,6 +94,7 @@ class ServiceBusSessionReceiver implements AsyncCloseable, AutoCloseable {
                     receiveLink.addCredits(Math.max(0, prefetch - receiveLink.getCredits())).subscribe();
                 }
             })
+            .limitRate(1)
             .takeUntilOther(cancelReceiveProcessor)
             .map(message -> {
                 final ServiceBusReceivedMessage deserialized = messageSerializer.deserialize(message,

Original file line number	Diff line number	Diff line change
`@@ -324,8 +324,7 @@ private Flux<ServiceBusMessageContext> getSession(Scheduler scheduler, boolean d`
`324`	`324`	`if (receiverOptions.isRollingSessionReceiver()) {`
`325`	`325`	`onSessionRequest(1L);`
`326`	`326`	`}`
`327`		`- }))`
`328`		`- .publishOn(scheduler, 1);`
	`327`	`+ }));`
`329`	`328`	`}`
`330`	`329`
`331`	`330`	`private Mono<ServiceBusManagementNode> getManagementNode() {`
Original file line number	Diff line number	Diff line change
`@@ -94,6 +94,7 @@ class ServiceBusSessionReceiver implements AsyncCloseable, AutoCloseable {`
`94`	`94`	`receiveLink.addCredits(Math.max(0, prefetch - receiveLink.getCredits())).subscribe();`
`95`	`95`	`}`
`96`	`96`	`})`
	`97`	`+ .limitRate(1)`
`97`	`98`	`.takeUntilOther(cancelReceiveProcessor)`
`98`	`99`	`.map(message -> {`
`99`	`100`	`final ServiceBusReceivedMessage deserialized = messageSerializer.deserialize(message,`