Configurable TCP server connection backlog

src/v/config/configuration.cc

@@ -61,6 +61,12 @@ configuration::configuration()
       required::no,
       tls_config(),
       tls_config::validate)
+  , rpc_server_listen_backlog(
+      *this,
+      "rpc_server_listen_backlog",
+      "TCP connection queue length for Kafka server and internal RPC server",
+      required::no,
+      std::nullopt)
   , enable_coproc(
       *this, "enable_coproc", "Enable coprocessing mode", required::no, false)
   , coproc_supervisor_server(

src/v/config/configuration.h

@@ -48,6 +48,7 @@ struct configuration final : public config_store {
     // Network
     property<unresolved_address> rpc_server;
     property<tls_config> rpc_server_tls;
+    property<std::optional<int>> rpc_server_listen_backlog;
     // Coproc
     property<bool> enable_coproc;
     property<unresolved_address> coproc_supervisor_server;

src/v/redpanda/application.cc

@@ -647,6 +647,8 @@ void application::wire_up_redpanda_services() {
               c.max_service_memory_per_core = memory_groups::rpc_total_memory();
               c.disable_metrics = rpc::metrics_disabled(
                 config::shard_local_cfg().disable_metrics());
+              c.listen_backlog
+                = config::shard_local_cfg().rpc_server_listen_backlog;
               auto rpc_builder = config::shard_local_cfg()
                                    .rpc_server_tls()
                                    .get_credentials_builder()
@@ -741,6 +743,8 @@ void application::wire_up_redpanda_services() {
           return ss::async([this, &c] {
               c.max_service_memory_per_core
                 = memory_groups::kafka_total_memory();
+              c.listen_backlog
+                = config::shard_local_cfg().rpc_server_listen_backlog;
               auto& tls_config
                 = config::shard_local_cfg().kafka_api_tls.value();
               for (const auto& ep : config::shard_local_cfg().kafka_api()) {

src/v/rpc/server.cc

@@ -46,6 +46,9 @@ void server::start() {
             ss::listen_options lo;
             lo.reuse_address = true;
             lo.lba = cfg.load_balancing_algo;
+            if (cfg.listen_backlog.has_value()) {
+                lo.listen_backlog = cfg.listen_backlog.value();
+            }
 
             if (!endpoint.credentials) {
                 ss = ss::engine().listen(endpoint.addr, lo);

src/v/rpc/types.h

@@ -207,6 +207,7 @@ struct server_endpoint {
 struct server_configuration {
     std::vector<server_endpoint> addrs;
     int64_t max_service_memory_per_core;
+    std::optional<int> listen_backlog;
     metrics_disabled disable_metrics = metrics_disabled::no;
     ss::sstring name;
     // we use the same default as seastar for load balancing algorithm