Add an option to drop pooled connection after a 503 response We observe that sometimes Gerrit primary pods get a 503 response when sending REST calls to each other. This happens when istio is used to manage connections and one of the pods is either down or restarted. In such cases, istio often continues to return 503 response even if the target pod, and its IP address, is unavailable. The problem continues until such a connection eventually expires and gets removed from the connection pool. This often leads to missed or delayed messages between the primary pods which leads to diverging indexes and inconsistent caches. This change adds a new option, http.reuseConnectionAfter503. When set to false, a connection over which we receive a 503 response will be immediately declared as non-reusable and will be removed from the apache's http client connection pool. Change-Id: I441c3ade9561e44a8841f346928d1273e38e39df
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/Configuration.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/Configuration.java index 4dc9f45..15425d3 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/Configuration.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/Configuration.java
@@ -516,6 +516,8 @@ static final String MAX_TRIES_KEY = "maxTries"; static final String RETRY_INTERVAL_KEY = "retryInterval"; static final String THREAD_POOL_SIZE_KEY = "threadPoolSize"; + static final String REUSE_CONNECTION_AFTER_503_KEY = "reuseConnectionAfter503"; + static final boolean DEFAULT_REUSE_CONNECTION_AFTER_503 = true; private final String user; private final String password; @@ -524,6 +526,7 @@ private final int maxTries; private final Duration retryInterval; private final int threadPoolSize; + private final boolean reuseConnectionAfter503; private Http(Config cfg) { user = Strings.nullToEmpty(cfg.getString(HTTP_SECTION, null, USER_KEY)); @@ -533,6 +536,9 @@ maxTries = getMaxTries(cfg, HTTP_SECTION, MAX_TRIES_KEY, DEFAULT_MAX_TRIES); retryInterval = getDuration(cfg, HTTP_SECTION, RETRY_INTERVAL_KEY, DEFAULT_RETRY_INTERVAL); threadPoolSize = getInt(cfg, HTTP_SECTION, THREAD_POOL_SIZE_KEY, DEFAULT_THREAD_POOL_SIZE); + reuseConnectionAfter503 = + cfg.getBoolean( + HTTP_SECTION, REUSE_CONNECTION_AFTER_503_KEY, DEFAULT_REUSE_CONNECTION_AFTER_503); } public String user() { @@ -562,6 +568,10 @@ public int threadPoolSize() { return threadPoolSize; } + + public boolean reuseConnectionAfter503() { + return reuseConnectionAfter503; + } } /** Common parameters to cache, event, index and websession */
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/HttpClientProvider.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/HttpClientProvider.java index a71e03f..fc90a10 100644 --- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/HttpClientProvider.java +++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/HttpClientProvider.java
@@ -14,6 +14,8 @@ package com.ericsson.gerrit.plugins.highavailability.forwarder.rest; +import static javax.servlet.http.HttpServletResponse.SC_SERVICE_UNAVAILABLE; + import com.ericsson.gerrit.plugins.highavailability.Configuration; import com.google.common.flogger.FluentLogger; import com.google.inject.Inject; @@ -24,6 +26,7 @@ import javax.net.ssl.SSLContext; import javax.net.ssl.TrustManager; import javax.net.ssl.X509TrustManager; +import org.apache.http.ConnectionReuseStrategy; import org.apache.http.auth.AuthScope; import org.apache.http.auth.UsernamePasswordCredentials; import org.apache.http.client.config.RequestConfig; @@ -34,6 +37,7 @@ import org.apache.http.conn.socket.PlainConnectionSocketFactory; import org.apache.http.conn.ssl.NoopHostnameVerifier; import org.apache.http.conn.ssl.SSLConnectionSocketFactory; +import org.apache.http.impl.DefaultConnectionReuseStrategy; import org.apache.http.impl.client.BasicCredentialsProvider; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClients; @@ -64,6 +68,7 @@ .setConnectionManager(customConnectionManager()) .setDefaultCredentialsProvider(buildCredentials()) .setDefaultRequestConfig(customRequestConfig()) + .setConnectionReuseStrategy(customConnectionReuseStrategy()) .build(); } @@ -77,6 +82,10 @@ } private HttpClientConnectionManager customConnectionManager() { + return buildConnectionManager(); + } + + PoolingHttpClientConnectionManager buildConnectionManager() { Registry<ConnectionSocketFactory> socketFactoryRegistry = RegistryBuilder.<ConnectionSocketFactory>create() .register("https", sslSocketFactory) @@ -90,6 +99,16 @@ return connManager; } + private ConnectionReuseStrategy customConnectionReuseStrategy() { + return (response, context) -> { + if (response.getStatusLine().getStatusCode() == SC_SERVICE_UNAVAILABLE + && !cfg.http().reuseConnectionAfter503()) { + return false; + } + return DefaultConnectionReuseStrategy.INSTANCE.keepAlive(response, context); + }; + } + private static SSLConnectionSocketFactory buildSslSocketFactory() { return new SSLConnectionSocketFactory(buildSslContext(), NoopHostnameVerifier.INSTANCE); }
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md index 737b354..1a3a498 100644 --- a/src/main/resources/Documentation/config.md +++ b/src/main/resources/Documentation/config.md
@@ -294,6 +294,14 @@ Value is expressed in Gerrit time values as in [websession.cleanupInterval](#websessioncleanupInterval). When not specified, the default value is set to 10 seconds. +```http.reuseConnectionAfter503``` +: Whether to reuse the HTTP connection to the peer instance after receiving a + 503 (Service Unavailable) response over it. When set to `false`, the plugin + will close the connection immediately on a 503 response and open a fresh one + for the next retry, which can help recover from situations where a peer has + transiently become unavailable without properly closing existing connections. + When not specified, the default value is `true`. + ```http.threadPoolSize``` : Maximum number of threads used to execute REST calls towards target instances.
diff --git a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/HttpClientProviderConnectionReuseTest.java b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/HttpClientProviderConnectionReuseTest.java new file mode 100644 index 0000000..3bd3b14 --- /dev/null +++ b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/HttpClientProviderConnectionReuseTest.java
@@ -0,0 +1,121 @@ +// Copyright (C) 2026 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.ericsson.gerrit.plugins.highavailability.forwarder.rest; + +import static com.github.tomakehurst.wiremock.client.WireMock.aResponse; +import static com.github.tomakehurst.wiremock.client.WireMock.get; +import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo; +import static com.google.common.truth.Truth.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.ericsson.gerrit.plugins.highavailability.Configuration; +import com.github.tomakehurst.wiremock.junit.WireMockRule; +import java.time.Duration; +import java.util.List; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.conn.PoolingHttpClientConnectionManager; +import org.junit.Rule; +import org.junit.Test; +import org.mockito.Answers; + +public class HttpClientProviderConnectionReuseTest { + + private static final String ENDPOINT = "/test"; + private static final Duration TIMEOUT = Duration.ofMillis(1000); + + @Rule public WireMockRule wireMock = new WireMockRule(0); + + private Configuration mockConfig(boolean reuseConnectionAfter503) { + Configuration cfg = mock(Configuration.class, Answers.RETURNS_DEEP_STUBS); + when(cfg.http().user()).thenReturn(""); + when(cfg.http().password()).thenReturn(""); + when(cfg.http().connectionTimeout()).thenReturn(TIMEOUT); + when(cfg.http().socketTimeout()).thenReturn(TIMEOUT); + when(cfg.http().reuseConnectionAfter503()).thenReturn(reuseConnectionAfter503); + return cfg; + } + + @Test + public void connectionIsReturnedToPoolAfter503WhenReuseIsEnabled() throws Exception { + wireMock.givenThat(get(urlEqualTo(ENDPOINT)).willReturn(aResponse().withStatus(503))); + + TestableHttpClientProvider provider = new TestableHttpClientProvider(mockConfig(true)); + try (CloseableHttpClient client = provider.get()) { + executeConsuming(client); + // Connection must be back in the pool (available == 1) because reuseConnectionAfter503=true + assertThat(provider.connectionManager().getTotalStats().getAvailable()).isEqualTo(1); + } + } + + @Test + public void connectionIsDiscardedAfter503WhenReuseIsDisabled() throws Exception { + wireMock.givenThat(get(urlEqualTo(ENDPOINT)).willReturn(aResponse().withStatus(503))); + + TestableHttpClientProvider provider = new TestableHttpClientProvider(mockConfig(false)); + try (CloseableHttpClient client = provider.get()) { + executeConsuming(client); + // Connection must have been discarded (available == 0) because reuseConnectionAfter503=false + assertThat(provider.connectionManager().getTotalStats().getAvailable()).isEqualTo(0); + } + } + + @Test + public void connectionIsReturnedToPoolAfter500RegardlessOfReuseOption() throws Exception { + wireMock.givenThat(get(urlEqualTo(ENDPOINT)).willReturn(aResponse().withStatus(500))); + + for (boolean reuseFlag : List.of(true, false)) { + TestableHttpClientProvider provider = new TestableHttpClientProvider(mockConfig(reuseFlag)); + try (CloseableHttpClient client = provider.get()) { + executeConsuming(client); + // 500 is not 503 — the reuseConnectionAfter503 option must have no effect + assertThat(provider.connectionManager().getTotalStats().getAvailable()).isEqualTo(1); + } + } + } + + private void executeConsuming(CloseableHttpClient client) throws Exception { + String url = "http://localhost:" + wireMock.port() + ENDPOINT; + try (CloseableHttpResponse response = client.execute(new HttpGet(url))) { + if (response.getEntity() != null) { + response.getEntity().getContent().close(); + } + } + } + + /** + * Extends HttpClientProvider to capture the PoolingHttpClientConnectionManager so tests can + * inspect pool statistics after requests complete. + */ + private static class TestableHttpClientProvider extends HttpClientProvider { + private PoolingHttpClientConnectionManager connManager; + + TestableHttpClientProvider(Configuration cfg) { + super(cfg); + } + + @Override + PoolingHttpClientConnectionManager buildConnectionManager() { + connManager = super.buildConnectionManager(); + return connManager; + } + + PoolingHttpClientConnectionManager connectionManager() { + return connManager; + } + } +}