diff --git a/api/envoy/api/v2/auth/cert.proto b/api/envoy/api/v2/auth/cert.proto
index 6f0eb533fc247..cd1df6b3c3470 100644
--- a/api/envoy/api/v2/auth/cert.proto
+++ b/api/envoy/api/v2/auth/cert.proto
@@ -293,6 +293,12 @@ message UpstreamTlsContext {
   //
   //   TLS renegotiation is considered insecure and shouldn't be used unless absolutely necessary.
   bool allow_renegotiation = 3;
+
+  // Maximum number of session keys (Pre-Shared Keys for TLSv1.3+, Session IDs and Session Tickets
+  // for TLSv1.2 and older) to store for the purpose of session resumption.
+  //
+  // Defaults to 1, setting this to 0 disables session resumption.
+  google.protobuf.UInt32Value max_session_keys = 4;
 }
 
 message DownstreamTlsContext {
diff --git a/include/envoy/ssl/context_config.h b/include/envoy/ssl/context_config.h
index 1f475d5fab33b..a7afb1aedf87e 100644
--- a/include/envoy/ssl/context_config.h
+++ b/include/envoy/ssl/context_config.h
@@ -81,6 +81,11 @@ class ClientContextConfig : public virtual ContextConfig {
    * @return true if server-initiated TLS renegotiation will be allowed.
    */
   virtual bool allowRenegotiation() const PURE;
+
+  /**
+   * @return The maximum number of session keys to store.
+   */
+  virtual size_t maxSessionKeys() const PURE;
 };
 
 typedef std::unique_ptr<ClientContextConfig> ClientContextConfigPtr;
diff --git a/source/common/ssl/BUILD b/source/common/ssl/BUILD
index 8bbadc9332ef7..11fb5c9b61907 100644
--- a/source/common/ssl/BUILD
+++ b/source/common/ssl/BUILD
@@ -66,7 +66,10 @@ envoy_cc_library(
         "context_impl.h",
         "context_manager_impl.h",
     ],
-    external_deps = ["ssl"],
+    external_deps = [
+        "abseil_synchronization",
+        "ssl",
+    ],
     deps = [
         ":utility_lib",
         "//include/envoy/ssl:context_config_interface",
diff --git a/source/common/ssl/context_config_impl.cc b/source/common/ssl/context_config_impl.cc
index a3aec386e4f2e..eeb3acab8de99 100644
--- a/source/common/ssl/context_config_impl.cc
+++ b/source/common/ssl/context_config_impl.cc
@@ -247,7 +247,8 @@ ClientContextConfigImpl::ClientContextConfigImpl(
     const envoy::api::v2::auth::UpstreamTlsContext& config,
     Server::Configuration::TransportSocketFactoryContext& factory_context)
     : ContextConfigImpl(config.common_tls_context(), factory_context),
-      server_name_indication_(config.sni()), allow_renegotiation_(config.allow_renegotiation()) {
+      server_name_indication_(config.sni()), allow_renegotiation_(config.allow_renegotiation()),
+      max_session_keys_(PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, max_session_keys, 1)) {
   // BoringSSL treats this as a C string, so embedded NULL characters will not
   // be handled correctly.
   if (server_name_indication_.find('\0') != std::string::npos) {
diff --git a/source/common/ssl/context_config_impl.h b/source/common/ssl/context_config_impl.h
index 7a853c69e539e..2914e91410f76 100644
--- a/source/common/ssl/context_config_impl.h
+++ b/source/common/ssl/context_config_impl.h
@@ -95,10 +95,12 @@ class ClientContextConfigImpl : public ContextConfigImpl, public ClientContextCo
   // Ssl::ClientContextConfig
   const std::string& serverNameIndication() const override { return server_name_indication_; }
   bool allowRenegotiation() const override { return allow_renegotiation_; }
+  size_t maxSessionKeys() const override { return max_session_keys_; }
 
 private:
   const std::string server_name_indication_;
   const bool allow_renegotiation_;
+  const size_t max_session_keys_;
 };
 
 class ServerContextConfigImpl : public ContextConfigImpl, public ServerContextConfig {
diff --git a/source/common/ssl/context_impl.cc b/source/common/ssl/context_impl.cc
index 2fd2eee4ea1a0..6fd8e3230320b 100644
--- a/source/common/ssl/context_impl.cc
+++ b/source/common/ssl/context_impl.cc
@@ -275,7 +275,7 @@ std::vector<uint8_t> ContextImpl::parseAlpnProtocols(const std::string& alpn_pro
   return out;
 }
 
-bssl::UniquePtr<SSL> ContextImpl::newSsl(absl::optional<std::string>) const {
+bssl::UniquePtr<SSL> ContextImpl::newSsl(absl::optional<std::string>) {
   return bssl::UniquePtr<SSL>(SSL_new(ctx_.get()));
 }
 
@@ -490,16 +490,27 @@ ClientContextImpl::ClientContextImpl(Stats::Scope& scope, const ClientContextCon
                                      TimeSource& time_source)
     : ContextImpl(scope, config, time_source),
       server_name_indication_(config.serverNameIndication()),
-      allow_renegotiation_(config.allowRenegotiation()) {
+      allow_renegotiation_(config.allowRenegotiation()),
+      max_session_keys_(config.maxSessionKeys()) {
   if (!parsed_alpn_protocols_.empty()) {
     int rc = SSL_CTX_set_alpn_protos(ctx_.get(), &parsed_alpn_protocols_[0],
                                      parsed_alpn_protocols_.size());
     RELEASE_ASSERT(rc == 0, "");
   }
+
+  if (max_session_keys_ > 0) {
+    SSL_CTX_set_session_cache_mode(ctx_.get(), SSL_SESS_CACHE_CLIENT);
+    SSL_CTX_sess_set_new_cb(ctx_.get(), [](SSL* ssl, SSL_SESSION* session) -> int {
+      ContextImpl* context_impl =
+          static_cast<ContextImpl*>(SSL_CTX_get_ex_data(SSL_get_SSL_CTX(ssl), sslContextIndex()));
+      ClientContextImpl* client_context_impl = dynamic_cast<ClientContextImpl*>(context_impl);
+      RELEASE_ASSERT(client_context_impl != nullptr, ""); // for Coverity
+      return client_context_impl->newSessionKey(session);
+    });
+  }
 }
 
-bssl::UniquePtr<SSL>
-ClientContextImpl::newSsl(absl::optional<std::string> override_server_name) const {
+bssl::UniquePtr<SSL> ClientContextImpl::newSsl(absl::optional<std::string> override_server_name) {
   bssl::UniquePtr<SSL> ssl_con(ContextImpl::newSsl(absl::nullopt));
 
   std::string server_name_indication =
@@ -514,9 +525,51 @@ ClientContextImpl::newSsl(absl::optional<std::string> override_server_name) cons
     SSL_set_renegotiate_mode(ssl_con.get(), ssl_renegotiate_freely);
   }
 
+  if (max_session_keys_ > 0) {
+    if (session_keys_single_use_) {
+      // Stored single-use session keys, use write/write locks.
+      absl::WriterMutexLock l(&session_keys_mu_);
+      if (!session_keys_.empty()) {
+        // Use the most recently stored session key, since it has the highest
+        // probability of still being recognized/accepted by the server.
+        SSL_SESSION* session = session_keys_.front().get();
+        SSL_set_session(ssl_con.get(), session);
+        // Remove single-use session key (TLS 1.3) after first use.
+        if (SSL_SESSION_should_be_single_use(session)) {
+          session_keys_.pop_front();
+        }
+      }
+    } else {
+      // Never stored single-use session keys, use read/write locks.
+      absl::ReaderMutexLock l(&session_keys_mu_);
+      if (!session_keys_.empty()) {
+        // Use the most recently stored session key, since it has the highest
+        // probability of still being recognized/accepted by the server.
+        SSL_SESSION* session = session_keys_.front().get();
+        SSL_set_session(ssl_con.get(), session);
+      }
+    }
+  }
+
   return ssl_con;
 }
 
+int ClientContextImpl::newSessionKey(SSL_SESSION* session) {
+  // In case we ever store single-use session key (TLS 1.3),
+  // we need to switch to using write/write locks.
+  if (SSL_SESSION_should_be_single_use(session)) {
+    session_keys_single_use_ = true;
+  }
+  absl::WriterMutexLock l(&session_keys_mu_);
+  // Evict oldest entries.
+  while (session_keys_.size() >= max_session_keys_) {
+    session_keys_.pop_back();
+  }
+  // Add new session key at the front of the queue, so that it's used first.
+  session_keys_.push_front(bssl::UniquePtr<SSL_SESSION>(session));
+  return 1; // Tell BoringSSL that we took ownership of the session.
+}
+
 ServerContextImpl::ServerContextImpl(Stats::Scope& scope, const ServerContextConfig& config,
                                      const std::vector<std::string>& server_names,
                                      TimeSource& time_source)
diff --git a/source/common/ssl/context_impl.h b/source/common/ssl/context_impl.h
index 4fb733df025be..e52aaa0e7219c 100644
--- a/source/common/ssl/context_impl.h
+++ b/source/common/ssl/context_impl.h
@@ -1,5 +1,6 @@
 #pragma once
 
+#include <deque>
 #include <functional>
 #include <string>
 #include <vector>
@@ -11,6 +12,7 @@
 
 #include "common/ssl/context_manager_impl.h"
 
+#include "absl/synchronization/mutex.h"
 #include "absl/types/optional.h"
 #include "openssl/ssl.h"
 
@@ -42,7 +44,7 @@ struct SslStats {
 
 class ContextImpl : public virtual Context {
 public:
-  virtual bssl::UniquePtr<SSL> newSsl(absl::optional<std::string> override_server_name) const;
+  virtual bssl::UniquePtr<SSL> newSsl(absl::optional<std::string> override_server_name);
 
   /**
    * Logs successful TLS handshake and updates stats.
@@ -143,11 +145,17 @@ class ClientContextImpl : public ContextImpl, public ClientContext {
   ClientContextImpl(Stats::Scope& scope, const ClientContextConfig& config,
                     TimeSource& time_source);
 
-  bssl::UniquePtr<SSL> newSsl(absl::optional<std::string> override_server_name) const override;
+  bssl::UniquePtr<SSL> newSsl(absl::optional<std::string> override_server_name) override;
 
 private:
+  int newSessionKey(SSL_SESSION* session);
+
   const std::string server_name_indication_;
   const bool allow_renegotiation_;
+  const size_t max_session_keys_;
+  absl::Mutex session_keys_mu_;
+  std::deque<bssl::UniquePtr<SSL_SESSION>> session_keys_ GUARDED_BY(session_keys_mu_);
+  bool session_keys_single_use_{false};
 };
 
 class ServerContextImpl : public ContextImpl, public ServerContext {
diff --git a/test/common/ssl/ssl_socket_test.cc b/test/common/ssl/ssl_socket_test.cc
index cb2f43c761a0d..257baec1d025e 100644
--- a/test/common/ssl/ssl_socket_test.cc
+++ b/test/common/ssl/ssl_socket_test.cc
@@ -35,6 +35,7 @@
 
 using testing::_;
 using testing::DoAll;
+using testing::InSequence;
 using testing::Invoke;
 using testing::NiceMock;
 using testing::Return;
@@ -100,9 +101,9 @@ void testUtil(const std::string& client_ctx_yaml, const std::string& server_ctx_
   client_connection->addConnectionCallbacks(client_connection_callbacks);
   client_connection->connect();
 
-  size_t counter = 0;
-  auto stopSecondTime = [&]() {
-    if (++counter == 2) {
+  size_t connect_count = 0;
+  auto connect_second_time = [&]() {
+    if (++connect_count == 2) {
       if (!expected_digest.empty()) {
         // Assert twice to ensure a cached value is returned and still valid.
         EXPECT_EQ(expected_digest, server_connection->ssl()->sha256PeerCertificateDigest());
@@ -131,24 +132,26 @@ void testUtil(const std::string& client_ctx_yaml, const std::string& server_ctx_
       dispatcher.exit();
     }
   };
-  auto closeSecondTime = [&]() {
-    if (++counter == 2) {
+
+  size_t close_count = 0;
+  auto close_second_time = [&close_count, &dispatcher]() {
+    if (++close_count == 2) {
       dispatcher.exit();
     }
   };
 
   if (expect_success) {
     EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
-        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { stopSecondTime(); }));
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { connect_second_time(); }));
     EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
-        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { stopSecondTime(); }));
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { connect_second_time(); }));
     EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::LocalClose));
     EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::LocalClose));
   } else {
     EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::RemoteClose))
-        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { closeSecondTime(); }));
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { close_second_time(); }));
     EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::RemoteClose))
-        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { closeSecondTime(); }));
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { close_second_time(); }));
   }
 
   dispatcher.run(Event::Dispatcher::RunType::Block);
@@ -236,9 +239,9 @@ const std::string testUtilV2(
   client_connection->addConnectionCallbacks(client_connection_callbacks);
   client_connection->connect();
 
-  size_t counter = 0;
-  auto stopSecondTime = [&]() {
-    if (++counter == 2) {
+  size_t connect_count = 0;
+  auto connect_second_time = [&]() {
+    if (++connect_count == 2) {
       if (!expected_server_cert_digest.empty()) {
         EXPECT_EQ(expected_server_cert_digest,
                   client_connection->ssl()->sha256PeerCertificateDigest());
@@ -283,27 +286,29 @@ const std::string testUtilV2(
       dispatcher.exit();
     }
   };
-  auto closeSecondTime = [&]() {
-    if (++counter == 2) {
+
+  size_t close_count = 0;
+  auto close_second_time = [&close_count, &dispatcher]() {
+    if (++close_count == 2) {
       dispatcher.exit();
     }
   };
 
   if (expect_success) {
     EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
-        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { stopSecondTime(); }));
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { connect_second_time(); }));
     EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
         .WillOnce(Invoke([&](Network::ConnectionEvent) -> void {
           EXPECT_EQ(expected_requested_server_name, server_connection->requestedServerName());
-          stopSecondTime();
+          connect_second_time();
         }));
     EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::LocalClose));
     EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::LocalClose));
   } else {
     EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::RemoteClose))
-        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { closeSecondTime(); }));
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { close_second_time(); }));
     EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::RemoteClose))
-        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { closeSecondTime(); }));
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { close_second_time(); }));
   }
 
   dispatcher.run(Event::Dispatcher::RunType::Block);
@@ -2037,8 +2042,9 @@ void testTicketSessionResumption(const std::string& server_ctx_yaml1,
 
   // Different tests have different order of whether client or server gets Connected event
   // first, so always wait until both have happened.
-  unsigned connect_count = 0;
-  auto stopSecondTime = [&]() {
+  size_t connect_count = 0;
+  auto connect_second_time = [&connect_count, &dispatcher, &server_connection,
+                              &client_connection]() {
     connect_count++;
     if (connect_count == 2) {
       client_connection->close(Network::ConnectionCloseType::NoFlush);
@@ -2048,9 +2054,9 @@ void testTicketSessionResumption(const std::string& server_ctx_yaml1,
   };
 
   EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
-      .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { stopSecondTime(); }));
+      .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { connect_second_time(); }));
   EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
-      .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { stopSecondTime(); }));
+      .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { connect_second_time(); }));
   EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::LocalClose));
   EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::LocalClose));
 
@@ -2461,6 +2467,259 @@ TEST_P(SslSocketTest, ClientAuthCrossListenerSessionResumption) {
   EXPECT_EQ(0UL, client_stats_store.counter("ssl.session_reused").value());
 }
 
+void testClientSessionResumption(const std::string& server_ctx_yaml,
+                                 const std::string& client_ctx_yaml, bool expect_reuse,
+                                 const Network::Address::IpVersion version) {
+  InSequence s;
+
+  testing::NiceMock<Server::Configuration::MockTransportSocketFactoryContext> factory_context;
+  Event::SimulatedTimeSystem time_system;
+  ContextManagerImpl manager(time_system);
+
+  envoy::api::v2::auth::DownstreamTlsContext server_ctx_proto;
+  MessageUtil::loadFromYaml(TestEnvironment::substitute(server_ctx_yaml), server_ctx_proto);
+  auto server_cfg = std::make_unique<ServerContextConfigImpl>(server_ctx_proto, factory_context);
+  Stats::IsolatedStoreImpl server_stats_store;
+  ServerSslSocketFactory server_ssl_socket_factory(std::move(server_cfg), manager,
+                                                   server_stats_store, std::vector<std::string>{});
+
+  Network::TcpListenSocket socket(Network::Test::getCanonicalLoopbackAddress(version), nullptr,
+                                  true);
+  NiceMock<Network::MockListenerCallbacks> callbacks;
+  Network::MockConnectionHandler connection_handler;
+  Event::DispatcherImpl dispatcher(time_system);
+  Network::ListenerPtr listener = dispatcher.createListener(socket, callbacks, true, false);
+
+  Network::ConnectionPtr server_connection;
+  Network::MockConnectionCallbacks server_connection_callbacks;
+
+  envoy::api::v2::auth::UpstreamTlsContext client_ctx_proto;
+  MessageUtil::loadFromYaml(TestEnvironment::substitute(client_ctx_yaml), client_ctx_proto);
+  auto client_cfg = std::make_unique<ClientContextConfigImpl>(client_ctx_proto, factory_context);
+  Stats::IsolatedStoreImpl client_stats_store;
+  ClientSslSocketFactory client_ssl_socket_factory(std::move(client_cfg), manager,
+                                                   client_stats_store);
+  Network::ClientConnectionPtr client_connection = dispatcher.createClientConnection(
+      socket.localAddress(), Network::Address::InstanceConstSharedPtr(),
+      client_ssl_socket_factory.createTransportSocket(nullptr), nullptr);
+
+  Network::MockConnectionCallbacks client_connection_callbacks;
+  client_connection->addConnectionCallbacks(client_connection_callbacks);
+  client_connection->connect();
+
+  size_t connect_count = 0;
+  auto connect_second_time = [&connect_count, &server_connection]() {
+    if (++connect_count == 2) {
+      server_connection->close(Network::ConnectionCloseType::NoFlush);
+    }
+  };
+
+  size_t close_count = 0;
+  auto close_second_time = [&close_count, &dispatcher]() {
+    if (++close_count == 2) {
+      dispatcher.exit();
+    }
+  };
+
+  // WillRepeatedly doesn't work with InSequence.
+  EXPECT_CALL(callbacks, onAccept_(_, _))
+      .WillOnce(Invoke([&](Network::ConnectionSocketPtr& socket, bool) -> void {
+        Network::ConnectionPtr new_connection = dispatcher.createServerConnection(
+            std::move(socket), server_ssl_socket_factory.createTransportSocket(nullptr));
+        callbacks.onNewConnection(std::move(new_connection));
+      }));
+  EXPECT_CALL(callbacks, onNewConnection_(_))
+      .WillOnce(Invoke([&](Network::ConnectionPtr& conn) -> void {
+        server_connection = std::move(conn);
+        server_connection->addConnectionCallbacks(server_connection_callbacks);
+      }));
+
+  const bool expect_tls13 =
+      client_ctx_proto.common_tls_context().tls_params().tls_maximum_protocol_version() ==
+          envoy::api::v2::auth::TlsParameters::TLSv1_3 &&
+      server_ctx_proto.common_tls_context().tls_params().tls_maximum_protocol_version() ==
+          envoy::api::v2::auth::TlsParameters::TLSv1_3;
+
+  // The order of "Connected" events depends on the version of the TLS protocol (1.3 or older).
+  if (expect_tls13) {
+    EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { connect_second_time(); }));
+    EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { connect_second_time(); }));
+  } else {
+    EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { connect_second_time(); }));
+    EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { connect_second_time(); }));
+  }
+  EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::LocalClose))
+      .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { close_second_time(); }));
+  EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::RemoteClose))
+      .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { close_second_time(); }));
+
+  dispatcher.run(Event::Dispatcher::RunType::Block);
+
+  EXPECT_EQ(0UL, server_stats_store.counter("ssl.session_reused").value());
+  EXPECT_EQ(0UL, client_stats_store.counter("ssl.session_reused").value());
+
+  connect_count = 0;
+  close_count = 0;
+
+  client_connection = dispatcher.createClientConnection(
+      socket.localAddress(), Network::Address::InstanceConstSharedPtr(),
+      client_ssl_socket_factory.createTransportSocket(nullptr), nullptr);
+  client_connection->addConnectionCallbacks(client_connection_callbacks);
+  client_connection->connect();
+
+  // WillRepeatedly doesn't work with InSequence.
+  EXPECT_CALL(callbacks, onAccept_(_, _))
+      .WillOnce(Invoke([&](Network::ConnectionSocketPtr& socket, bool) -> void {
+        Network::ConnectionPtr new_connection = dispatcher.createServerConnection(
+            std::move(socket), server_ssl_socket_factory.createTransportSocket(nullptr));
+        callbacks.onNewConnection(std::move(new_connection));
+      }));
+  EXPECT_CALL(callbacks, onNewConnection_(_))
+      .WillOnce(Invoke([&](Network::ConnectionPtr& conn) -> void {
+        server_connection = std::move(conn);
+        server_connection->addConnectionCallbacks(server_connection_callbacks);
+      }));
+
+  // The order of "Connected" events depends on the version of the TLS protocol (1.3 or older),
+  // and whether or not the session was successfully resumed.
+  if (expect_tls13 || expect_reuse) {
+    EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { connect_second_time(); }));
+    EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { connect_second_time(); }));
+  } else {
+    EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { connect_second_time(); }));
+    EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::Connected))
+        .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { connect_second_time(); }));
+  }
+  EXPECT_CALL(server_connection_callbacks, onEvent(Network::ConnectionEvent::LocalClose))
+      .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { close_second_time(); }));
+  EXPECT_CALL(client_connection_callbacks, onEvent(Network::ConnectionEvent::RemoteClose))
+      .WillOnce(Invoke([&](Network::ConnectionEvent) -> void { close_second_time(); }));
+
+  dispatcher.run(Event::Dispatcher::RunType::Block);
+
+  EXPECT_EQ(expect_reuse ? 1UL : 0UL, server_stats_store.counter("ssl.session_reused").value());
+  EXPECT_EQ(expect_reuse ? 1UL : 0UL, client_stats_store.counter("ssl.session_reused").value());
+}
+
+// Test client session resumption using default settings (should be enabled).
+TEST_P(SslSocketTest, ClientSessionResumptionDefault) {
+  const std::string server_ctx_yaml = R"EOF(
+  common_tls_context:
+    tls_certificates:
+      certificate_chain:
+        filename: "{{ test_tmpdir }}/unittestcert.pem"
+      private_key:
+        filename: "{{ test_tmpdir }}/unittestkey.pem"
+)EOF";
+
+  const std::string client_ctx_yaml = R"EOF(
+  common_tls_context:
+)EOF";
+
+  testClientSessionResumption(server_ctx_yaml, client_ctx_yaml, true, GetParam());
+}
+
+// Make sure client session resumption is not happening with TLS 1.0-1.2 when it's disabled.
+TEST_P(SslSocketTest, ClientSessionResumptionDisabledTls12) {
+  const std::string server_ctx_yaml = R"EOF(
+  common_tls_context:
+    tls_certificates:
+      certificate_chain:
+        filename: "{{ test_tmpdir }}/unittestcert.pem"
+      private_key:
+        filename: "{{ test_tmpdir }}/unittestkey.pem"
+)EOF";
+
+  const std::string client_ctx_yaml = R"EOF(
+  common_tls_context:
+  max_session_keys: 0
+)EOF";
+
+  testClientSessionResumption(server_ctx_yaml, client_ctx_yaml, false, GetParam());
+}
+
+// Test client session resumption with TLS 1.0-1.2.
+TEST_P(SslSocketTest, ClientSessionResumptionEnabledTls12) {
+  const std::string server_ctx_yaml = R"EOF(
+  common_tls_context:
+    tls_params:
+      tls_minimum_protocol_version: TLSv1_0
+      tls_maximum_protocol_version: TLSv1_2
+    tls_certificates:
+      certificate_chain:
+        filename: "{{ test_tmpdir }}/unittestcert.pem"
+      private_key:
+        filename: "{{ test_tmpdir }}/unittestkey.pem"
+)EOF";
+
+  const std::string client_ctx_yaml = R"EOF(
+  common_tls_context:
+    tls_params:
+      tls_minimum_protocol_version: TLSv1_0
+      tls_maximum_protocol_version: TLSv1_2
+  max_session_keys: 2
+)EOF";
+
+  testClientSessionResumption(server_ctx_yaml, client_ctx_yaml, true, GetParam());
+}
+
+// Make sure client session resumption is not happening with TLS 1.3 when it's disabled.
+TEST_P(SslSocketTest, ClientSessionResumptionDisabledTls13) {
+  const std::string server_ctx_yaml = R"EOF(
+  common_tls_context:
+    tls_params:
+      tls_minimum_protocol_version: TLSv1_3
+      tls_maximum_protocol_version: TLSv1_3
+    tls_certificates:
+      certificate_chain:
+        filename: "{{ test_tmpdir }}/unittestcert.pem"
+      private_key:
+        filename: "{{ test_tmpdir }}/unittestkey.pem"
+)EOF";
+
+  const std::string client_ctx_yaml = R"EOF(
+  common_tls_context:
+    tls_params:
+      tls_minimum_protocol_version: TLSv1_3
+      tls_maximum_protocol_version: TLSv1_3
+  max_session_keys: 0
+)EOF";
+
+  testClientSessionResumption(server_ctx_yaml, client_ctx_yaml, false, GetParam());
+}
+
+// Test client session resumption with TLS 1.3 (it's different than in older versions of TLS).
+TEST_P(SslSocketTest, ClientSessionResumptionEnabledTls13) {
+  const std::string server_ctx_yaml = R"EOF(
+  common_tls_context:
+    tls_params:
+      tls_minimum_protocol_version: TLSv1_3
+      tls_maximum_protocol_version: TLSv1_3
+    tls_certificates:
+      certificate_chain:
+        filename: "{{ test_tmpdir }}/unittestcert.pem"
+      private_key:
+        filename: "{{ test_tmpdir }}/unittestkey.pem"
+)EOF";
+
+  const std::string client_ctx_yaml = R"EOF(
+  common_tls_context:
+    tls_params:
+      tls_minimum_protocol_version: TLSv1_3
+      tls_maximum_protocol_version: TLSv1_3
+  max_session_keys: 2
+)EOF";
+
+  testClientSessionResumption(server_ctx_yaml, client_ctx_yaml, true, GetParam());
+}
+
 TEST_P(SslSocketTest, SslError) {
   const std::string server_ctx_yaml = R"EOF(
   common_tls_context: