Merge pull request #11584 from ztlpn/fix-9243

rpc/conn_cache: return errc::shutting_down after shutdown

src/v/cluster/cluster_utils.cc

@@ -355,6 +355,8 @@ cluster::errc map_update_interruption_error_code(std::error_code ec) {
         case rpc::errc::client_request_timeout:
         case rpc::errc::connection_timeout:
             return errc::timeout;
+        case rpc::errc::shutting_down:
+            return errc::shutting_down;
         case rpc::errc::disconnected_endpoint:
         case rpc::errc::exponential_backoff:
         case rpc::errc::missing_node_rpc_client:

src/v/redpanda/admin_server.cc

@@ -1009,8 +1009,8 @@ ss::future<> admin_server::throw_on_error(
         case rpc::errc::success:
             co_return;
         case rpc::errc::disconnected_endpoint:
-            [[fallthrough]];
         case rpc::errc::exponential_backoff:
+        case rpc::errc::shutting_down:
             throw ss::httpd::base_exception(
               fmt::format("Not ready: {}", ec.message()),
               ss::http::reply::status_type::service_unavailable);

src/v/rpc/connection_cache.cc

@@ -76,6 +76,7 @@ ss::future<> connection_cache::remove_all() {
 /// \brief closes all client connections
 ss::future<> connection_cache::do_shutdown() {
     auto units = co_await _mutex.get_units();
+    _shutting_down = true;
     // Exchange ensures the cache is invalidated and concurrent
     // accesses wait on the mutex to populate new entries.
     auto cache = std::exchange(_cache, {});

src/v/rpc/connection_cache.h

@@ -92,6 +92,10 @@ class connection_cache final
           shard,
           [node_id, f = std::forward<Func>(f), connection_timeout](
             rpc::connection_cache& cache) mutable {
+              if (cache._shutting_down) {
+                  return ss::futurize<ret_t>::convert(
+                    rpc::make_error_code(errc::shutting_down));
+              }
               if (!cache.contains(node_id)) {
                   // No client available
                   return ss::futurize<ret_t>::convert(
@@ -159,6 +163,7 @@ class connection_cache final
     transport_version _default_transport_version{transport_version::v2};
     ss::gate _gate;
     ss::optimized_optional<ss::abort_source::subscription> _as_subscription;
+    bool _shutting_down = false;
 };
 inline ss::shard_id connection_cache::shard_for(
   model::node_id self,

src/v/rpc/errc.h

@@ -25,6 +25,7 @@ enum class errc {
     method_not_found,
     version_not_supported,
     connection_timeout,
+    shutting_down,
 
     // Used when receiving an undefined errc (e.g. from a newer version of
     // Redpanda).
@@ -49,6 +50,8 @@ struct errc_category final : public std::error_category {
             return "rpc::errc::version_not_supported";
         case errc::connection_timeout:
             return "rpc::errc::connection_timeout";
+        case errc::shutting_down:
+            return "rpc::errc::shutting_down";
         default:
             return "rpc::errc::unknown(" + std::to_string(c) + ")";
         }