diff --git a/sdk/core/azure-core-amqp/CHANGELOG.md b/sdk/core/azure-core-amqp/CHANGELOG.md
index 16652ac226..259e3233a1 100644
--- a/sdk/core/azure-core-amqp/CHANGELOG.md
+++ b/sdk/core/azure-core-amqp/CHANGELOG.md
@@ -8,6 +8,8 @@
 
 ### Bugs Fixed
 
+- Fixed uAMQP session destruction after a failed `End`. Native resources and the connection polling reference are released when the last session owner is destroyed; callers must still end sessions before destruction.
+
 - Fixed uAMQP message annotation encoding so messages with annotations, including Event Hubs partition keys, include the required AMQP section descriptor. [[#7426]](https://github.com/Azure/azure-sdk-for-cpp/issues/7426)
 
 ### Other Changes
diff --git a/sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/session.hpp b/sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/session.hpp
index 8b0ccd36f9..0d869fc649 100644
--- a/sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/session.hpp
+++ b/sdk/core/azure-core-amqp/inc/azure/core/amqp/internal/session.hpp
@@ -233,11 +233,17 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
 
     /** @brief Ends operations on the session.
      *
+     * @note With uAMQP, a failed protocol shutdown throws. The native session and its polling
+     * reference are then released when the last session owner is destroyed. Calling End is
+     * still required before destroying a begun session.
      */
     void End(Azure::Core::Context const& context);
 
     /** @brief Ends operations on the session.
      *
+     * @note With uAMQP, a failed protocol shutdown throws. The native session and its polling
+     * reference are then released when the last session owner is destroyed. Calling End is
+     * still required before destroying a begun session.
      */
     void End(
         std::string const& condition_value,
diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/session_impl.hpp b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/session_impl.hpp
index 75a1d64d2f..0c2a446258 100644
--- a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/session_impl.hpp
+++ b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/private/session_impl.hpp
@@ -42,6 +42,9 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
 
   class SessionImpl final : public std::enable_shared_from_this<SessionImpl> {
   public:
+    // Internal seam for exercising native shutdown failures without a live broker.
+    using EndSession = int (*)(SESSION_HANDLE, char const*, char const*);
+
     SessionImpl(
         std::shared_ptr<_detail::ConnectionImpl> parentConnection,
         _internal::Endpoint& newEndpoint,
@@ -50,7 +53,8 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     SessionImpl(
         std::shared_ptr<_detail::ConnectionImpl> parentConnection,
         _internal::SessionOptions const& options,
-        _internal::SessionEvents* eventHandler = nullptr);
+        _internal::SessionEvents* eventHandler = nullptr,
+        EndSession endSession = session_end);
     ~SessionImpl() noexcept;
 
     SessionImpl(SessionImpl const&) = delete;
@@ -79,8 +83,11 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
 
   private:
     SessionImpl();
+    EndSession m_endSession{session_end};
     bool m_connectionAsyncStarted{false};
     bool m_isBegun{false};
+    // A failed protocol shutdown still requires native destruction and a polling release.
+    bool m_endFailed{false};
     std::shared_ptr<_detail::ConnectionImpl> m_connectionToPoll;
     UniqueAmqpSession m_session;
     _internal::SessionOptions m_options;
diff --git a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/session.cpp b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/session.cpp
index 25c994cfe9..1b05a6b99b 100644
--- a/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/session.cpp
+++ b/sdk/core/azure-core-amqp/src/impl/uamqp/amqp/session.cpp
@@ -34,19 +34,23 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
 
   SessionImpl::~SessionImpl() noexcept
   {
-    if (m_isBegun)
+    if (m_isBegun && !m_endFailed)
     {
       AZURE_ASSERT_MSG(false, "Session was not ended before destruction.");
     }
 
-    // If we have a mismatched begin/end pair, we need to stop polling on the connection so it
-    // gets cleaned up properly.
+    // A failed End leaves the native session alive while management clients and links own it.
+    // The last owner reaches here after those objects have gone. session_destroy frees the
+    // endpoint and native resources even if the protocol end frame cannot be sent.
+    {
+      auto lock{m_connectionToPoll->Lock()};
+      m_session.reset();
+    }
+    // Release our polling reference exactly once, outside the connection lock.
     if (m_connectionAsyncStarted)
     {
       m_connectionToPoll->EnableAsyncOperation(false);
     }
-    auto lock{m_connectionToPoll->Lock()};
-    m_session.reset();
   }
 
   SessionImpl::SessionImpl(
@@ -87,8 +91,9 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
   SessionImpl::SessionImpl(
       std::shared_ptr<_detail::ConnectionImpl> connection,
       _internal::SessionOptions const& options,
-      _internal::SessionEvents* eventHandler)
-      : m_connectionToPoll(connection),
+      _internal::SessionEvents* eventHandler,
+      EndSession endSession)
+      : m_endSession{endSession}, m_connectionToPoll(connection),
         m_session{session_create(*connection, SessionImpl::OnLinkAttachedFn, this)},
         m_options{options}, m_eventHandler{eventHandler}
 
@@ -159,28 +164,13 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     }
 
     m_isBegun = true;
+    m_endFailed = false;
 
     // Mark the connection as async so that we can use the async APIs.
     GetConnection()->EnableAsyncOperation(true);
     m_connectionAsyncStarted = true;
   }
-  void SessionImpl::End(Azure::Core::Context const&)
-  {
-    if (!m_isBegun)
-    {
-      throw std::runtime_error("Session End without corresponding Begin.");
-    }
-    // When we end the session, it clears all the links, so we need to ensure that the
-    //    m_newLinkAttachedQueue.Clear();
-    if (session_end(m_session.get(), nullptr, nullptr))
-    {
-      throw std::runtime_error("Could not begin session");
-    }
-    // Mark the connection as async so that we can use the async APIs.
-    GetConnection()->EnableAsyncOperation(false);
-    m_connectionAsyncStarted = false;
-    m_isBegun = false;
-  }
+  void SessionImpl::End(Azure::Core::Context const& context) { End({}, {}, context); }
 
   void SessionImpl::End(
       const std::string& condition,
@@ -191,19 +181,20 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
     {
       throw std::runtime_error("Session End without corresponding Begin.");
     }
-    // When we end the session, it clears all the links, so we need to ensure that the
-    //    m_newLinkAttachedQueue.Clear();
-    if (session_end(
+    // The native end releases the link endpoints, even if sending the end frame fails.
+    if (m_endSession(
             m_session.get(),
             condition.empty() ? nullptr : condition.c_str(),
             description.empty() ? nullptr : description.c_str()))
     {
-      throw std::runtime_error("Could not begin session");
+      m_endFailed = true;
+      throw std::runtime_error("Could not end session");
     }
-    // Mark the connection as async so that we can use the async APIs.
+    // Release the polling reference acquired by Begin.
     GetConnection()->EnableAsyncOperation(false);
     m_connectionAsyncStarted = false;
     m_isBegun = false;
+    m_endFailed = false;
   }
 
   void SessionImpl::SendDetach(
diff --git a/sdk/core/azure-core-amqp/test/ut/CMakeLists.txt b/sdk/core/azure-core-amqp/test/ut/CMakeLists.txt
index 5984ff201d..acab1a49ca 100644
--- a/sdk/core/azure-core-amqp/test/ut/CMakeLists.txt
+++ b/sdk/core/azure-core-amqp/test/ut/CMakeLists.txt
@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+# cspell: words aziotsharedutil
+
 cmake_minimum_required (VERSION 3.13)
 
 set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
@@ -29,6 +31,7 @@ add_executable(azure-core-amqp-tests
   async_operation_queue_tests.cpp
   azure_core_amqp_tests.cpp
   claim_based_security_tests.cpp
+  cleanup_test_listener.hpp
   connection_string_tests.cpp
   connection_tests.cpp
   global_state_tests.cpp
@@ -37,6 +40,7 @@ add_executable(azure-core-amqp-tests
   message_source_target.cpp
   operation_timeout_tests.cpp
   mock_amqp_server.hpp
+  session_cleanup_tests.cpp
   session_tests.cpp
   amqp_performative_tests.cpp
   ${UAMQP_ONLY_TESTS}
@@ -44,7 +48,8 @@ add_executable(azure-core-amqp-tests
 
 
 if (USE_UAMQP)
-target_include_directories(azure-core-amqp-tests PRIVATE ../../src/amqp/impl/uamqp/amqp/private)
+target_include_directories(azure-core-amqp-tests SYSTEM PRIVATE ${UAMQP_INC_FOLDER})
+target_link_libraries(azure-core-amqp-tests PRIVATE umock_c azure_macro_utils_c aziotsharedutil)
 endif()
 if (USE_RUST_AMQP)
 target_include_directories(azure-core-amqp-tests PRIVATE ../../src/impl/rust_amqp/rust_amqp/rust_wrapper)
diff --git a/sdk/core/azure-core-amqp/test/ut/cleanup_test_listener.hpp b/sdk/core/azure-core-amqp/test/ut/cleanup_test_listener.hpp
new file mode 100644
index 0000000000..7b6ed9110d
--- /dev/null
+++ b/sdk/core/azure-core-amqp/test/ut/cleanup_test_listener.hpp
@@ -0,0 +1,76 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+#pragma once
+
+#include <azure/core/amqp/internal/common/global_state.hpp>
+#include <azure/core/platform.hpp>
+
+#if defined(AZ_PLATFORM_WINDOWS)
+#include <winsock2.h>
+#else
+#include <unistd.h>
+
+#include <netinet/in.h>
+#include <sys/socket.h>
+#endif
+
+#include <cstdint>
+#include <stdexcept>
+
+namespace Azure { namespace Core { namespace Amqp { namespace Tests {
+  // Allows Begin to connect over TCP without starting a peer AMQP session. Cleanup
+  // fault tests do not need peer callbacks, and must not race a mock server's session teardown.
+  class CleanupTestListener final {
+  public:
+    CleanupTestListener()
+    {
+      Common::_detail::GlobalStateHolder::GlobalStateInstance();
+      m_socket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+      if (m_socket == InvalidSocket)
+      {
+        throw std::runtime_error("Could not create cleanup test listener");
+      }
+      sockaddr_in address{};
+      address.sin_family = AF_INET;
+      address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+#if defined(AZ_PLATFORM_WINDOWS)
+      int size = sizeof(address);
+#else
+      socklen_t size = sizeof(address);
+#endif
+      if (bind(m_socket, reinterpret_cast<sockaddr*>(&address), sizeof(address)) != 0
+          || listen(m_socket, 1) != 0
+          || getsockname(m_socket, reinterpret_cast<sockaddr*>(&address), &size) != 0)
+      {
+        Close();
+        throw std::runtime_error("Could not listen on cleanup test port");
+      }
+      m_port = ntohs(address.sin_port);
+    }
+
+    ~CleanupTestListener() { Close(); }
+    CleanupTestListener(CleanupTestListener const&) = delete;
+    CleanupTestListener& operator=(CleanupTestListener const&) = delete;
+    std::uint16_t GetPort() const { return m_port; }
+
+  private:
+    using Socket = decltype(socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
+    static constexpr Socket InvalidSocket = static_cast<Socket>(-1);
+    Socket m_socket{InvalidSocket};
+    std::uint16_t m_port{};
+
+    void Close()
+    {
+      if (m_socket != InvalidSocket)
+      {
+#if defined(AZ_PLATFORM_WINDOWS)
+        closesocket(m_socket);
+#else
+        close(m_socket);
+#endif
+        m_socket = InvalidSocket;
+      }
+    }
+  };
+}}}} // namespace Azure::Core::Amqp::Tests
diff --git a/sdk/core/azure-core-amqp/test/ut/session_cleanup_tests.cpp b/sdk/core/azure-core-amqp/test/ut/session_cleanup_tests.cpp
new file mode 100644
index 0000000000..f103766e9b
--- /dev/null
+++ b/sdk/core/azure-core-amqp/test/ut/session_cleanup_tests.cpp
@@ -0,0 +1,142 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+#if ENABLE_UAMQP
+#include "../../src/impl/uamqp/amqp/private/session_impl.hpp"
+#include "cleanup_test_listener.hpp"
+
+#include <azure/core/amqp/internal/management.hpp>
+
+#include <cstdlib>
+
+#include <gtest/gtest.h>
+
+namespace Azure { namespace Core { namespace Amqp { namespace Tests {
+  namespace {
+    int FailEnd(SESSION_HANDLE, char const*, char const*) { return 1; }
+
+    void FailedEndWithSharedOwner(bool withError)
+    {
+      std::weak_ptr<_detail::SessionImpl> sessionLifetime;
+      std::weak_ptr<_detail::ConnectionImpl> connectionLifetime;
+      {
+        CleanupTestListener server;
+        _internal::ConnectionOptions options;
+        options.Port = server.GetPort();
+        _internal::Connection connection("localhost", nullptr, options);
+        connectionLifetime = _detail::ConnectionFactory::GetImpl(connection);
+        std::unique_ptr<_internal::ManagementClient> management;
+        {
+          auto impl = std::make_shared<_detail::SessionImpl>(
+              _detail::ConnectionFactory::GetImpl(connection),
+              _internal::SessionOptions{},
+              nullptr,
+              FailEnd);
+          sessionLifetime = impl;
+          auto session = _detail::SessionFactory::CreateFromInternal(impl);
+          session.Begin({});
+          management = std::make_unique<_internal::ManagementClient>(
+              session.CreateManagementClient("hub", {}));
+          if (withError)
+          {
+            EXPECT_THROW(session.End("amqp:internal-error", "shutdown", {}), std::runtime_error);
+          }
+          else
+          {
+            EXPECT_THROW(session.End({}), std::runtime_error);
+          }
+        }
+        // A management client still owns the session. Native destruction must wait for it.
+        EXPECT_FALSE(sessionLifetime.expired());
+        management.reset();
+        EXPECT_TRUE(sessionLifetime.expired());
+      }
+      Common::_detail::GlobalStateHolder::GlobalStateInstance()->AssertIdle();
+      EXPECT_TRUE(connectionLifetime.expired());
+    }
+
+    int FailOnceEnd(SESSION_HANDLE session, char const* condition, char const* description)
+    {
+      static bool failed = false;
+      if (!failed)
+      {
+        failed = true;
+        return 1;
+      }
+      return session_end(session, condition, description);
+    }
+  } // namespace
+
+  class SessionCleanupDeathTest : public ::testing::Test {
+  protected:
+    void SetUp() override
+    {
+      // Re-exec the child instead of inheriting AMQP polling-thread mutexes across fork.
+      ::testing::FLAGS_gtest_death_test_style = "threadsafe";
+    }
+  };
+
+  TEST_F(SessionCleanupDeathTest, FailedEndReleasesLastOwner)
+  {
+    EXPECT_EXIT(
+        {
+          FailedEndWithSharedOwner(false);
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+
+  TEST_F(SessionCleanupDeathTest, FailedEndWithErrorReleasesLastOwner)
+  {
+    EXPECT_EXIT(
+        {
+          FailedEndWithSharedOwner(true);
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+
+  TEST_F(SessionCleanupDeathTest, FailedEndCanBeRetried)
+  {
+    EXPECT_EXIT(
+        {
+          {
+            CleanupTestListener server;
+            _internal::ConnectionOptions options;
+            options.Port = server.GetPort();
+            _internal::Connection connection("localhost", nullptr, options);
+            auto session = _detail::SessionFactory::CreateFromInternal(
+                std::make_shared<_detail::SessionImpl>(
+                    _detail::ConnectionFactory::GetImpl(connection),
+                    _internal::SessionOptions{},
+                    nullptr,
+                    FailOnceEnd));
+            session.Begin({});
+            EXPECT_THROW(session.End({}), std::runtime_error);
+            EXPECT_NO_THROW(session.End("amqp:internal-error", "retry", {}));
+            EXPECT_THROW(session.End({}), std::runtime_error);
+          }
+          Common::_detail::GlobalStateHolder::GlobalStateInstance()->AssertIdle();
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+
+  TEST_F(SessionCleanupDeathTest, MissingEndStillAsserts)
+  {
+    EXPECT_DEATH(
+        {
+          CleanupTestListener server;
+          _internal::ConnectionOptions options;
+          options.Port = server.GetPort();
+          _internal::Connection connection("localhost", nullptr, options);
+          auto session = connection.CreateSession();
+          session.Begin({});
+        },
+        "");
+  }
+}}}} // namespace Azure::Core::Amqp::Tests
+#endif
diff --git a/sdk/eventhubs/assets.json b/sdk/eventhubs/assets.json
index 521a91878e..1b0d927a32 100644
--- a/sdk/eventhubs/assets.json
+++ b/sdk/eventhubs/assets.json
@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "cpp",
   "TagPrefix": "cpp/eventhubs",
-  "Tag": "cpp/eventhubs_54bfb9d8e7"
+  "Tag": "cpp/eventhubs_4c104aa882"
 }
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md b/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md
index 577f4b3819..46aa127097 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md
+++ b/sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md
@@ -8,6 +8,11 @@
 
 ### Bugs Fixed
 
+- Fixed uAMQP properties queries ignoring cancellation and deadlines while waiting for another properties query to finish.
+
+- Fixed producer cleanup after partition discovery fails during management-client initialization or session shutdown. Cleanup now ends an acquired session independently of management-client creation and supports repeated close calls.
+- Synchronized uAMQP properties queries with cleanup during producer recovery. Cleanup cancels an active query before releasing its management client and session, without cancelling the caller's context. Rust cancellation and concurrent cleanup are deferred to [azure-sdk-for-rust#5527](https://github.com/Azure/azure-sdk-for-rust/issues/5527).
+
 ### Other Changes
 
 ## 1.0.0-beta.15 (2026-09-02)
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_utilities.hpp b/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_utilities.hpp
index 78fd0d68ac..3a7e8ac95a 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_utilities.hpp
+++ b/sdk/eventhubs/azure-messaging-eventhubs/src/private/eventhubs_utilities.hpp
@@ -21,6 +21,7 @@
 #include <chrono>
 #include <cstdint>
 #include <memory>
+#include <mutex>
 #include <string>
 #include <utility>
 
@@ -144,21 +145,41 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail
 
   class EventHubsPropertiesClient {
   public:
+    using ManagementClientFactory = Azure::Core::Amqp::_internal::ManagementClient (*)(
+        Azure::Core::Amqp::_internal::Session const&,
+        std::string const&,
+        Azure::Core::Amqp::_internal::ManagementClientOptions const&);
+
+    static Azure::Core::Amqp::_internal::ManagementClient CreateManagementClient(
+        Azure::Core::Amqp::_internal::Session const& session,
+        std::string const& eventHub,
+        Azure::Core::Amqp::_internal::ManagementClientOptions const& options)
+    {
+      return session.CreateManagementClient(eventHub, options);
+    }
+
+    // Internal injection point for failures between beginning the session and assigning the client.
+    EventHubsPropertiesClient(
+        Azure::Core::Amqp::_internal::Session session,
+        std::string eventHubName,
+        ManagementClientFactory createManagementClient = CreateManagementClient)
+        : m_session{std::move(session)}, m_eventHub{std::move(eventHubName)},
+          m_createManagementClient{createManagementClient}
+    {
+    }
+
     EventHubsPropertiesClient(
         const Azure::Core::Amqp::_internal::Connection& connection,
         std::string eventHubName)
-        : m_session{connection.CreateSession()}, m_eventHub{std::move(eventHubName)} {};
+        : EventHubsPropertiesClient(connection.CreateSession(), std::move(eventHubName)) {};
 
     ~EventHubsPropertiesClient()
     {
-      // A destructor is noexcept, so an exception here terminates the process. An idle
-      // link detach after 30 minutes makes a throwing close the ordinary case here.
+      // Finish both management and session cleanup, including partially initialized clients.
+      // Explicit Close and destruction use the same ownership checks.
       try
       {
-        if (m_managementClientIsOpen)
-        {
-          m_managementClient->Close();
-        }
+        Close({});
       }
       catch (std::exception const& ex)
       {
@@ -174,7 +195,13 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail
         Core::Context const& context)
     {
 
-      EnsureManagementClient(context);
+#if defined(ENABLE_UAMQP)
+      auto lock = AcquireManagementLock(context);
+      auto operationContext = CreateOperationContext(context);
+#else
+      auto const* operationContext = &context;
+#endif
+      EnsureManagementClient(*operationContext);
 
       // Send a message to the management endpoint to retrieve the properties of the eventhub.
       Azure::Core::Amqp::Models::AmqpMessage message;
@@ -186,7 +213,7 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail
           "com.microsoft:eventhub" /* type of operation */,
           "" /* locales */,
           message,
-          context);
+          *operationContext);
 
       Models::EventHubProperties properties;
       if (result.Status != Azure::Core::Amqp::_internal::ManagementOperationStatus::Ok)
@@ -212,10 +239,11 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail
         }
         auto bodyMap = body.AsMap();
         properties.Name = static_cast<std::string>(bodyMap["name"]);
-        properties.CreatedOn = Azure::DateTime(std::chrono::system_clock::from_time_t(
-            std::chrono::duration_cast<std::chrono::seconds>(
-                static_cast<std::chrono::milliseconds>(bodyMap["created_at"].AsTimestamp()))
-                .count()));
+        properties.CreatedOn = Azure::DateTime(
+            std::chrono::system_clock::from_time_t(
+                std::chrono::duration_cast<std::chrono::seconds>(
+                    static_cast<std::chrono::milliseconds>(bodyMap["created_at"].AsTimestamp()))
+                    .count()));
         auto partitions = bodyMap["partition_ids"].AsArray();
         for (const auto& partition : partitions)
         {
@@ -230,7 +258,13 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail
         std::string const& partitionId,
         Core::Context const& context)
     {
-      EnsureManagementClient(context);
+#if defined(ENABLE_UAMQP)
+      auto lock = AcquireManagementLock(context);
+      auto operationContext = CreateOperationContext(context);
+#else
+      auto const* operationContext = &context;
+#endif
+      EnsureManagementClient(*operationContext);
 
       // Send a message to the management endpoint to retrieve the properties of the eventhub.
       Azure::Core::Amqp::Models::AmqpMessage message;
@@ -244,7 +278,7 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail
           "com.microsoft:partition" /* type of operation */,
           "" /* locales */,
           message,
-          context);
+          *operationContext);
 
       Models::EventHubPartitionProperties properties;
       if (result.Status != Azure::Core::Amqp::_internal::ManagementOperationStatus::Ok)
@@ -272,11 +306,12 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail
         // For <reasons> the last enqueued offset is returned as a string. Convert to an int64.
         properties.LastEnqueuedOffset = static_cast<std::string>(bodyMap["last_enqueued_offset"]);
 
-        properties.LastEnqueuedTimeUtc = Azure::DateTime(std::chrono::system_clock::from_time_t(
-            std::chrono::duration_cast<std::chrono::seconds>(
-                static_cast<std::chrono::milliseconds>(
-                    bodyMap["last_enqueued_time_utc"].AsTimestamp()))
-                .count()));
+        properties.LastEnqueuedTimeUtc = Azure::DateTime(
+            std::chrono::system_clock::from_time_t(
+                std::chrono::duration_cast<std::chrono::seconds>(
+                    static_cast<std::chrono::milliseconds>(
+                        bodyMap["last_enqueued_time_utc"].AsTimestamp()))
+                    .count()));
         properties.IsEmpty = bodyMap["is_partition_empty"];
       }
       return properties;
@@ -284,28 +319,43 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail
 
     void Close(Azure::Core::Context const& context)
     {
-      std::unique_lock<std::mutex> lock(m_managementClientMutex);
-      if (m_managementClient)
+#if defined(ENABLE_UAMQP)
       {
-        if (m_managementClientIsOpen)
+        // Cancel before waiting for the operation lock: a query may have no caller deadline.
+        // Reject queries on stale references once producer/consumer invalidation starts.
+        std::lock_guard<std::mutex> cancellationLock(m_operationContextMutex);
+        m_isClosed = true;
+        if (auto operationContext = m_operationContext.lock())
         {
-          // Drop the flag first, so a throwing close cannot make the destructor close again.
-          m_managementClientIsOpen = false;
-          try
-          {
-            m_managementClient->Close(context);
-          }
-          catch (std::exception const& ex)
-          {
-            Azure::Core::Diagnostics::_internal::Log::Stream(
-                Azure::Core::Diagnostics::Logger::Level::Warning)
-                << "Exception while closing the management client: " << ex.what();
-          }
+          operationContext->Cancel();
         }
-        // Its own try block, so a failed management close does not strand the session.
+      }
+#endif
+      std::unique_lock<std::timed_mutex> lock(m_managementClientMutex);
+      if (m_managementClientIsOpen)
+      {
+        // ManagementClient::Close releases its links even when it reports a close failure.
+        m_managementClientIsOpen = false;
+        try
+        {
+          m_managementClient->Close(context);
+        }
+        catch (std::exception const& ex)
+        {
+          Azure::Core::Diagnostics::_internal::Log::Stream(
+              Azure::Core::Diagnostics::Logger::Level::Warning)
+              << "Exception while closing the management client: " << ex.what();
+        }
+      }
+      // Drop link ownership before ending the session. Its native end frees link endpoints.
+      m_managementClient.reset();
+      // Beginning the session precedes management creation, which can throw before assignment.
+      if (m_sessionIsBegun)
+      {
         try
         {
           m_session.End(context);
+          m_sessionIsBegun = false;
         }
         catch (std::exception const& ex)
         {
@@ -317,13 +367,51 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail
     }
 
   private:
-    void EnsureManagementClient(Azure::Core::Context const& context)
+#if defined(ENABLE_UAMQP)
+    std::unique_lock<std::timed_mutex> AcquireManagementLock(Azure::Core::Context const& context)
     {
+      context.ThrowIfCancelled();
+      std::unique_lock<std::timed_mutex> lock(m_managementClientMutex, std::defer_lock);
+      // A query waiting behind an unanswered operation must still observe its own cancellation.
+      while (!lock.try_lock_for(std::chrono::milliseconds(10)))
+      {
+        context.ThrowIfCancelled();
+      }
+      context.ThrowIfCancelled();
+      return lock;
+    }
+
+    std::shared_ptr<Azure::Core::Context> CreateOperationContext(
+        Azure::Core::Context const& context)
+    {
+      std::lock_guard<std::mutex> lock(m_operationContextMutex);
+      if (m_isClosed)
+      {
+        throw Azure::Core::OperationCancelledException("Properties client is closed.");
+      }
+      // A child inherits caller cancellation but can be cancelled without changing the caller.
+      auto operationContext
+          = std::make_shared<Azure::Core::Context>(context.WithDeadline(context.GetDeadline()));
+      m_operationContext = operationContext;
+      return operationContext;
+    }
+#endif
 
-      std::unique_lock<std::mutex> lock(m_managementClientMutex);
+    void EnsureManagementClient(Azure::Core::Context const& context)
+    {
+#if !defined(ENABLE_UAMQP)
+      // Rust management calls do not yet observe cancellation of the child context.
+      // Keep the lock limited to initialization until that support is available.
+      std::unique_lock<std::timed_mutex> lock(m_managementClientMutex);
+#endif
+      // With uAMQP, the getter holds this lock through operation completion.
       if (!m_managementClient)
       {
-        m_session.Begin(context);
+        if (!m_sessionIsBegun)
+        {
+          m_session.Begin(context);
+          m_sessionIsBegun = true;
+        }
 
         // Create a management client off the session.
         // Eventhubs management APIs return a status code in the "status-code" application
@@ -332,7 +420,7 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail
         managementClientOptions.EnableTrace = _detail::EnableAmqpTrace;
         managementClientOptions.ExpectedStatusCodeKeyName = "status-code";
         m_managementClient = std::make_unique<Azure::Core::Amqp::_internal::ManagementClient>(
-            m_session.CreateManagementClient(m_eventHub, managementClientOptions));
+            m_createManagementClient(m_session, m_eventHub, managementClientOptions));
 
         auto openResult{m_managementClient->Open(context)};
         if (openResult != Azure::Core::Amqp::_internal::ManagementOpenStatus::Ok)
@@ -343,17 +431,18 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace _detail
       }
     }
 
-    std::unique_ptr<Azure::Core::Amqp::_internal::ManagementClient> const& GetManagementClient()
-    {
-      std::unique_lock<std::mutex> lock(m_managementClientMutex);
-      return m_managementClient;
-    }
-
-    std::mutex m_managementClientMutex;
+    std::timed_mutex m_managementClientMutex;
+#if defined(ENABLE_UAMQP)
+    std::mutex m_operationContextMutex;
+    std::weak_ptr<Azure::Core::Context> m_operationContext;
+    bool m_isClosed{false};
+#endif
     Azure::Core::Amqp::_internal::Session m_session;
     std::unique_ptr<Azure::Core::Amqp::_internal::ManagementClient> m_managementClient;
     bool m_managementClientIsOpen{false};
+    bool m_sessionIsBegun{false};
     std::string m_eventHub;
+    ManagementClientFactory m_createManagementClient;
   };
 
   class EventHubsUtilities {
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/CMakeLists.txt b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/CMakeLists.txt
index b21bcfe95b..759dbc7315 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/CMakeLists.txt
+++ b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/CMakeLists.txt
@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+# cspell: words umock aziotsharedutil
+
 cmake_minimum_required (VERSION 3.13)
 
 project (azure-messaging-eventhubs-test LANGUAGES CXX)
@@ -29,6 +31,7 @@ add_executable (
     processor_load_balancer_test.cpp
     processor_test.cpp
     producer_client_test.cpp
+    properties_cleanup_test.cpp
     reattach_connection_string_test.cpp
     reattach_policy_test.cpp
     retry_operation_test.cpp
@@ -49,7 +52,13 @@ endif()
 if(USE_RUST_AMQP)
   target_compile_definitions(azure-messaging-eventhubs-test PRIVATE ENABLE_RUST_AMQP)
 else()
-  target_compile_definitions(azure-messaging-eventhubs-test PRIVATE ENABLE_UAMQP)
+  find_package(umock_c CONFIG REQUIRED)
+  find_package(azure_macro_utils_c CONFIG REQUIRED)
+  find_package(azure_c_shared_utility CONFIG REQUIRED)
+  target_compile_definitions(azure-messaging-eventhubs-test PRIVATE ENABLE_UAMQP _azure_TESTING_BUILD)
+  target_include_directories(azure-messaging-eventhubs-test SYSTEM PRIVATE
+    ${CMAKE_CURRENT_SOURCE_DIR}/../../../../core/azure-core-amqp/src/impl/uamqp/vendor/azure-uamqp-c/inc)
+  target_link_libraries(azure-messaging-eventhubs-test PRIVATE umock_c azure_macro_utils_c aziotsharedutil)
 endif()
 
 target_link_libraries(
diff --git a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/properties_cleanup_test.cpp b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/properties_cleanup_test.cpp
new file mode 100644
index 0000000000..ac245b8888
--- /dev/null
+++ b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/properties_cleanup_test.cpp
@@ -0,0 +1,603 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+#if ENABLE_UAMQP
+#include "../../../../core/azure-core-amqp/src/impl/uamqp/amqp/private/session_impl.hpp"
+#include "../../../../core/azure-core-amqp/test/ut/cleanup_test_listener.hpp"
+#include "../../../../core/azure-core-amqp/test/ut/mock_amqp_server.hpp"
+#include "private/eventhubs_utilities.hpp"
+
+#include <azure/core/diagnostics/logger.hpp>
+#include <azure/core/platform.hpp>
+#include <azure/messaging/eventhubs/producer_client.hpp>
+
+#include <atomic>
+#include <cstdlib>
+#include <future>
+
+#include <gtest/gtest.h>
+
+namespace {
+std::uint16_t AvailablePort()
+{
+  Azure::Core::Amqp::Tests::CleanupTestListener listener;
+  return listener.GetPort();
+}
+} // namespace
+
+namespace Azure { namespace Messaging { namespace EventHubs { namespace Test {
+  namespace {
+    namespace Amqp = Azure::Core::Amqp;
+    using _detail::EventHubsPropertiesClient;
+
+    Amqp::_internal::ManagementClient FailCreation(
+        Amqp::_internal::Session const&,
+        std::string const&,
+        Amqp::_internal::ManagementClientOptions const&)
+    {
+      throw std::runtime_error("injected management creation failure");
+    }
+
+    int FailEnd(SESSION_HANDLE, char const*, char const*) { return 1; }
+
+    class PropertiesEndpoint final : public Amqp::Tests::MessageTests::MockServiceEndpoint {
+    public:
+      PropertiesEndpoint(bool failDiscovery = false, std::promise<void>* requestReceived = nullptr)
+          : MockServiceEndpoint("$management", {}), m_failDiscovery{failDiscovery},
+            m_requestReceived{requestReceived}
+      {
+      }
+
+      int RequestCount() const { return m_requestCount.load(); }
+
+    private:
+      std::atomic<int> m_requestCount{0};
+      bool m_failDiscovery;
+      std::promise<void>* m_requestReceived;
+      void MessageReceived(
+          std::string const&,
+          std::shared_ptr<Amqp::Models::AmqpMessage> const& request) override
+      {
+        auto const requestCount = ++m_requestCount;
+        if (m_requestReceived)
+        {
+          // Acknowledge the first request without replying to any pending query.
+          if (requestCount == 1)
+          {
+            m_requestReceived->set_value();
+          }
+          return;
+        }
+        Amqp::Models::AmqpMessage response;
+        response.ApplicationProperties["status-code"] = m_failDiscovery ? 500 : 200;
+        response.Properties.CorrelationId = request->Properties.MessageId;
+        Amqp::Models::AmqpMap body;
+        body["name"] = "hub";
+        body["created_at"]
+            = Amqp::Models::AmqpTimestamp(std::chrono::milliseconds(0)).AsAmqpValue();
+        body["partition_ids"] = Amqp::Models::AmqpArray{"0", "1"}.AsAmqpValue();
+        body["partition"] = "0";
+        body["begin_sequence_number"] = std::int64_t{0};
+        body["last_enqueued_sequence_number"] = std::int64_t{1};
+        body["last_enqueued_offset"] = "1";
+        body["last_enqueued_time_utc"]
+            = Amqp::Models::AmqpTimestamp(std::chrono::milliseconds(0)).AsAmqpValue();
+        body["is_partition_empty"] = false;
+        response.SetBody(body.AsAmqpValue());
+        auto result = GetMessageSender().Send(response, GetListenerContext());
+        EXPECT_EQ(std::get<0>(result), Amqp::_internal::MessageSendStatus::Ok);
+      }
+    };
+
+    void ProducerDiscoveryFailure()
+    {
+      {
+        Amqp::Tests::MessageTests::AmqpServerMock server(AvailablePort());
+        server.AddServiceEndpoint(std::make_shared<PropertiesEndpoint>(true));
+        server.StartListening();
+        {
+          ProducerClientOptions options;
+          options.RetryOptions.MaxRetries = 0;
+          ProducerClient producer(
+              "Endpoint=sb://localhost:" + std::to_string(server.GetPort())
+                  + "/;SharedAccessKeyName=test;SharedAccessKey=YWJj;UseDevelopmentEmulator=true",
+              "hub",
+              options);
+          auto context = Azure::Core::Context().WithDeadline(
+              std::chrono::system_clock::now() + std::chrono::seconds(10));
+          EXPECT_THROW(producer.GetEventHubProperties(context), EventHubsException);
+          producer.Close(context);
+          producer.Close(context);
+        }
+        server.StopListening();
+      }
+      Amqp::Common::_detail::GlobalStateHolder::GlobalStateInstance()->AssertIdle();
+    }
+
+    std::atomic<int> endCalls{0};
+    int CountEnd(SESSION_HANDLE session, char const* condition, char const* description)
+    {
+      ++endCalls;
+      return session_end(session, condition, description);
+    }
+
+    template <typename T> void RequireReady(std::future<T>& operation)
+    {
+      if (operation.wait_for(std::chrono::seconds(10)) != std::future_status::ready)
+      {
+        ADD_FAILURE() << "Concurrent properties operation did not finish";
+        // Avoid hanging in a future destructor; the parent death test reports the failure.
+        std::exit(1);
+      }
+    }
+
+    void ConcurrentPropertiesClose(
+        bool partitionProperties,
+        bool cancelCaller,
+        bool pauseBeforeSend = false)
+    {
+      {
+        std::promise<void> requestReceived;
+        auto received = requestReceived.get_future();
+        std::promise<void> resumeQuery;
+        auto resume = resumeQuery.get_future();
+        if (pauseBeforeSend)
+        {
+          Azure::Core::Diagnostics::Logger::SetLevel(
+              Azure::Core::Diagnostics::Logger::Level::Verbose);
+          Azure::Core::Diagnostics::Logger::SetListener(
+              [&](Azure::Core::Diagnostics::Logger::Level, std::string const& line) {
+                if (line.find("ManagementClient::ExecuteOperation:") != std::string::npos
+                    && line.find("Create Queue for request.") != std::string::npos)
+                {
+                  // Pause inside ExecuteOperation, before it sends, to expose destruction of an
+                  // active management client even when ordinary close callbacks finish quickly.
+                  requestReceived.set_value();
+                  RequireReady(resume);
+                }
+              });
+        }
+        Amqp::Tests::MessageTests::AmqpServerMock server(AvailablePort());
+        server.AddServiceEndpoint(
+            std::make_shared<PropertiesEndpoint>(
+                false, pauseBeforeSend ? nullptr : &requestReceived));
+        server.StartListening();
+        {
+          Amqp::_internal::ConnectionOptions options;
+          options.Port = server.GetPort();
+          Amqp::_internal::Connection connection("localhost", nullptr, options);
+          auto session = Amqp::_detail::SessionFactory::CreateFromInternal(
+              std::make_shared<Amqp::_detail::SessionImpl>(
+                  Amqp::_detail::ConnectionFactory::GetImpl(connection),
+                  Amqp::_internal::SessionOptions{},
+                  nullptr,
+                  CountEnd));
+          endCalls = 0;
+          EventHubsPropertiesClient properties(session, "hub");
+          // No deadline: Close must not depend on the caller eventually timing out.
+          Azure::Core::Context queryContext;
+          auto query = std::async(std::launch::async, [&]() {
+            if (partitionProperties)
+            {
+              properties.GetEventHubsPartitionProperties("hub", "0", queryContext);
+            }
+            else
+            {
+              properties.GetEventHubsProperties("hub", queryContext);
+            }
+          });
+          RequireReady(received);
+          received.get();
+          if (cancelCaller)
+          {
+            queryContext.Cancel();
+          }
+          std::promise<void> closeStarted;
+          auto started = closeStarted.get_future();
+          auto close = std::async(std::launch::async, [&]() {
+            closeStarted.set_value();
+            properties.Close({});
+          });
+          RequireReady(started);
+          if (pauseBeforeSend)
+          {
+            EXPECT_EQ(close.wait_for(std::chrono::milliseconds(200)), std::future_status::timeout);
+            EXPECT_EQ(endCalls.load(), 0);
+            resumeQuery.set_value();
+          }
+          RequireReady(query);
+          EXPECT_ANY_THROW(query.get());
+          RequireReady(close);
+          EXPECT_NO_THROW(close.get());
+          EXPECT_EQ(queryContext.IsCancelled(), cancelCaller);
+          properties.Close({});
+          EXPECT_EQ(endCalls.load(), 1);
+
+          // A stale properties-client reference must not reopen the session after invalidation.
+          EXPECT_THROW(
+              properties.GetEventHubsProperties("hub", {}), Core::OperationCancelledException);
+          EXPECT_THROW(
+              properties.GetEventHubsPartitionProperties("hub", "0", {}),
+              Core::OperationCancelledException);
+          EXPECT_EQ(endCalls.load(), 1);
+          Azure::Core::Diagnostics::Logger::SetListener(nullptr);
+        }
+        server.StopListening();
+      }
+      Amqp::Common::_detail::GlobalStateHolder::GlobalStateInstance()->AssertIdle();
+    }
+
+    void CancelWaitingPropertiesQuery(
+        bool firstPartitionProperties,
+        bool secondPartitionProperties,
+        bool expireDeadline)
+    {
+      {
+        std::promise<void> requestReceived;
+        auto received = requestReceived.get_future();
+        auto endpoint = std::make_shared<PropertiesEndpoint>(false, &requestReceived);
+        Amqp::Tests::MessageTests::AmqpServerMock server(AvailablePort());
+        server.AddServiceEndpoint(endpoint);
+        server.StartListening();
+        {
+          Amqp::_internal::ConnectionOptions options;
+          options.Port = server.GetPort();
+          Amqp::_internal::Connection connection("localhost", nullptr, options);
+          auto session = Amqp::_detail::SessionFactory::CreateFromInternal(
+              std::make_shared<Amqp::_detail::SessionImpl>(
+                  Amqp::_detail::ConnectionFactory::GetImpl(connection),
+                  Amqp::_internal::SessionOptions{},
+                  nullptr,
+                  CountEnd));
+          endCalls = 0;
+          EventHubsPropertiesClient properties(session, "hub");
+          auto query = [&](bool partitionProperties, Core::Context const& context) {
+            if (partitionProperties)
+            {
+              properties.GetEventHubsPartitionProperties("hub", "0", context);
+            }
+            else
+            {
+              properties.GetEventHubsProperties("hub", context);
+            }
+          };
+          // Establish an unanswered operation with no deadline before starting the waiter.
+          Core::Context firstContext;
+          auto first = std::async(
+              std::launch::async, [&]() { query(firstPartitionProperties, firstContext); });
+          RequireReady(received);
+          received.get();
+
+          Core::Context secondContext;
+          std::promise<void> secondStarted;
+          auto started = secondStarted.get_future();
+          auto second = std::async(std::launch::async, [&]() {
+            auto context = expireDeadline
+                ? secondContext.WithDeadline(
+                      std::chrono::system_clock::now() + std::chrono::seconds(1))
+                : secondContext;
+            secondStarted.set_value();
+            query(secondPartitionProperties, context);
+          });
+          RequireReady(started);
+          EXPECT_EQ(second.wait_for(std::chrono::milliseconds(100)), std::future_status::timeout);
+          if (!expireDeadline)
+          {
+            secondContext.Cancel();
+          }
+          // The watchdog exits the child on regression, without hanging on future destruction.
+          RequireReady(second);
+          EXPECT_THROW(second.get(), Core::OperationCancelledException);
+          EXPECT_EQ(first.wait_for(std::chrono::milliseconds(100)), std::future_status::timeout);
+          EXPECT_FALSE(firstContext.IsCancelled());
+          EXPECT_EQ(endpoint->RequestCount(), 1);
+          EXPECT_EQ(endCalls.load(), 0);
+
+          auto close = std::async(std::launch::async, [&]() { properties.Close({}); });
+          RequireReady(first);
+          EXPECT_THROW(first.get(), Core::OperationCancelledException);
+          RequireReady(close);
+          EXPECT_NO_THROW(close.get());
+          EXPECT_FALSE(firstContext.IsCancelled());
+          properties.Close({});
+          EXPECT_EQ(endCalls.load(), 1);
+          EXPECT_THROW(
+              properties.GetEventHubsProperties("hub", {}), Core::OperationCancelledException);
+          EXPECT_THROW(
+              properties.GetEventHubsPartitionProperties("hub", "0", {}),
+              Core::OperationCancelledException);
+          EXPECT_EQ(endCalls.load(), 1);
+          EXPECT_EQ(endpoint->RequestCount(), 1);
+        }
+        server.StopListening();
+      }
+      Amqp::Common::_detail::GlobalStateHolder::GlobalStateInstance()->AssertIdle();
+    }
+
+    void SuccessfulDiscovery(bool cancelledClose)
+    {
+      std::atomic<bool> closeFailureLogged{false};
+      Azure::Core::Diagnostics::Logger::SetLevel(Azure::Core::Diagnostics::Logger::Level::Warning);
+      Azure::Core::Diagnostics::Logger::SetListener(
+          [&](Azure::Core::Diagnostics::Logger::Level, std::string const& line) {
+            if (line.find("Exception while closing the management client") != std::string::npos)
+            {
+              closeFailureLogged = true;
+            }
+          });
+      {
+        Amqp::Tests::MessageTests::AmqpServerMock server(AvailablePort());
+        server.AddServiceEndpoint(std::make_shared<PropertiesEndpoint>());
+        server.StartListening();
+        {
+          Amqp::_internal::ConnectionOptions options;
+          options.Port = server.GetPort();
+          Amqp::_internal::Connection connection("localhost", nullptr, options);
+          auto session = Amqp::_detail::SessionFactory::CreateFromInternal(
+              std::make_shared<Amqp::_detail::SessionImpl>(
+                  Amqp::_detail::ConnectionFactory::GetImpl(connection),
+                  Amqp::_internal::SessionOptions{},
+                  nullptr,
+                  CountEnd));
+          endCalls = 0;
+          {
+            EventHubsPropertiesClient properties(session, "hub");
+            auto context = Azure::Core::Context().WithDeadline(
+                std::chrono::system_clock::now() + std::chrono::seconds(10));
+            auto result = properties.GetEventHubsProperties("hub", context);
+            EXPECT_EQ(result.Name, "hub");
+            EXPECT_EQ(result.PartitionIds, (std::vector<std::string>{"0", "1"}));
+            auto partition = properties.GetEventHubsPartitionProperties("hub", "0", context);
+            EXPECT_EQ(partition.PartitionId, "0");
+            EXPECT_EQ(partition.LastEnqueuedOffset, "1");
+            if (cancelledClose)
+            {
+              context.Cancel();
+              // Stop client connection polling after draining its current poll. The peer cannot
+              // complete DETACH before the cancelled close waits, so the close reliably throws.
+              Amqp::Common::_detail::GlobalStateHolder::GlobalStateInstance()->RemovePollable(
+                  Amqp::_detail::ConnectionFactory::GetImpl(connection));
+            }
+            properties.Close(context);
+            properties.Close(context);
+          }
+          EXPECT_EQ(endCalls.load(), 1);
+        }
+        server.StopListening();
+      }
+      Azure::Core::Diagnostics::Logger::SetListener(nullptr);
+      EXPECT_EQ(closeFailureLogged.load(), cancelledClose);
+      Amqp::Common::_detail::GlobalStateHolder::GlobalStateInstance()->AssertIdle();
+    }
+
+    void PartialInitialization(bool explicitClose, bool failEnd)
+    {
+      std::weak_ptr<Amqp::_detail::ConnectionImpl> connectionLifetime;
+      {
+        Amqp::Tests::CleanupTestListener server;
+        Amqp::_internal::ConnectionOptions options;
+        options.Port = server.GetPort();
+        Amqp::_internal::Connection connection("localhost", nullptr, options);
+        connectionLifetime = Amqp::_detail::ConnectionFactory::GetImpl(connection);
+        auto session = Amqp::_detail::SessionFactory::CreateFromInternal(
+            std::make_shared<Amqp::_detail::SessionImpl>(
+                Amqp::_detail::ConnectionFactory::GetImpl(connection),
+                Amqp::_internal::SessionOptions{},
+                nullptr,
+                failEnd ? FailEnd : session_end));
+        {
+          EventHubsPropertiesClient properties(session, "hub", FailCreation);
+          // Retrying discovery must not acquire a second polling reference.
+          for (int attempt = 0; attempt != 2; ++attempt)
+          {
+            try
+            {
+              properties.GetEventHubsProperties("hub", {});
+              ADD_FAILURE() << "Expected management creation to fail";
+            }
+            catch (std::runtime_error const& error)
+            {
+              EXPECT_STREQ(error.what(), "injected management creation failure");
+            }
+          }
+          if (explicitClose)
+          {
+            properties.Close({});
+            properties.Close({});
+          }
+        }
+      }
+      Amqp::Common::_detail::GlobalStateHolder::GlobalStateInstance()->AssertIdle();
+      EXPECT_TRUE(connectionLifetime.expired());
+    }
+
+    void FailedOpen(bool failEnd)
+    {
+      {
+        Amqp::Tests::CleanupTestListener server;
+        Amqp::_internal::ConnectionOptions options;
+        options.Port = server.GetPort();
+        Amqp::_internal::Connection connection("localhost", nullptr, options);
+        auto session = Amqp::_detail::SessionFactory::CreateFromInternal(
+            std::make_shared<Amqp::_detail::SessionImpl>(
+                Amqp::_detail::ConnectionFactory::GetImpl(connection),
+                Amqp::_internal::SessionOptions{},
+                nullptr,
+                failEnd ? FailEnd : session_end));
+        EventHubsPropertiesClient properties(session, "hub");
+        // Enter initialization before cancellation so this still exercises a failed open.
+        auto context = Azure::Core::Context().WithDeadline(
+            std::chrono::system_clock::now() + std::chrono::seconds(1));
+        try
+        {
+          properties.GetEventHubsProperties("hub", context);
+          ADD_FAILURE() << "Expected management open to fail";
+        }
+        catch (std::runtime_error const& error)
+        {
+          EXPECT_STREQ(error.what(), "Could not open Management client");
+        }
+        properties.Close({});
+        properties.Close({});
+      }
+      Amqp::Common::_detail::GlobalStateHolder::GlobalStateInstance()->AssertIdle();
+    }
+  } // namespace
+
+  class PropertiesCleanupDeathTest : public ::testing::Test {
+  protected:
+    void SetUp() override
+    {
+      // Re-exec the child instead of inheriting AMQP polling-thread mutexes across fork.
+      ::testing::FLAGS_gtest_death_test_style = "threadsafe";
+    }
+  };
+
+  TEST_F(PropertiesCleanupDeathTest, SuccessfulDiscoveryAndRepeatedClose)
+  {
+    EXPECT_EXIT(
+        {
+          SuccessfulDiscovery(false);
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+
+  class ConcurrentPropertiesCleanupDeathTest
+      : public PropertiesCleanupDeathTest,
+        public ::testing::WithParamInterface<std::tuple<bool, bool>> {};
+
+  TEST_P(ConcurrentPropertiesCleanupDeathTest, CloseDuringPendingQuery)
+  {
+    auto const partitionProperties = std::get<0>(GetParam());
+    auto const cancelCaller = std::get<1>(GetParam());
+    EXPECT_EXIT(
+        {
+          ConcurrentPropertiesClose(partitionProperties, cancelCaller);
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+
+  TEST_P(ConcurrentPropertiesCleanupDeathTest, CloseBeforeManagementSend)
+  {
+    auto const partitionProperties = std::get<0>(GetParam());
+    auto const cancelCaller = std::get<1>(GetParam());
+    EXPECT_EXIT(
+        {
+          ConcurrentPropertiesClose(partitionProperties, cancelCaller, true);
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+
+  INSTANTIATE_TEST_SUITE_P(
+      PropertiesAndPartitions,
+      ConcurrentPropertiesCleanupDeathTest,
+      ::testing::Combine(::testing::Bool(), ::testing::Bool()));
+
+  class WaitingPropertiesCancellationDeathTest
+      : public PropertiesCleanupDeathTest,
+        public ::testing::WithParamInterface<std::tuple<bool, bool, bool>> {};
+
+  TEST_P(WaitingPropertiesCancellationDeathTest, CancelWhileAnotherQueryIsPending)
+  {
+    auto const firstPartitionProperties = std::get<0>(GetParam());
+    auto const secondPartitionProperties = std::get<1>(GetParam());
+    auto const expireDeadline = std::get<2>(GetParam());
+    EXPECT_EXIT(
+        {
+          CancelWaitingPropertiesQuery(
+              firstPartitionProperties, secondPartitionProperties, expireDeadline);
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+
+  INSTANTIATE_TEST_SUITE_P(
+      PropertiesAndPartitions,
+      WaitingPropertiesCancellationDeathTest,
+      ::testing::Combine(::testing::Bool(), ::testing::Bool(), ::testing::Bool()));
+
+  TEST_F(PropertiesCleanupDeathTest, ManagementCloseFailureStillEndsSession)
+  {
+    EXPECT_EXIT(
+        {
+          SuccessfulDiscovery(true);
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+
+  TEST_F(PropertiesCleanupDeathTest, CreationFailureThenRepeatedClose)
+  {
+    EXPECT_EXIT(
+        {
+          PartialInitialization(true, false);
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+
+  TEST_F(PropertiesCleanupDeathTest, CreationFailureThenDestruction)
+  {
+    EXPECT_EXIT(
+        {
+          PartialInitialization(false, false);
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+
+  TEST_F(PropertiesCleanupDeathTest, CreationFailureAndSessionEndFailure)
+  {
+    EXPECT_EXIT(
+        {
+          PartialInitialization(true, true);
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+
+  TEST_F(PropertiesCleanupDeathTest, OpenFailureThenRepeatedClose)
+  {
+    EXPECT_EXIT(
+        {
+          FailedOpen(false);
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+
+  TEST_F(PropertiesCleanupDeathTest, OpenFailureAndSessionEndFailure)
+  {
+    EXPECT_EXIT(
+        {
+          FailedOpen(true);
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+
+  TEST_F(PropertiesCleanupDeathTest, ProducerDiscoveryFailureThenRepeatedClose)
+  {
+    EXPECT_EXIT(
+        {
+          ProducerDiscoveryFailure();
+          std::exit(::testing::Test::HasFailure() ? 1 : 0);
+        },
+        ::testing::ExitedWithCode(0),
+        "");
+  }
+}}}} // namespace Azure::Messaging::EventHubs::Test
+#endif