Don't drain from session processor as it is not necessary for FIFO

sdk/servicebus/Azure.Messaging.ServiceBus/src/Amqp/AmqpReceiver.cs

@@ -369,8 +369,10 @@ public override int PrefetchCount
                     messagesReceived as IReadOnlyCollection<AmqpMessage> ?? messagesReceived?.ToList() ?? s_emptyAmqpMessageList;
 
                 // If this is a session receiver and we didn't receive all requested messages, we need to drain the credits
-                // to ensure FIFO ordering within each session.
-                if (_isSessionReceiver && messageList.Count < maxMessages)
+                // to ensure FIFO ordering within each session. We exclude session processor since those will always receive a single message
+                // at a time, and if there are no messages, the session will be closed. The session won't be closed in the case that
+                // MaxConcurrentCallsPerSession > 1, but in that case FIFO is not possible to guarantee.
+                if (_isSessionReceiver && !_isProcessor && messageList.Count < maxMessages)
                 {
                     await link.DrainAsyc(cancellationToken).ConfigureAwait(false);
                 }

sdk/servicebus/Azure.Messaging.ServiceBus/tests/Processor/SessionProcessorLiveTests.cs

@@ -2679,6 +2679,54 @@ async Task ProcessMessage(ProcessSessionMessageEventArgs args)
             }
         }
 
+        [Test]
+        public async Task SessionOrderingIsGuaranteedProcessor()
+        {
+            await using (var scope = await ServiceBusScope.CreateWithQueue(enablePartitioning: false, enableSession: true))
+            {
+                await using var client = new ServiceBusClient(TestEnvironment.ServiceBusConnectionString);
+                long lastSequenceNumber = 0;
+
+                await using var processor = client.CreateSessionProcessor(scope.QueueName,
+                    new ServiceBusSessionProcessorOptions
+                    {
+                        MaxConcurrentCallsPerSession = 1, MaxConcurrentSessions = 1
+                    });
+                processor.ProcessMessageAsync += ProcessMessage;
+                processor.ProcessErrorAsync += SessionErrorHandler;
+
+                var sender = client.CreateSender(scope.QueueName);
+
+                CancellationTokenSource cts = new CancellationTokenSource();
+                cts.CancelAfter(TimeSpan.FromSeconds(60));
+                await processor.StartProcessingAsync();
+                await SendMessagesAsync();
+
+                await processor.StopProcessingAsync();
+
+                async Task SendMessagesAsync()
+                {
+                    while (!cts.IsCancellationRequested)
+                    {
+                        await sender.SendMessageAsync(ServiceBusTestUtilities.GetMessage("session"));
+                        await Task.Delay(TimeSpan.FromMilliseconds(100));
+                    }
+                }
+
+                Task ProcessMessage(ProcessSessionMessageEventArgs args)
+                {
+                    if (args.Message.SequenceNumber != lastSequenceNumber + 1)
+                    {
+                        Assert.Fail(
+                            $"Last sequence number: {lastSequenceNumber}, current sequence number: {args.Message.SequenceNumber}");
+                    }
+
+                    lastSequenceNumber = args.Message.SequenceNumber;
+                    return Task.CompletedTask;
+                }
+            }
+        }
+
         private Task SessionErrorHandler(ProcessErrorEventArgs args)
         {
             // If the connection drops due to network flakiness