Azure · pranz1996 · Jul 10, 2026 · Jul 10, 2026 · Jul 10, 2026 · Jul 10, 2026
@@ -12,6 +12,7 @@
 - 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))
+- Fixed a bug where closing a `PEEK_LOCK` receiver did not release messages that had been prefetched into the client buffer or were still in flight, so they remained locked at the broker until lock expiry — delaying their redelivery and inflating their delivery count. On close, a non-session `PEEK_LOCK` receiver now drains the link (stopping the broker and flushing in-flight transfers) and releases the buffered messages (`released` disposition), so the broker can redeliver them immediately without incrementing the delivery count. ([#42917](https://github.com/Azure/azure-sdk-for-python/issues/42917))
 
 ## 7.14.3 (2025-11-11)
 

@@ -538,6 +538,15 @@ def _renew_locks(self, *lock_tokens: str, **kwargs: Any) -> Any:
 
     def _close_handler(self):
         self._message_iter = None
+        if (
+            self._handler
+            and self._receive_mode == ServiceBusReceiveMode.PEEK_LOCK
+            and not self._session
+        ):
+            # Drain the link and release buffered/in-flight messages so they are not
+            # left locked at the broker until lock expiry (delaying redelivery,
+            # inflating delivery count).
+            self._amqp_transport.drain_receive_link_and_release_messages(self._handler)
         super(ServiceBusReceiver, self)._close_handler()
 
     @property

@@ -302,6 +302,14 @@ def reset_link_credit(handler, link_credit):
         :param int link_credit: The link credit.
         """
 
+    @staticmethod
+    @abstractmethod
+    def drain_receive_link_and_release_messages(handler):
+        """
+        Drain the receive link and release buffered/in-flight messages on close.
+        :param ~uamqp.ReceiveClient or ~pyamqp.ReceiveClient handler: The handler.
+        """
+
     @staticmethod
     @abstractmethod
     def settle_message_via_receiver_link(

@@ -157,6 +157,10 @@ def is_retryable(self, error):
     ERROR_CODE_TIMEOUT: OperationTimeoutError,
 }
 
+# Safety cap (seconds) for draining a receive link on close; the drain loop exits
+# early on quiescence, so a normal close returns well before this.
+RECEIVE_LINK_DRAIN_TIMEOUT = 5
+
 
 class PyamqpTransport(AmqpTransport):  # pylint: disable=too-many-public-methods
     """
@@ -799,6 +803,53 @@ def reset_link_credit(handler: "ReceiveClient", link_credit: int) -> None:
         """
         handler._link.flow(link_credit=link_credit)  # pylint: disable=protected-access
 
+    @staticmethod
+    def drain_receive_link_and_release_messages(handler: "ReceiveClient") -> None:
+        """
+        Drain the receive link and release buffered/in-flight messages on close, so
+        they are not left locked at the broker until lock expiry. Intended for
+        non-session PEEK_LOCK receivers only (gated by the caller).
+        :param ReceiveClient handler: Client whose receive link to drain.
+        :rtype: None
+        """
+        # pylint: disable=protected-access
+        link = handler._link
+        if link is None or link._is_closed:
+            return  # cannot drain or settle through a closed/absent link
+        try:
+            if link.current_link_credit > 0:
+                # Outstanding credit: stop the broker and pump in-flight transfers
+                # into the buffer before releasing them below.
+                outstanding = link.current_link_credit
+                link.flow(link_credit=0, drain=True)
+                deadline = time.time() + RECEIVE_LINK_DRAIN_TIMEOUT
+                idle_cycles = 0
+                while True:
+                    remaining = deadline - time.time()
+                    if remaining <= 0:
+                        break
+                    before = handler._received_messages.qsize()
+                    # listen() directly, not do_work() (which re-issues credit at 0 and
+                    # undoes the drain). batch=outstanding assembles multi-frame transfers
+                    # in one cycle; cap each read by the remaining budget so close stays
+                    # bounded. pyamqp drops the drain echo, so stop after 2 idle cycles.
+                    wait = remaining if handler._socket_timeout is None else min(handler._socket_timeout, remaining)
+                    handler._connection.listen(wait=wait, batch=outstanding)
+                    if handler._received_messages.qsize() == before:
+                        idle_cycles += 1
+                        if idle_cycles >= 2:
+                            break
+                    else:
+                        idle_cycles = 0
+            # Always release buffered deliveries (incl. the credit==0 prefetch case) so
+            # they are not left locked until lock expiry.
+            while not handler._received_messages.empty():
+                frame, _ = handler._received_messages.get_nowait()
+                handler.settle_messages(frame[1], frame[2], "released")
+                handler._received_messages.task_done()
+        except Exception:  # pylint: disable=broad-except
+            pass  # a faulted/detached link must not block close
+
     @staticmethod
     def settle_message_via_receiver_link(
         handler: ReceiveClient,

@@ -848,6 +848,15 @@ def reset_link_credit(handler: "ReceiveClient", link_credit: int) -> None:
             """
             handler.message_handler.reset_link_credit(link_credit)
 
+        @staticmethod
+        def drain_receive_link_and_release_messages(handler: "ReceiveClient") -> None:
+            """
+            No-op for uamqp: drain-on-close is only implemented for the pyamqp
+            transport (the default). uamqp is deprecated.
+            :param ~uamqp.ReceiveClient handler: The handler.
+            :rtype: None
+            """
+
         # Executes message settlement, implementation is in settle_message_via_receiver_link_impl
         # May be able to remove and just call methods in private method.
         @staticmethod

@@ -527,6 +527,15 @@ async def _renew_locks(self, *lock_tokens: str, timeout: Optional[float] = None)
 
     async def _close_handler(self):
         self._message_iter = None
+        if (
+            self._handler
+            and self._receive_mode == ServiceBusReceiveMode.PEEK_LOCK
+            and not self._session
+        ):
+            # Drain the link and release buffered/in-flight messages so they are not
+            # left locked at the broker until lock expiry (delaying redelivery,
+            # inflating delivery count).
+            await self._amqp_transport.drain_receive_link_and_release_messages_async(self._handler)
         await super(ServiceBusReceiver, self)._close_handler()
 
     @property

@@ -244,6 +244,16 @@ async def reset_link_credit_async(handler, link_credit):
         :rtype: None
         """
 
+    @staticmethod
+    @abstractmethod
+    async def drain_receive_link_and_release_messages_async(handler):
+        """
+        Drain the receive link and release buffered/in-flight messages on close.
+        :param ~uamqp.ReceiveClientAsync
+         or ~pyamqp.aio.ReceiveClientAsync handler: Client whose receive link to drain.
+        :rtype: None
+        """
+
     @staticmethod
     @abstractmethod
     async def settle_message_via_receiver_link_async(

@@ -42,7 +42,7 @@
     OPERATION_TIMEOUT,
     NEXT_AVAILABLE_SESSION,
 )
-from ..._transport._pyamqp_transport import PyamqpTransport
+from ..._transport._pyamqp_transport import PyamqpTransport, RECEIVE_LINK_DRAIN_TIMEOUT
 from ...exceptions import (
     OperationTimeoutError,
     ServiceBusConnectionError,
@@ -301,6 +301,54 @@ async def reset_link_credit_async(handler: "ReceiveClientAsync", link_credit: in
         """
         await handler._link.flow(link_credit=link_credit)  # pylint: disable=protected-access
 
+    @staticmethod
+    async def drain_receive_link_and_release_messages_async(handler: "ReceiveClientAsync") -> None:
+        """
+        Drain the receive link and release buffered/in-flight messages on close, so
+        they are not left locked at the broker until lock expiry. Intended for
+        non-session PEEK_LOCK receivers only (gated by the caller).
+        :param ReceiveClientAsync handler: Client whose receive link to drain.
+        :rtype: None
+        """
+        # pylint: disable=protected-access
+        link = handler._link
+        if link is None or link._is_closed:
+            return  # cannot drain or settle through a closed/absent link
+        try:
+            if link.current_link_credit > 0:
+                # Outstanding credit: stop the broker and pump in-flight transfers
+                # into the buffer before releasing them below.
+                outstanding = link.current_link_credit
+                await link.flow(link_credit=0, drain=True)
+                deadline = time.time() + RECEIVE_LINK_DRAIN_TIMEOUT
+                idle_cycles = 0
+                while True:
+                    remaining = deadline - time.time()
+                    if remaining <= 0:
+                        break
+                    before = handler._received_messages.qsize()
+                    # listen() directly, not do_work_async() (which re-issues credit at
+                    # 0 and undoes the drain). batch=outstanding assembles multi-frame
+                    # transfers in one cycle; cap each read by the remaining budget so
+                    # close stays bounded. pyamqp drops the drain echo, so stop after 2
+                    # idle cycles.
+                    wait = remaining if handler._socket_timeout is None else min(handler._socket_timeout, remaining)
+                    await handler._connection.listen(wait=wait, batch=outstanding)
+                    if handler._received_messages.qsize() == before:
+                        idle_cycles += 1
+                        if idle_cycles >= 2:
+                            break
+                    else:
+                        idle_cycles = 0
+            # Always release buffered deliveries (incl. the credit==0 prefetch case) so
+            # they are not left locked until lock expiry.
+            while not handler._received_messages.empty():
+                frame, _ = handler._received_messages.get_nowait()
+                await handler.settle_messages_async(frame[1], frame[2], "released")
+                handler._received_messages.task_done()
+        except Exception:  # pylint: disable=broad-except
+            pass  # a faulted/detached link must not block close
+
     @staticmethod
     async def settle_message_via_receiver_link_async(
         handler: "ReceiveClientAsync",

@@ -248,6 +248,15 @@ async def reset_link_credit_async(handler: "ReceiveClientAsync", link_credit: in
             """
             await handler.message_handler.reset_link_credit_async(link_credit)
 
+        @staticmethod
+        async def drain_receive_link_and_release_messages_async(handler: "ReceiveClientAsync") -> None:
+            """
+            No-op for uamqp: drain-on-close is only implemented for the pyamqp
+            transport (the default). uamqp is deprecated.
+            :param ReceiveClientAsync handler: The handler.
+            :rtype: None
+            """
+
         @staticmethod
         async def settle_message_via_receiver_link_async(
             handler: "ReceiveClientAsync",