fix(servicebus): annotate async auto_lock_renewer with aio AutoLockRenewer (#47948)

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -11,6 +11,7 @@
 - Fixed a bug where messages returned by `receive_deferred_messages` had a `lock_token` of `None`, which prevented settling (completing, abandoning, dead-lettering, deferring) or renewing the lock on a deferred message in `PEEK_LOCK` mode. The lock token is now read from the `lock-token` field of the management-link response for deferred messages. ([#42454](https://github.com/Azure/azure-sdk-for-python/issues/42454))
 - Read `com.microsoft:max-message-batch-size` vendor property from the AMQP sender link to correctly limit batch size on Premium large-message entities, where `max-message-size` can be up to 100 MB but the batch limit is 1 MB.
 - Fixed a bug where sending a batched or multi-message payload with `uamqp_transport=True` raised `TypeError: 'BatchMessage' object is not subscriptable` (and a masked `AttributeError` on the list path) when the first message carried a `message_id`, `session_id`, or `partition_key`. The batch envelope properties are now set through the transport-appropriate code path. (regression from [#42598](https://github.com/Azure/azure-sdk-for-python/pull/42598))
+- Fixed a bug where the async receiver factory methods on `azure.servicebus.aio.ServiceBusClient` (`get_queue_receiver`, `get_subscription_receiver`) and the async `ServiceBusReceiver` annotated the `auto_lock_renewer` keyword with the synchronous `AutoLockRenewer`, causing static type checkers to reject the documented `azure.servicebus.aio.AutoLockRenewer` usage. The annotation now references the async `AutoLockRenewer`, matching the docstrings and runtime behavior. ([#47948](https://github.com/Azure/azure-sdk-for-python/issues/47948))
 
 ## 7.14.3 (2025-11-11)
 

sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_client_async.py

@@ -21,7 +21,6 @@
 from ._servicebus_sender_async import ServiceBusSender
 from ._servicebus_receiver_async import ServiceBusReceiver
 from .._common._configuration import Configuration
-from .._common.auto_lock_renewer import AutoLockRenewer
 from .._common.utils import generate_dead_letter_entity_name, strip_protocol_from_uri
 from .._common.constants import (
     ServiceBusSubQueue,
@@ -32,6 +31,7 @@
 
 if TYPE_CHECKING:
     from azure.core.credentials_async import AsyncTokenCredential
+    from ._async_auto_lock_renewer import AutoLockRenewer
 
 NextAvailableSessionType = Literal[ServiceBusSessionFilter.NEXT_AVAILABLE]
 
@@ -352,7 +352,7 @@ def get_queue_receiver(
         sub_queue: Optional[Union[ServiceBusSubQueue, str]] = None,
         receive_mode: Union[ServiceBusReceiveMode, str] = ServiceBusReceiveMode.PEEK_LOCK,
         max_wait_time: Optional[float] = None,
-        auto_lock_renewer: Optional[AutoLockRenewer] = None,
+        auto_lock_renewer: Optional["AutoLockRenewer"] = None,
         prefetch_count: int = 0,
         **kwargs: Any,
     ) -> ServiceBusReceiver:
@@ -539,7 +539,7 @@ def get_subscription_receiver(
         sub_queue: Optional[Union[ServiceBusSubQueue, str]] = None,
         receive_mode: Union[ServiceBusReceiveMode, str] = ServiceBusReceiveMode.PEEK_LOCK,
         max_wait_time: Optional[float] = None,
-        auto_lock_renewer: Optional[AutoLockRenewer] = None,
+        auto_lock_renewer: Optional["AutoLockRenewer"] = None,
         prefetch_count: int = 0,
         client_identifier: Optional[str] = None,
         socket_timeout: Optional[float] = None,

sdk/servicebus/azure-servicebus/azure/servicebus/aio/_servicebus_receiver_async.py

@@ -65,7 +65,7 @@
     from .._pyamqp.aio._authentication_async import JWTTokenAuthAsync as pyamqp_JWTTokenAuthAsync
     from azure.core.credentials_async import AsyncTokenCredential
     from azure.core.credentials import AzureSasCredential, AzureNamedKeyCredential
-    from .._common.auto_lock_renewer import AutoLockRenewer
+    from ._async_auto_lock_renewer import AutoLockRenewer
 
 _LOGGER = logging.getLogger(__name__)
 

sdk/servicebus/azure-servicebus/samples/async_samples/auto_lock_renew_async.py

@@ -9,6 +9,8 @@
 Example to show usage of AutoLockRenewer asynchronously:
     1. Automatically renew locks on messages received from non-sessionful entity
     2. Automatically renew locks on the session of sessionful entity
+    3. Automatically renew locks by passing the AutoLockRenewer to the receiver factory
+       via the ``auto_lock_renewer`` keyword
 
 We do not guarantee that this SDK is thread-safe. We do not recommend reusing the ServiceBusClient,
  ServiceBusSender, ServiceBusReceiver across threads. It is up to the running 
@@ -26,6 +28,8 @@
 FULLY_QUALIFIED_NAMESPACE = os.environ["SERVICEBUS_FULLY_QUALIFIED_NAMESPACE"]
 QUEUE_NAME = os.environ["SERVICEBUS_QUEUE_NAME"]
 SESSION_QUEUE_NAME = os.environ["SERVICEBUS_SESSION_QUEUE_NAME"]
+TOPIC_NAME = os.environ["SERVICEBUS_TOPIC_NAME"]
+SUBSCRIPTION_NAME = os.environ["SERVICEBUS_SUBSCRIPTION_NAME"]
 
 
 async def renew_lock_on_message_received_from_non_sessionful_entity():
@@ -58,6 +62,57 @@ async def renew_lock_on_message_received_from_non_sessionful_entity():
         await renewer.close()
 
 
+async def renew_lock_via_receiver_auto_lock_renewer_keyword() -> None:
+    # The `-> None` return annotation is intentional: it makes mypy type-check this
+    # function body (untyped bodies are skipped), so the auto_lock_renewer= calls below
+    # act as a static regression guard for both receiver-factory annotations
+    # (get_queue_receiver and get_subscription_receiver).
+    credential = DefaultAzureCredential()
+    servicebus_client = ServiceBusClient(FULLY_QUALIFIED_NAMESPACE, credential)
+
+    async with servicebus_client:
+        async with servicebus_client.get_queue_sender(queue_name=QUEUE_NAME) as sender:
+            await sender.send_messages(ServiceBusMessage("message"))
+            print("Send message to non-sessionful queue.")
+
+        # Pass the AutoLockRenewer directly to the receiver factory via the auto_lock_renewer
+        # keyword. Received messages are then automatically registered for lock renewal on
+        # receipt, without a manual renewer.register(...) call. The `async with` form ensures
+        # the renewer is shut down even if the receiver block raises.
+        async with AutoLockRenewer() as renewer:
+            async with servicebus_client.get_queue_receiver(
+                queue_name=QUEUE_NAME, auto_lock_renewer=renewer, prefetch_count=10
+            ) as receiver:
+                received_msgs = await receiver.receive_messages(max_message_count=10, max_wait_time=5)
+
+                await asyncio.sleep(100)  # message handling for long period (E.g. application logic)
+
+                for msg in received_msgs:
+                    await receiver.complete_message(msg)
+                print("Complete queue messages.")
+
+        async with servicebus_client.get_topic_sender(topic_name=TOPIC_NAME) as sender:
+            await sender.send_messages(ServiceBusMessage("message"))
+            print("Send message to topic.")
+
+        # The auto_lock_renewer keyword behaves the same on the subscription receiver factory,
+        # so the type check also covers the get_subscription_receiver annotation from issue #47948.
+        async with AutoLockRenewer() as renewer:
+            async with servicebus_client.get_subscription_receiver(
+                topic_name=TOPIC_NAME,
+                subscription_name=SUBSCRIPTION_NAME,
+                auto_lock_renewer=renewer,
+                prefetch_count=10,
+            ) as receiver:
+                received_msgs = await receiver.receive_messages(max_message_count=10, max_wait_time=5)
+
+                await asyncio.sleep(100)  # message handling for long period (E.g. application logic)
+
+                for msg in received_msgs:
+                    await receiver.complete_message(msg)
+                print("Complete subscription messages.")
+
+
 async def renew_lock_on_session_of_the_sessionful_entity():
     credential = DefaultAzureCredential()
     servicebus_client = ServiceBusClient(FULLY_QUALIFIED_NAMESPACE, credential)
@@ -134,5 +189,6 @@ async def on_lock_renew_failure_callback(renewable, error):
 
 
 asyncio.run(renew_lock_on_message_received_from_non_sessionful_entity())
+asyncio.run(renew_lock_via_receiver_auto_lock_renewer_keyword())
 asyncio.run(renew_lock_on_session_of_the_sessionful_entity())
 asyncio.run(renew_lock_with_lock_renewal_failure_callback())