Add setting for keep-alive duration for oidc back-channel

In some environment, the back-channel connection can be dropped
without sending a TCP RST to ES. When that happens, reusing the same
connection results into timeout error.

This PR adds a new http.connection_pool_ttl setting to control how long
a connection in the OIDC back-channel pool can be idle before it is
closed. This allows ES to more actively close idle connections to avoid
the timeout issue.

Resolves: elastic#75515

docs/reference/settings/security-settings.asciidoc

@@ -1858,6 +1858,21 @@ connections allowed per endpoint.
 Defaults to `200`.
 // end::oidc-http-max-endpoint-connections-tag[]
 
+// tag::oidc-http-connection-pool-ttl-tag[]
+`http.connection_pool_ttl` {ess-icon}::
+(<<static-cluster-setting,Static>>)
+Controls the behavior of the http client used for back-channel communication to
+the OpenID Connect Provider endpoints. Specifies the time-to-live of connections
+in the connection pool. If a connection is not re-used within this timeout,
+it is closed.
+By default, the time-to-live is 3 minutes meaning that connections are closed
+after being idle for 3 minutes.
+
+Configure this setting to `-1` to let the server dictates this value using the `Keep-Alive` HTTP
+response header. If the header is not set by the server, the time-to-live is infinite meaning
+that connections never expire.
+// end::oidc-http-connection-pool-ttl-tag[]
+
 [discrete]
 [[ref-oidc-ssl-settings]]
 ===== OpenID Connect realm SSL settings

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/oidc/OpenIdConnectRealmSettings.java

@@ -23,6 +23,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
 public class OpenIdConnectRealmSettings {
@@ -212,6 +213,12 @@ private OpenIdConnectRealmSettings() {}
         "http.max_endpoint_connections",
         key -> Setting.intSetting(key, 200, Setting.Property.NodeScope)
     );
+
+    public static final Setting.AffixSetting<TimeValue> HTTP_CONNECTION_POOL_TTL = Setting.affixKeySetting(
+        RealmSettings.realmSettingPrefix(TYPE),
+        "http.connection_pool_ttl",
+        key -> Setting.timeSetting(key, new TimeValue(3, TimeUnit.MINUTES), Setting.Property.NodeScope)
+    );
     public static final Setting.AffixSetting<String> HTTP_PROXY_HOST = Setting.affixKeySetting(
         RealmSettings.realmSettingPrefix(TYPE),
         "http.proxy.host",
@@ -307,6 +314,7 @@ public static Set<Setting.AffixSetting<?>> getSettings() {
             HTTP_SOCKET_TIMEOUT,
             HTTP_MAX_CONNECTIONS,
             HTTP_MAX_ENDPOINT_CONNECTIONS,
+            HTTP_CONNECTION_POOL_TTL,
             HTTP_PROXY_HOST,
             HTTP_PROXY_PORT,
             HTTP_PROXY_SCHEME,

x-pack/plugin/security/qa/smoke-test-all-realms/build.gradle

@@ -76,6 +76,7 @@ testClusters.matching { it.name == 'javaRestTest' }.configureEach {
   setting 'xpack.security.authc.realms.oidc.openid7.op.authorization_endpoint', 'https://op.example.com/auth'
   setting 'xpack.security.authc.realms.oidc.openid7.op.jwkset_path', 'oidc-jwkset.json'
   setting 'xpack.security.authc.realms.oidc.openid7.claims.principal', 'sub'
+  setting 'xpack.security.authc.realms.oidc.openid7.http.connection_pool_ttl', '1m'
   keystore 'xpack.security.authc.realms.oidc.openid7.rp.client_secret', 'this-is-my-secret'
   //  - JWT (works)
   setting 'xpack.security.authc.realms.jwt.jwt8.order', '8'

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authc/oidc/OpenIdConnectAuthenticator.java

@@ -60,6 +60,7 @@
 import org.apache.http.config.RegistryBuilder;
 import org.apache.http.entity.ContentType;
 import org.apache.http.impl.auth.BasicScheme;
+import org.apache.http.impl.client.DefaultConnectionKeepAliveStrategy;
 import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
 import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
 import org.apache.http.impl.nio.client.HttpAsyncClients;
@@ -81,6 +82,7 @@
 import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.core.CheckedRunnable;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.watcher.FileChangesListener;
@@ -114,6 +116,7 @@
 
 import static org.elasticsearch.core.Strings.format;
 import static org.elasticsearch.xpack.core.security.authc.oidc.OpenIdConnectRealmSettings.ALLOWED_CLOCK_SKEW;
+import static org.elasticsearch.xpack.core.security.authc.oidc.OpenIdConnectRealmSettings.HTTP_CONNECTION_POOL_TTL;
 import static org.elasticsearch.xpack.core.security.authc.oidc.OpenIdConnectRealmSettings.HTTP_CONNECTION_READ_TIMEOUT;
 import static org.elasticsearch.xpack.core.security.authc.oidc.OpenIdConnectRealmSettings.HTTP_CONNECT_TIMEOUT;
 import static org.elasticsearch.xpack.core.security.authc.oidc.OpenIdConnectRealmSettings.HTTP_MAX_CONNECTIONS;
@@ -707,7 +710,15 @@ private CloseableHttpAsyncClient createHttpClient() {
                     .build();
                 HttpAsyncClientBuilder httpAsyncClientBuilder = HttpAsyncClients.custom()
                     .setConnectionManager(connectionManager)
-                    .setDefaultRequestConfig(requestConfig);
+                    .setDefaultRequestConfig(requestConfig)
+                    .setKeepAliveStrategy((response, context) -> {
+                        final TimeValue timeValue = realmConfig.getSetting(HTTP_CONNECTION_POOL_TTL);
+                        if (TimeValue.MINUS_ONE.equals(timeValue)) {
+                            return DefaultConnectionKeepAliveStrategy.INSTANCE.getKeepAliveDuration(response, context);
+                        } else {
+                            return timeValue.millis();
+                        }
+                    });
                 if (realmConfig.hasSetting(HTTP_PROXY_HOST)) {
                     httpAsyncClientBuilder.setProxy(
                         new HttpHost(