diff --git a/.changelog/feature-async-http-api.json b/.changelog/feature-async-http-api.json
new file mode 100644
index 000000000000..942204c9b01c
--- /dev/null
+++ b/.changelog/feature-async-http-api.json
@@ -0,0 +1,6 @@
+{
+  "type": "feature",
+  "category": "aws-cpp-sdk-core",
+  "contributor": "sbiscigl",
+  "description": "create async http api"
+}
diff --git a/src/aws-cpp-sdk-core/include/aws/core/http/HttpClient.h b/src/aws-cpp-sdk-core/include/aws/core/http/HttpClient.h
index 9451d20c0e1f..ceaa6dd27a9c 100644
--- a/src/aws-cpp-sdk-core/include/aws/core/http/HttpClient.h
+++ b/src/aws-cpp-sdk-core/include/aws/core/http/HttpClient.h
@@ -91,6 +91,28 @@ namespace Aws
                 false};
             }
 
+            /**
+             * Starts an http request and returns as soon as it has been started. onResponseComplete is
+             * invoked with the finished response, on whichever thread the implementation completes on.
+             * Only implemented by http clients that can drive a request without a thread of their own.
+             */
+            virtual Aws::Crt::Optional<Aws::Client::AWSError<Aws::Client::CoreErrors>> MakeRequestAsync(
+                const std::shared_ptr<HttpRequest>& request,
+                std::function<void(std::shared_ptr<HttpResponse>)> onResponseComplete,
+                std::function<void(const std::shared_ptr<Aws::Http::Connection>&)> onClientConnectionAvailable = nullptr,
+                Aws::Utils::RateLimits::RateLimiterInterface* readLimiter = nullptr,
+                Aws::Utils::RateLimits::RateLimiterInterface* writeLimiter = nullptr) const {
+              AWS_UNREFERENCED_PARAM(request);
+              AWS_UNREFERENCED_PARAM(onResponseComplete);
+              AWS_UNREFERENCED_PARAM(onClientConnectionAvailable);
+              AWS_UNREFERENCED_PARAM(readLimiter);
+              AWS_UNREFERENCED_PARAM(writeLimiter);
+              return Aws::Client::AWSError<Aws::Client::CoreErrors>{Aws::Client::CoreErrors::NOT_IMPLEMENTED,
+                "NotImplemented",
+                "async requests are not supported on this http client",
+                false};
+            }
+
         protected:
             bool m_bad;
 
diff --git a/src/aws-cpp-sdk-core/include/aws/core/http/HttpConnection.h b/src/aws-cpp-sdk-core/include/aws/core/http/HttpConnection.h
index 869b88270b10..2b7648319568 100644
--- a/src/aws-cpp-sdk-core/include/aws/core/http/HttpConnection.h
+++ b/src/aws-cpp-sdk-core/include/aws/core/http/HttpConnection.h
@@ -18,6 +18,7 @@ class AWS_CORE_API Connection {
   virtual std::shared_ptr<Aws::Http::ClientStream> NewClientStream(
     const std::shared_ptr<HttpRequest>& request,
     std::function<void(int errorCode)> onStreamComplete) = 0;
+  virtual void Close() = 0;
 };
 }  // namespace Http
 }  // namespace Aws
diff --git a/src/aws-cpp-sdk-core/include/aws/core/http/crt/CRTHttpClient.h b/src/aws-cpp-sdk-core/include/aws/core/http/crt/CRTHttpClient.h
index 55e8c1d459d8..88b08fda8b4d 100644
--- a/src/aws-cpp-sdk-core/include/aws/core/http/crt/CRTHttpClient.h
+++ b/src/aws-cpp-sdk-core/include/aws/core/http/crt/CRTHttpClient.h
@@ -33,6 +33,14 @@ namespace Aws
         struct ClientConfiguration;
     } // namespace Client
 
+    namespace Utils
+    {
+        namespace Threading
+        {
+            class WaitGroup;
+        }
+    }
+
     namespace Http
     {
         /**
@@ -52,6 +60,13 @@ namespace Aws
                 Aws::Utils::RateLimits::RateLimiterInterface* readLimiter,
                 Aws::Utils::RateLimits::RateLimiterInterface* writeLimiter) const override;
 
+            Aws::Crt::Optional<Aws::Client::AWSError<Aws::Client::CoreErrors>> MakeRequestAsync(
+                const std::shared_ptr<HttpRequest>& request,
+                std::function<void(std::shared_ptr<HttpResponse>)> onResponseComplete,
+                std::function<void(const std::shared_ptr<Aws::Http::Connection>&)> onClientConnectionAvailable = nullptr,
+                Aws::Utils::RateLimits::RateLimiterInterface* readLimiter = nullptr,
+                Aws::Utils::RateLimits::RateLimiterInterface* writeLimiter = nullptr) const override;
+
             bool IsDefaultAwsHttpClient() const override { return true; }
 
             Aws::Crt::Optional<Aws::Client::AWSError<Aws::Client::CoreErrors>> AcquireConnection(
@@ -71,6 +86,8 @@ namespace Aws
             Crt::Io::ClientBootstrap& m_bootstrap;
             Client::ClientConfiguration m_configuration;
 
+            std::shared_ptr<Aws::Utils::Threading::WaitGroup> m_requestLatch;
+
             std::shared_ptr<Crt::Http::HttpClientConnectionManager> GetWithCreateConnectionManagerForRequest(const std::shared_ptr<HttpRequest>& request, const Crt::Http::HttpClientConnectionOptions& connectionOptions) const;
             Crt::Http::HttpClientConnectionOptions CreateConnectionOptionsForRequest(const std::shared_ptr<HttpRequest>& request) const;
             void CheckAndInitializeProxySettings(const Aws::Client::ClientConfiguration& clientConfig);
diff --git a/src/aws-cpp-sdk-core/include/aws/core/utils/threading/WaitGroup.h b/src/aws-cpp-sdk-core/include/aws/core/utils/threading/WaitGroup.h
new file mode 100644
index 000000000000..43a3197394c9
--- /dev/null
+++ b/src/aws-cpp-sdk-core/include/aws/core/utils/threading/WaitGroup.h
@@ -0,0 +1,30 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#pragma once
+
+#include <aws/core/Core_EXPORTS.h>
+#include <aws/core/utils/memory/AWSMemory.h>
+
+#include <cstddef>
+
+namespace Aws {
+namespace Utils {
+namespace Threading {
+class AWS_CORE_API WaitGroup {
+ public:
+  WaitGroup();
+  ~WaitGroup();
+  void Add(size_t count = 1);
+  void Done();
+  void Wait();
+
+ private:
+  struct WaitGroupImpl;
+  Aws::UniquePtr<WaitGroupImpl> m_impl;
+};
+}  // namespace Threading
+}  // namespace Utils
+}  // namespace Aws
diff --git a/src/aws-cpp-sdk-core/source/http/crt/CRTHttpClient.cpp b/src/aws-cpp-sdk-core/source/http/crt/CRTHttpClient.cpp
index 48643255910d..4bde4e8620b4 100644
--- a/src/aws-cpp-sdk-core/source/http/crt/CRTHttpClient.cpp
+++ b/src/aws-cpp-sdk-core/source/http/crt/CRTHttpClient.cpp
@@ -9,43 +9,18 @@
 #include <aws/core/utils/crypto/Hash.h>
 #include <aws/core/utils/logging/LogMacros.h>
 #include <aws/core/utils/ratelimiter/RateLimiterInterface.h>
+#include <aws/core/utils/threading/Semaphore.h>
+#include <aws/core/utils/threading/WaitGroup.h>
 #include <aws/crt/http/HttpConnectionManager.h>
 #include <aws/crt/http/HttpRequestResponse.h>
 
+#include <atomic>
+#include <mutex>
 #include <utility>
 
 static const char *const CRT_HTTP_CLIENT_TAG = "CRTHttpClient";
 
 namespace {
-// Just a wrapper around a Condition Variable and a mutex, which handles wait and timed waits while protecting
-// from spurious wakeups.
-class AsyncWaiter {
- public:
-  AsyncWaiter() = default;
-  AsyncWaiter(const AsyncWaiter&) = delete;
-  AsyncWaiter& operator=(const AsyncWaiter&) = delete;
-
-  void Wakeup() {
-    std::lock_guard<std::mutex> locker(m_lock);
-    m_wakeupIntentional = true;
-    m_cvar.notify_one();
-  }
-
-  void WaitOnCompletion() {
-    std::unique_lock<std::mutex> uniqueLocker(m_lock);
-    m_cvar.wait(uniqueLocker, [this]() { return m_wakeupIntentional; });
-  }
-
-  bool WaitOnCompletionFor(const size_t ms) {
-    std::unique_lock<std::mutex> uniqueLocker(m_lock);
-    return m_cvar.wait_for(uniqueLocker, std::chrono::milliseconds(ms), [this]() { return m_wakeupIntentional; });
-  }
-
- private:
-  std::mutex m_lock;
-  std::condition_variable m_cvar;
-  bool m_wakeupIntentional{false};
-};
 
 void AddRequestMetadataToCrtRequest(const std::shared_ptr<Aws::Http::HttpRequest>& request,
                                     const std::shared_ptr<Aws::Crt::Http::HttpRequest>& crtRequest) {
@@ -131,31 +106,50 @@ void OnIncomingHeadersBlockDone(Aws::Crt::Http::HttpStream& stream, enum aws_htt
   response->SetResponseCode((Aws::Http::HttpResponseCode)stream.GetResponseStatusCode());
 }
 
-// Request is done. If there was an error set it, otherwise just wake up the cvar.
-void OnStreamComplete(Aws::Crt::Http::HttpStream&, int errorCode, AsyncWaiter& waiter,
-                      const std::shared_ptr<Aws::Http::HttpResponse>& response) {
+struct AsyncRequestState {
+  AsyncRequestState(const std::shared_ptr<Aws::Http::HttpRequest>& request,
+                    std::function<void(std::shared_ptr<Aws::Http::HttpResponse>)> onResponseComplete)
+      : request(request),
+        response(Aws::MakeShared<Aws::Http::Standard::StandardHttpResponse>(CRT_HTTP_CLIENT_TAG, request)),
+        crtRequest(Aws::Crt::MakeShared<Aws::Crt::Http::HttpRequest>(Aws::Crt::g_allocator)),
+        onComplete(std::move(onResponseComplete)) {}
+
+  std::shared_ptr<Aws::Http::HttpRequest> request;
+  std::shared_ptr<Aws::Http::HttpResponse> response;
+  std::shared_ptr<Aws::Crt::Http::HttpRequest> crtRequest;
+  std::function<void(std::shared_ptr<Aws::Http::HttpResponse>)> onComplete;
+  std::atomic<bool> finished{false};
+  std::shared_ptr<Aws::Utils::Threading::WaitGroup> latch;
+
+  void Finish() {
+    if (finished.exchange(true)) {
+      return;
+    }
+    if (onComplete) {
+      onComplete(response);
+    }
+    if (latch) {
+      latch->Done();
+    }
+  }
+};
+
+void OnStreamCompleteAsync(int errorCode, const std::shared_ptr<AsyncRequestState>& state) {
   if (errorCode) {
-    // TODO: get the right error parsed out.
-    response->SetClientErrorType(Aws::Client::CoreErrors::NETWORK_CONNECTION);
-    response->SetClientErrorMessage(aws_error_debug_str(errorCode));
+    state->response->SetClientErrorType(Aws::Client::CoreErrors::NETWORK_CONNECTION);
+    state->response->SetClientErrorMessage(aws_error_debug_str(errorCode));
   }
 
-  waiter.Wakeup();
+  state->Finish();
 }
 
-// if the connection acquisition failed, go ahead and fail the request and wakeup the cvar.
-// If it succeeded go ahead and make the request.
-void OnClientConnectionAvailable(std::shared_ptr<Aws::Crt::Http::HttpClientConnection> connection, int errorCode,
-                                 std::shared_ptr<Aws::Crt::Http::HttpClientConnection>& connectionReference,
-                                 Aws::Crt::Http::HttpRequestOptions& requestOptions, AsyncWaiter& waiter,
-                                 const std::shared_ptr<Aws::Http::HttpRequest>& request,
-                                 const std::shared_ptr<Aws::Http::HttpResponse>& response, const Aws::Http::HttpClient& client) {
-  bool shouldContinueRequest = client.ContinueRequest(*request);
-
+void OnClientConnectionAvailableAsync(const std::shared_ptr<Aws::Crt::Http::HttpClientConnection>& connection, int errorCode,
+                                      const std::shared_ptr<AsyncRequestState>& state, const Aws::Http::HttpClient& client) {
+  bool shouldContinueRequest = client.ContinueRequest(*state->request);
   if (!shouldContinueRequest) {
-    response->SetClientErrorType(Aws::Client::CoreErrors::USER_CANCELLED);
-    response->SetClientErrorMessage("Request cancelled by user's continuation handler");
-    waiter.Wakeup();
+    state->response->SetClientErrorType(Aws::Client::CoreErrors::USER_CANCELLED);
+    state->response->SetClientErrorMessage("Request cancelled by user's continuation handler");
+    state->Finish();
     return;
   }
 
@@ -163,8 +157,36 @@ void OnClientConnectionAvailable(std::shared_ptr<Aws::Crt::Http::HttpClientConne
   if (connection) {
     AWS_LOGSTREAM_DEBUG(CRT_HTTP_CLIENT_TAG, "Obtained connection handle " << (void*)connection.get());
 
+    Aws::Crt::Http::HttpRequestOptions requestOptions{};
+    requestOptions.request = state->crtRequest.get();
+
+    requestOptions.onIncomingBody = [&client, state](Aws::Crt::Http::HttpStream& stream, const Aws::Crt::ByteCursor& body) {
+      if (!client.ContinueRequest(*state->request) || !client.IsRequestProcessingEnabled()) {
+        AWS_LOGSTREAM_INFO(CRT_HTTP_CLIENT_TAG, "Request canceled. Canceling request by closing the connection.");
+        stream.GetConnection().Close();
+        return;
+      }
+      OnResponseBodyReceived(stream, body, state->response, state->request);
+    };
+
+    requestOptions.onIncomingHeaders = [state](Aws::Crt::Http::HttpStream& stream, enum aws_http_header_block block,
+                                               const Aws::Crt::Http::HttpHeader* headersArray, std::size_t headersCount) {
+      OnIncomingHeaders(stream, block, headersArray, headersCount, state->response);
+    };
+
+    requestOptions.onIncomingHeadersBlockDone = [state](Aws::Crt::Http::HttpStream& stream, enum aws_http_header_block block) {
+      OnIncomingHeadersBlockDone(stream, block, state->response);
+      auto& headersHandler = state->request->GetHeadersReceivedEventHandler();
+      if (headersHandler) {
+        headersHandler(state->request.get(), state->response.get());
+      }
+    };
+
+    requestOptions.onStreamComplete = [state](Aws::Crt::Http::HttpStream&, int streamErrorCode) {
+      OnStreamCompleteAsync(streamErrorCode, state);
+    };
+
     auto clientStream = connection->NewClientStream(requestOptions);
-    connectionReference = connection;
 
     if (clientStream && clientStream->Activate()) {
       return;
@@ -176,12 +198,19 @@ void OnClientConnectionAvailable(std::shared_ptr<Aws::Crt::Http::HttpClientConne
 
   const char* errorMsg = aws_error_debug_str(finalErrorCode);
   AWS_LOGSTREAM_ERROR(CRT_HTTP_CLIENT_TAG, "Obtaining connection failed because " << errorMsg);
-  response->SetClientErrorType(Aws::Client::CoreErrors::NETWORK_CONNECTION);
-  response->SetClientErrorMessage(errorMsg);
+  state->response->SetClientErrorType(Aws::Client::CoreErrors::NETWORK_CONNECTION);
+  state->response->SetClientErrorMessage(errorMsg);
 
-  waiter.Wakeup();
+  state->Finish();
 }
 
+struct SyncRequestState {
+  Aws::Utils::Threading::Semaphore signal{0, 1};
+  std::shared_ptr<Aws::Http::HttpResponse> response;
+  std::mutex connectionLock;
+  std::shared_ptr<Aws::Http::Connection> connection;
+};
+
 class CRTClientStream : public Aws::Http::ClientStream {
  public:
   CRTClientStream(std::shared_ptr<Aws::Crt::Http::HttpClientStream> stream, std::shared_ptr<Aws::Http::HttpResponse> response,
@@ -221,6 +250,12 @@ class CRTConnection : public Aws::Http::Connection {
   explicit CRTConnection(std::shared_ptr<Aws::Crt::Http::HttpClientConnection> connection) : m_connection(std::move(connection)) {}
   ~CRTConnection() override = default;
 
+  void Close() override {
+    if (m_connection) {
+      m_connection->Close();
+    }
+  }
+
   std::shared_ptr<Aws::Http::ClientStream> NewClientStream(const std::shared_ptr<Aws::Http::HttpRequest>& request,
                                                            std::function<void(int errorCode)> onStreamComplete) override {
     auto crtRequest = Aws::Crt::MakeShared<Aws::Crt::Http::HttpRequest>(Aws::Crt::g_allocator);
@@ -360,7 +395,8 @@ namespace Aws
     namespace Http
     {
         CRTHttpClient::CRTHttpClient(const Aws::Client::ClientConfiguration& clientConfig, Crt::Io::ClientBootstrap& bootstrap) : 
-            HttpClient(), m_context(), m_proxyOptions(), m_bootstrap(bootstrap), m_configuration(clientConfig)
+            HttpClient(), m_context(), m_proxyOptions(), m_bootstrap(bootstrap), m_configuration(clientConfig),
+            m_requestLatch(Aws::MakeShared<Aws::Utils::Threading::WaitGroup>(CRT_HTTP_CLIENT_TAG))
         {
             //first need to figure TLS out...
             Crt::Io::TlsContextOptions tlsContextOptions = Crt::Io::TlsContextOptions::InitDefaultClient();
@@ -405,10 +441,10 @@ namespace Aws
             m_context = std::move(newContext);
         }
 
-        // this isn't entirely necessary, but if you want to be nice to debuggers and memory checkers, let's go ahead
-        // and shut everything down cleanly.
         CRTHttpClient::~CRTHttpClient()
         {
+            m_requestLatch->Wait();
+
             Aws::Vector<std::future<void>> shutdownFutures;
 
             for (auto& managerPair : m_connectionPools) 
@@ -425,124 +461,105 @@ namespace Aws
             m_connectionPools.clear();
         }
 
-        std::shared_ptr<HttpResponse> CRTHttpClient::MakeRequest(const std::shared_ptr<HttpRequest>& request,
-                                                                 Aws::Utils::RateLimits::RateLimiterInterface*,
-                                                                 Aws::Utils::RateLimits::RateLimiterInterface*) const
+        Aws::Crt::Optional<Aws::Client::AWSError<Aws::Client::CoreErrors>> CRTHttpClient::MakeRequestAsync(
+            const std::shared_ptr<HttpRequest>& request,
+            std::function<void(std::shared_ptr<HttpResponse>)> onResponseComplete,
+            std::function<void(const std::shared_ptr<Aws::Http::Connection>&)> onClientConnectionAvailable,
+            Aws::Utils::RateLimits::RateLimiterInterface*,
+            Aws::Utils::RateLimits::RateLimiterInterface*) const
         {
-            auto crtRequest = Crt::MakeShared<Crt::Http::HttpRequest>(Crt::g_allocator);
-            auto response = Aws::MakeShared<Standard::StandardHttpResponse>(CRT_HTTP_CLIENT_TAG, request);
+            auto state = Aws::MakeShared<AsyncRequestState>(CRT_HTTP_CLIENT_TAG, request, std::move(onResponseComplete));
 
             auto requestConnOptions = CreateConnectionOptionsForRequest(request);
             auto connectionManager = GetWithCreateConnectionManagerForRequest(request, requestConnOptions);
 
             if (!connectionManager)
             {
-                response->SetClientErrorMessage(aws_error_debug_str(aws_last_error()));
-                response->SetClientErrorType(Client::CoreErrors::INVALID_PARAMETER_COMBINATION);
-                return response;
+                return Aws::Client::AWSError<Aws::Client::CoreErrors>{
+                    Aws::Client::CoreErrors::INVALID_PARAMETER_COMBINATION,
+                    "InvalidParameterCombination",
+                    aws_error_debug_str(aws_last_error()),
+                    false};
             }
-            AddRequestMetadataToCrtRequest(request, crtRequest);
+
+            AddRequestMetadataToCrtRequest(request, state->crtRequest);
 
             // Set the request body stream on the crt request. Setup the write rate limiter if present
             if (request->GetContentBody())
             {
                 bool isStreaming = request->IsEventStreamRequest();
-                crtRequest->SetBody(Aws::MakeShared<SDKAdaptingInputStream>(CRT_HTTP_CLIENT_TAG, m_configuration.writeRateLimiter, request->GetContentBody(), *this, *request, isStreaming));
+                state->crtRequest->SetBody(Aws::MakeShared<SDKAdaptingInputStream>(CRT_HTTP_CLIENT_TAG, m_configuration.writeRateLimiter, request->GetContentBody(), *this, *request, isStreaming));
             }
 
-            Crt::Http::HttpRequestOptions requestOptions;
-            requestOptions.request = crtRequest.get();
+            state->latch = m_requestLatch;
+            m_requestLatch->Add();
 
-            requestOptions.onIncomingBody =
-                [this, request, response](Crt::Http::HttpStream& stream, const Crt::ByteCursor& body)
-            {
-                if (!ContinueRequest(*request) || !IsRequestProcessingEnabled())
-                {
-                    AWS_LOGSTREAM_INFO(CRT_HTTP_CLIENT_TAG, "Request canceled. Canceling request by closing the connection.");
-                    stream.GetConnection().Close();
-                    return;
-                }
-                OnResponseBodyReceived(stream, body, response, request);
-            };
+            connectionManager->AcquireConnection(
+                    [state, this, onClientConnectionAvailable](std::shared_ptr<Crt::Http::HttpClientConnection> connection, int errorCode)
+                    {
+                        if (connection && onClientConnectionAvailable)
+                        {
+                            onClientConnectionAvailable(Aws::MakeShared<CRTConnection>(CRT_HTTP_CLIENT_TAG, connection));
+                        }
+                        OnClientConnectionAvailableAsync(connection, errorCode, state, *this);
+                    });
 
-            requestOptions.onIncomingHeaders =
-                [response](Crt::Http::HttpStream& stream, enum aws_http_header_block block, const Crt::Http::HttpHeader* headersArray, std::size_t headersCount)
-            {
-                OnIncomingHeaders(stream, block, headersArray, headersCount, response);
-            };
+            return {};
+        }
 
-            // This will arrive at or around the same time as the headers. Use it to set the response code on the response
-            requestOptions.onIncomingHeadersBlockDone =
-                [request, response](Crt::Http::HttpStream& stream, enum aws_http_header_block block)
-            {
-                OnIncomingHeadersBlockDone(stream, block, response);
-                auto& headersHandler = request->GetHeadersReceivedEventHandler();
-                if (headersHandler)
-                {
-                    headersHandler(request.get(), response.get());
-                }
-            };
+        std::shared_ptr<HttpResponse> CRTHttpClient::MakeRequest(const std::shared_ptr<HttpRequest>& request,
+                                                                 Aws::Utils::RateLimits::RateLimiterInterface* readLimiter,
+                                                                 Aws::Utils::RateLimits::RateLimiterInterface* writeLimiter) const
+        {
+            auto sync = Aws::MakeShared<SyncRequestState>(CRT_HTTP_CLIENT_TAG);
 
-            // CRT client is async only so we'll need to do the synchronous part ourselves.
-            // We'll use a condition variable and wait on it until the request completes or errors out.
-            AsyncWaiter waiter;
+            auto error = MakeRequestAsync(request,
+                [sync](std::shared_ptr<HttpResponse> response)
+                {
+                    sync->response = std::move(response);
+                    sync->signal.Release();
+                },
+                [sync](const std::shared_ptr<Aws::Http::Connection>& connection)
+                {
+                    std::lock_guard<std::mutex> lock(sync->connectionLock);
+                    sync->connection = connection;
+                },
+                readLimiter, writeLimiter);
 
-            requestOptions.onStreamComplete =
-                [&waiter, &response](Crt::Http::HttpStream& stream, int errorCode)
+            if (error.has_value())
             {
-                OnStreamComplete(stream, errorCode, waiter, response);
-            };
-
-            std::shared_ptr<Crt::Http::HttpClientConnection> connectionRef(nullptr);
-
-            // now we finally have the request, get a connection and make the request.
-            connectionManager->AcquireConnection(
-                    [&connectionRef, &requestOptions, response, &waiter, request, this]
-                    (std::shared_ptr<Crt::Http::HttpClientConnection> connection, int errorCode)
-                    {
-                        OnClientConnectionAvailable(connection, errorCode, connectionRef, requestOptions, waiter, request, response, *this);
-                    });
+                auto response = Aws::MakeShared<Standard::StandardHttpResponse>(CRT_HTTP_CLIENT_TAG, request);
+                response->SetClientErrorType(error->GetErrorType());
+                response->SetClientErrorMessage(error->GetMessage());
+                return response;
+            }
 
-            bool waiterTimedOut = false;
-            // Naive http request timeout implementation. This doesn't factor in how long it took to get the connection from the pool, and
-            // I'm undecided on the queueing theory implications of this decision so if this turns out to be the wrong granularity
-            // this is the section of code you should be changing. You can probably get "close" by having an additional
-            // atomic (not necessarily full on atomics implementation, but it needs to be the size of a WORD if it's not)
-            // counter that gets incremented in the acquireConnection callback as long as your connection timeout
-            // is shorter than your request timeout. Even if it's not, that would handle like.... 4-5 nines of getting this right.
-            // since in the worst case scenario, your connect timeout got preempted by the request timeout, and is it really worth
-            // all that effort if that's the worst thing that can happen?
-            if (m_configuration.requestTimeoutMs > 0 )
+            if (m_configuration.requestTimeoutMs > 0)
             {
-                waiterTimedOut = !waiter.WaitOnCompletionFor(m_configuration.requestTimeoutMs);
-
-                // if this is true, the waiter timed out without a terminal condition being woken up.
-                if (waiterTimedOut)
+                if (!sync->signal.WaitOneFor(m_configuration.requestTimeoutMs))
                 {
-                    // close the connection if it's still there so we can expedite anything we're waiting on.
-                    if (connectionRef)
+                    std::shared_ptr<Aws::Http::Connection> connection;
                     {
-                        connectionRef->Close();
+                        std::lock_guard<std::mutex> lock(sync->connectionLock);
+                        connection = sync->connection;
                     }
+                    if (connection)
+                    {
+                        connection->Close();
+                    }
+                    auto response = Aws::MakeShared<Standard::StandardHttpResponse>(CRT_HTTP_CLIENT_TAG, request);
+                    response->SetClientErrorType(Aws::Client::CoreErrors::REQUEST_TIMEOUT);
+                    response->SetClientErrorMessage("Request Timeout Has Expired");
+                    return response;
                 }
             }
-
-            // always wait, even if the above section timed out, because Wakeup() hasn't yet been called,
-            // and this means we're still waiting on some queued up callbacks to fire.
-            // going past this point before that occurs will cause a segfault when the callback DOES finally fire
-            // since the waiter is on the stack.
-            waiter.WaitOnCompletion();
-
-            // now handle if we timed out or not.
-            if (waiterTimedOut)
+            else
             {
-                response->SetClientErrorType(
-                        Aws::Client::CoreErrors::REQUEST_TIMEOUT);
-                response->SetClientErrorMessage("Request Timeout Has Expired");
+                sync->signal.WaitOne();
             }
 
             // TODO: is VOX support still a thing? If so we need to add the metrics for it.
-            return response;
+            return sync->response;
         }
 
         Aws::String CRTHttpClient::ResolveConnectionPoolKey(const URI& uri)
diff --git a/src/aws-cpp-sdk-core/source/utils/threading/WaitGroup.cpp b/src/aws-cpp-sdk-core/source/utils/threading/WaitGroup.cpp
new file mode 100644
index 000000000000..49ed3b46f444
--- /dev/null
+++ b/src/aws-cpp-sdk-core/source/utils/threading/WaitGroup.cpp
@@ -0,0 +1,55 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#include <aws/core/utils/threading/WaitGroup.h>
+
+#include <condition_variable>
+#include <mutex>
+
+namespace {
+const char WAIT_GROUP_TAG[] = "WaitGroup";
+}
+
+namespace Aws {
+namespace Utils {
+namespace Threading {
+struct WaitGroup::WaitGroupImpl {
+  void Add(size_t delta) {
+    std::lock_guard<std::mutex> lock(mutex);
+    count += delta;
+  }
+
+  void Done() {
+    std::lock_guard<std::mutex> lock(mutex);
+    if (count > 0) {
+      --count;
+    }
+    if (count == 0) {
+      syncPoint.notify_all();
+    }
+  }
+
+  void Wait() {
+    std::unique_lock<std::mutex> lock(mutex);
+    syncPoint.wait(lock, [this]() { return count == 0; });
+  }
+
+  std::mutex mutex;
+  std::condition_variable syncPoint;
+  size_t count{0};
+};
+
+WaitGroup::WaitGroup() : m_impl(Aws::MakeUnique<WaitGroupImpl>(WAIT_GROUP_TAG)) {}
+
+WaitGroup::~WaitGroup() = default;
+
+void WaitGroup::Add(size_t count) { m_impl->Add(count); }
+
+void WaitGroup::Done() { m_impl->Done(); }
+
+void WaitGroup::Wait() { m_impl->Wait(); }
+}  // namespace Threading
+}  // namespace Utils
+}  // namespace Aws
diff --git a/tests/aws-cpp-sdk-core-integration-tests/AsyncHttpRequestIntegrationTest.cpp b/tests/aws-cpp-sdk-core-integration-tests/AsyncHttpRequestIntegrationTest.cpp
new file mode 100644
index 000000000000..294032653d86
--- /dev/null
+++ b/tests/aws-cpp-sdk-core-integration-tests/AsyncHttpRequestIntegrationTest.cpp
@@ -0,0 +1,150 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#include <aws/core/Aws.h>
+#include <aws/core/client/ClientConfiguration.h>
+#include <aws/core/http/HttpClient.h>
+#include <aws/core/http/HttpClientFactory.h>
+#include <aws/core/http/HttpRequest.h>
+#include <aws/core/http/HttpResponse.h>
+#include <aws/core/utils/stream/ResponseStream.h>
+#include <gtest/gtest.h>
+
+#include <chrono>
+#include <condition_variable>
+#include <mutex>
+
+using namespace Aws;
+using namespace Aws::Client;
+using namespace Aws::Http;
+
+static const char ENDPOINT[] = "https://checkip.amazonaws.com";
+
+class AsyncHttpRequestIntegrationTest : public testing::Test
+{
+protected:
+    AsyncHttpRequestIntegrationTest()
+    {
+        InitAPI(m_options);
+    }
+
+    ~AsyncHttpRequestIntegrationTest()
+    {
+        ShutdownAPI(m_options);
+    }
+
+    SDKOptions m_options;
+};
+
+TEST_F(AsyncHttpRequestIntegrationTest, SynchronousGet)
+{
+    auto client = CreateHttpClient(ClientConfiguration());
+    auto request = CreateHttpRequest(Aws::String(ENDPOINT), HttpMethod::HTTP_GET,
+                                     Aws::Utils::Stream::DefaultResponseStreamFactoryMethod);
+
+    auto response = client->MakeRequest(request);
+
+    ASSERT_NE(nullptr, response);
+    ASSERT_FALSE(response->HasClientError()) << "sync GET returned a client error";
+    EXPECT_EQ(HttpResponseCode::OK, response->GetResponseCode());
+}
+
+#if AWS_SDK_USE_CRT_HTTP
+TEST_F(AsyncHttpRequestIntegrationTest, AsynchronousGet)
+{
+    std::mutex mutex;
+    std::condition_variable cv;
+    bool done = false;
+    std::shared_ptr<HttpResponse> asyncResponse;
+
+    auto client = CreateHttpClient(ClientConfiguration());
+    auto request = CreateHttpRequest(Aws::String(ENDPOINT), HttpMethod::HTTP_GET,
+                                     Aws::Utils::Stream::DefaultResponseStreamFactoryMethod);
+
+    auto error = client->MakeRequestAsync(request,
+        [&](std::shared_ptr<HttpResponse> response)
+        {
+            std::lock_guard<std::mutex> lock(mutex);
+            asyncResponse = std::move(response);
+            done = true;
+            cv.notify_one();
+        });
+
+    ASSERT_FALSE(error.has_value()) << error->GetMessage();
+
+    std::unique_lock<std::mutex> lock(mutex);
+    ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(30), [&] { return done; }))
+        << "async GET did not complete within 30 seconds";
+
+    ASSERT_NE(nullptr, asyncResponse);
+    ASSERT_FALSE(asyncResponse->HasClientError()) << "async GET returned a client error";
+    EXPECT_EQ(HttpResponseCode::OK, asyncResponse->GetResponseCode());
+}
+#endif
+
+#if AWS_SDK_USE_CRT_HTTP && !defined(_WIN32)
+
+#include <aws/core/client/CoreErrors.h>
+
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <atomic>
+#include <future>
+#include <string>
+#include <thread>
+
+TEST_F(AsyncHttpRequestIntegrationTest, RequestTimeoutClosesConnectionAndDrains)
+{
+    int listenFd = socket(AF_INET, SOCK_STREAM, 0);
+    ASSERT_GE(listenFd, 0);
+
+    sockaddr_in addr{};
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    addr.sin_port = 0;
+    ASSERT_EQ(0, bind(listenFd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)));
+    ASSERT_EQ(0, listen(listenFd, 1));
+
+    socklen_t addrLen = sizeof(addr);
+    ASSERT_EQ(0, getsockname(listenFd, reinterpret_cast<sockaddr*>(&addr), &addrLen));
+    unsigned short port = ntohs(addr.sin_port);
+
+    std::atomic<int> acceptedFd{-1};
+    std::thread accepter([listenFd, &acceptedFd]
+    {
+        acceptedFd.store(accept(listenFd, nullptr, nullptr));
+    });
+
+    ClientConfiguration config;
+    config.requestTimeoutMs = 500;
+    config.connectTimeoutMs = 5000;
+
+    std::string uri = "http://127.0.0.1:" + std::to_string(port) + "/";
+    auto client = CreateHttpClient(config);
+    auto request = CreateHttpRequest(Aws::String(uri.c_str()), HttpMethod::HTTP_GET,
+                                     Aws::Utils::Stream::DefaultResponseStreamFactoryMethod);
+
+    auto response = client->MakeRequest(request);
+
+    ASSERT_NE(nullptr, response);
+    ASSERT_TRUE(response->HasClientError()) << "expected the stalled request to time out";
+    EXPECT_EQ(CoreErrors::REQUEST_TIMEOUT, response->GetClientErrorType());
+
+    auto drained = std::async(std::launch::async, [&] { client.reset(); });
+    ASSERT_EQ(std::future_status::ready, drained.wait_for(std::chrono::seconds(10)))
+        << "~CRTHttpClient hung draining a timed-out request";
+
+    accepter.join();
+    int fd = acceptedFd.load();
+    if (fd >= 0)
+    {
+        close(fd);
+    }
+    close(listenFd);
+}
+#endif
diff --git a/tests/aws-cpp-sdk-core-integration-tests/CMakeLists.txt b/tests/aws-cpp-sdk-core-integration-tests/CMakeLists.txt
index e00d67793bb0..52ca8b2cf18f 100644
--- a/tests/aws-cpp-sdk-core-integration-tests/CMakeLists.txt
+++ b/tests/aws-cpp-sdk-core-integration-tests/CMakeLists.txt
@@ -16,9 +16,9 @@ endif()
 enable_testing()
 
 if(PLATFORM_ANDROID AND BUILD_SHARED_LIBS)
-    add_library(${PROJECT_NAME} ${CMAKE_CURRENT_SOURCE_DIR}/STSWebIdentityProviderIntegrationTest.cpp ${CMAKE_CURRENT_SOURCE_DIR}/DefaultCredentialsProviderChainIntegrationTest.cpp)
+    add_library(${PROJECT_NAME} ${CMAKE_CURRENT_SOURCE_DIR}/STSWebIdentityProviderIntegrationTest.cpp ${CMAKE_CURRENT_SOURCE_DIR}/DefaultCredentialsProviderChainIntegrationTest.cpp ${CMAKE_CURRENT_SOURCE_DIR}/AsyncHttpRequestIntegrationTest.cpp)
 else()
-    add_executable(${PROJECT_NAME} ${CMAKE_CURRENT_SOURCE_DIR}/STSWebIdentityProviderIntegrationTest.cpp ${CMAKE_CURRENT_SOURCE_DIR}/DefaultCredentialsProviderChainIntegrationTest.cpp)
+    add_executable(${PROJECT_NAME} ${CMAKE_CURRENT_SOURCE_DIR}/STSWebIdentityProviderIntegrationTest.cpp ${CMAKE_CURRENT_SOURCE_DIR}/DefaultCredentialsProviderChainIntegrationTest.cpp ${CMAKE_CURRENT_SOURCE_DIR}/AsyncHttpRequestIntegrationTest.cpp)
 endif()
 
 set_compiler_flags(${PROJECT_NAME})
diff --git a/tests/aws-cpp-sdk-core-tests/utils/threading/WaitGroupTest.cpp b/tests/aws-cpp-sdk-core-tests/utils/threading/WaitGroupTest.cpp
new file mode 100644
index 000000000000..379911bfe6b2
--- /dev/null
+++ b/tests/aws-cpp-sdk-core-tests/utils/threading/WaitGroupTest.cpp
@@ -0,0 +1,83 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#include <aws/testing/AwsCppSdkGTestSuite.h>
+#include <aws/core/utils/threading/WaitGroup.h>
+
+#include <atomic>
+#include <chrono>
+#include <future>
+#include <thread>
+#include <vector>
+
+using namespace Aws::Utils::Threading;
+
+class WaitGroupTest : public Aws::Testing::AwsCppSdkGTestSuite
+{
+};
+
+TEST_F(WaitGroupTest, WaitReturnsImmediatelyWhenEmpty)
+{
+    WaitGroup waitGroup;
+    auto waiter = std::async(std::launch::async, [&]() { waitGroup.Wait(); });
+    EXPECT_EQ(std::future_status::ready, waiter.wait_for(std::chrono::seconds(5)))
+        << "Wait() on an empty WaitGroup should not block";
+}
+
+TEST_F(WaitGroupTest, WaitBlocksUntilDone)
+{
+    WaitGroup waitGroup;
+    waitGroup.Add(1);
+
+    std::atomic<bool> released{false};
+    std::thread worker([&]()
+    {
+        released = true;
+        waitGroup.Done();
+    });
+
+    auto waiter = std::async(std::launch::async, [&]() { waitGroup.Wait(); });
+    ASSERT_EQ(std::future_status::ready, waiter.wait_for(std::chrono::seconds(5)))
+        << "Wait() did not return after the outstanding entry was Done()";
+    EXPECT_TRUE(released.load());
+
+    worker.join();
+}
+
+TEST_F(WaitGroupTest, WaitDrainsMultipleOutstanding)
+{
+    WaitGroup waitGroup;
+    const size_t entries = 8;
+    waitGroup.Add(entries);
+
+    std::vector<std::thread> workers;
+    for (size_t i = 0; i < entries; ++i)
+    {
+        workers.emplace_back([&]() { waitGroup.Done(); });
+    }
+
+    auto waiter = std::async(std::launch::async, [&]() { waitGroup.Wait(); });
+    ASSERT_EQ(std::future_status::ready, waiter.wait_for(std::chrono::seconds(5)))
+        << "Wait() did not drain all outstanding entries";
+
+    for (auto& worker : workers)
+    {
+        worker.join();
+    }
+}
+
+TEST_F(WaitGroupTest, IsReusableAcrossRounds)
+{
+    WaitGroup waitGroup;
+    for (int round = 0; round < 3; ++round)
+    {
+        waitGroup.Add(1);
+        std::thread worker([&]() { waitGroup.Done(); });
+        auto waiter = std::async(std::launch::async, [&]() { waitGroup.Wait(); });
+        ASSERT_EQ(std::future_status::ready, waiter.wait_for(std::chrono::seconds(5)))
+            << "Wait() did not return on round " << round;
+        worker.join();
+    }
+}
diff --git a/tests/testing-resources/include/aws/testing/mocks/http/MockConnection.h b/tests/testing-resources/include/aws/testing/mocks/http/MockConnection.h
index b16fba0ffa54..87e08c7fb805 100644
--- a/tests/testing-resources/include/aws/testing/mocks/http/MockConnection.h
+++ b/tests/testing-resources/include/aws/testing/mocks/http/MockConnection.h
@@ -55,6 +55,8 @@ class MockConnection : public Aws::Http::Connection {
     return Aws::MakeShared<MockClientStream>("MockConnection", m_testCase, std::move(onStreamComplete));
   }
 
+  void Close() override {}
+
  private:
   ConnectionTestCase m_testCase;
 };
\ No newline at end of file