Use singleton rather than hosted service for cleanup

sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Config/ServiceBusExtensionConfigProvider.cs

@@ -48,7 +48,8 @@ public ServiceBusExtensionConfigProvider(
             IConverterManager converterManager,
             ServiceBusClientFactory clientFactory,
             ConcurrencyManager concurrencyManager,
-            IDrainModeManager drainModeManager)
+            IDrainModeManager drainModeManager,
+            CleanupService cleanupService)
         {
             _options = options.Value;
             _messagingProvider = messagingProvider;

sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Config/ServiceBusHostBuilderExtensions.cs

@@ -110,7 +110,7 @@ public static IWebJobsBuilder AddServiceBus(this IWebJobsBuilder builder, Action
 
             builder.Services.AddAzureClientsCore();
             builder.Services.TryAddSingleton<MessagingProvider>();
-            builder.Services.AddHostedService<CachedClientCleanupService>();
+            builder.Services.AddSingleton<CleanupService>();
             builder.Services.AddSingleton<ServiceBusClientFactory>();
             #if NET6_0_OR_GREATER
             builder.Services.AddSingleton<SettlementService>();

sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Primitives/CleanupService.cs

@@ -0,0 +1,25 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Hosting;
+
+namespace Microsoft.Azure.WebJobs.ServiceBus
+{
+    internal class CleanupService : IAsyncDisposable
+    {
+        private readonly MessagingProvider _provider;
+
+        public CleanupService(MessagingProvider provider)
+        {
+            _provider = provider;
+        }
+
+        public async ValueTask DisposeAsync()
+        {
+            await _provider.DisposeAsync().ConfigureAwait(false);
+        }
+    }
+}

sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Primitives/MessagingProvider.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Concurrent;
+using System.Threading.Tasks;
 using Azure.Core;
 using Azure.Messaging.ServiceBus;
 using Microsoft.Azure.WebJobs.ServiceBus.Listeners;
@@ -218,5 +219,29 @@ private static string GenerateCacheKey(string fullyQualifiedNamespace, string en
         {
             return $"{fullyQualifiedNamespace}/{entityPath}";
         }
+
+        // This class does not implement IAsyncDisposable as doing so could break existing user code. We can consider making this break
+        // on the next major version upgrade.
+        internal async Task DisposeAsync()
+        {
+            foreach (var receiver in MessageReceiverCache.Values)
+            {
+                await receiver.DisposeAsync().ConfigureAwait(false);
+            }
+            MessageReceiverCache.Clear();
+
+            foreach (var sender in MessageSenderCache.Values)
+            {
+                await sender.DisposeAsync().ConfigureAwait(false);
+            }
+            MessageSenderCache.Clear();
+
+            foreach (var client in ClientCache.Values)
+            {
+                await client.DisposeAsync().ConfigureAwait(false);
+            }
+            ClientCache.Clear();
+            ActionsCache.Clear();
+        }
     }
 }

sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/tests/ServiceBusEndToEndTests.cs

@@ -619,6 +619,8 @@ public async Task TestBatch_DataContractPoco()
         public async Task BindToPoco()
         {
             var host = BuildHost<ServiceBusArgumentBindingJob>();
+            var provider = host.Services.GetService<MessagingProvider>();
+
             using (host)
             {
                 await WriteQueueMessage("{ Name: 'foo', Value: 'bar' }");
@@ -630,6 +632,10 @@ public async Task BindToPoco()
                 Assert.Contains("PocoValues(foo,bar)", logs);
                 await host.StopAsync();
             }
+            Assert.AreEqual(0, provider.ClientCache.Count);
+            Assert.AreEqual(0, provider.MessageReceiverCache.Count);
+            Assert.AreEqual(0, provider.MessageSenderCache.Count);
+            Assert.AreEqual(0, provider.ActionsCache.Count);
         }
 
         [Test]

sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/tests/WebJobsServiceBusTestBase.cs

@@ -336,12 +336,6 @@ public async Task StopAsync(CancellationToken cancellationToken)
 
                 QueueRuntimeProperties properties = await client.GetQueueRuntimePropertiesAsync(FirstQueueScope.QueueName, CancellationToken.None);
                 Assert.AreEqual(ExpectedRemainingMessages, properties.ActiveMessageCount);
-
-                var provider = _host.Services.GetService<MessagingProvider>();
-                Assert.AreEqual(0, provider.ClientCache.Count);
-                Assert.AreEqual(0, provider.MessageReceiverCache.Count);
-                Assert.AreEqual(0, provider.MessageSenderCache.Count);
-                Assert.AreEqual(0, provider.ActionsCache.Count);
             }
 
             private static bool IsError(LogMessage logMessage)