diff --git a/sdk/core/azure-core-amqp/test/ut/mock_amqp_server.hpp b/sdk/core/azure-core-amqp/test/ut/mock_amqp_server.hpp
index f50fb9f4b2..bb51335abd 100644
--- a/sdk/core/azure-core-amqp/test/ut/mock_amqp_server.hpp
+++ b/sdk/core/azure-core-amqp/test/ut/mock_amqp_server.hpp
@@ -16,6 +16,7 @@
 #include <azure/core/amqp/internal/network/socket_listener.hpp>
 #include <azure/core/amqp/internal/session.hpp>
 
+#include <atomic>
 #include <memory>
 
 #include <gtest/gtest.h>
@@ -57,8 +58,8 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
                                 public Azure::Core::Amqp::_internal::MessageSenderEvents {
     public:
       MockServiceEndpoint(std::string const& name, MockServiceEndpointOptions const& options)
-          : m_listenerContext{options.ListenerContext},
-            m_enableTrace{options.EnableTrace}, m_name{name}
+          : m_listenerContext{options.ListenerContext}, m_enableTrace{options.EnableTrace},
+            m_name{name}
       {
       }
 
@@ -185,6 +186,9 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
             sender.second->Close();
           }
         }
+        // Release link ownership before destroying sessions whose End can fail.
+        m_receiver.clear();
+        m_sender.clear();
       }
 
     protected:
@@ -625,6 +629,12 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
 
       void StopListening()
       {
+        if (!m_listening)
+        {
+          return;
+        }
+        // A destructor must not retry a partially completed shutdown.
+        m_listening = false;
         GTEST_LOG_(INFO) << "Stop listening";
         // Cancel the listener context, which will cause any WaitForXxx calls to exit.
 
@@ -643,17 +653,31 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
         {
           for (const auto& session : m_sessions)
           {
-            session->End({});
+            try
+            {
+              session->End({});
+            }
+            catch (std::runtime_error const& ex)
+            {
+              if (m_peerClosed.load() && std::string(ex.what()) == "Could not end session")
+              {
+                GTEST_LOG_(INFO) << "Session End after the mock peer closed reported: "
+                                 << ex.what();
+              }
+              else
+              {
+                ADD_FAILURE() << "Unexpected mock session shutdown failure: " << ex.what();
+              }
+            }
           }
-          // Note: clearing the sessions list destroys the session and calls session_end always, so
-          // it does not need to be called here.
+          // A failed End still needs native destruction to release its polling reference.
           m_sessions.clear();
         }
         for (const auto& connection : m_connections)
         {
           connection->Close({});
         }
-        m_listening = false;
+        m_connections.clear();
       }
 
       // Closes the connections without closing the links on them first, which is the shape of a
@@ -721,6 +745,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
         if (newState == Azure::Core::Amqp::_internal::ConnectionState::End
             || newState == Azure::Core::Amqp::_internal::ConnectionState::Error)
         {
+          m_peerClosed = true;
           // If the connection is closed, then we should close the connection.
           m_listenerContext.Cancel();
         }
@@ -796,6 +821,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
 
       bool m_enableTrace{true};
       bool m_listening{false};
+      std::atomic<bool> m_peerClosed{false};
 
       std::string m_connectionId;
       std::thread m_serverThread;
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
index f103766e9b..2663ff0764 100644
--- a/sdk/core/azure-core-amqp/test/ut/session_cleanup_tests.cpp
+++ b/sdk/core/azure-core-amqp/test/ut/session_cleanup_tests.cpp
@@ -65,6 +65,27 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
       }
       return session_end(session, condition, description);
     }
+
+    void RetryFailedEnd()
+    {
+      {
+        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();
+    }
   } // namespace
 
   class SessionCleanupDeathTest : public ::testing::Test {
@@ -102,23 +123,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
   {
     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();
+          RetryFailedEnd();
           std::exit(::testing::Test::HasFailure() ? 1 : 0);
         },
         ::testing::ExitedWithCode(0),
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
index ac245b8888..b09d24b12e 100644
--- a/sdk/eventhubs/azure-messaging-eventhubs/test/ut/properties_cleanup_test.cpp
+++ b/sdk/eventhubs/azure-messaging-eventhubs/test/ut/properties_cleanup_test.cpp
@@ -158,9 +158,8 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace Test {
               });
         }
         Amqp::Tests::MessageTests::AmqpServerMock server(AvailablePort());
-        server.AddServiceEndpoint(
-            std::make_shared<PropertiesEndpoint>(
-                false, pauseBeforeSend ? nullptr : &requestReceived));
+        server.AddServiceEndpoint(std::make_shared<PropertiesEndpoint>(
+            false, pauseBeforeSend ? nullptr : &requestReceived));
         server.StartListening();
         {
           Amqp::_internal::ConnectionOptions options;
@@ -364,6 +363,8 @@ namespace Azure { namespace Messaging { namespace EventHubs { namespace Test {
           EXPECT_EQ(endCalls.load(), 1);
         }
         server.StopListening();
+        server.StopListening();
+        Amqp::Common::_detail::GlobalStateHolder::GlobalStateInstance()->AssertIdle();
       }
       Azure::Core::Diagnostics::Logger::SetListener(nullptr);
       EXPECT_EQ(closeFailureLogged.load(), cancelledClose);