From 0b5b6909d149f344debad90087caa85411014766 Mon Sep 17 00:00:00 2001 From: hidingmyname <140602238+hidingmyname@users.noreply.github.com> Date: Fri, 6 Nov 2026 01:17:40 +0800 Subject: [PATCH] Rest5Client: do not retry requests with a non-repeatable entity A non-repeatable entity, such as an InputStreamEntity, is consumed by the first attempt. Retrying it on the next node sends an empty body when the length is unknown, so the request can appear to succeed, and hangs when a content length is set. Stop at the first failure for these requests, in the synchronous and asynchronous paths. Fixes #1356 --- .../rest5_client/low_level/Rest5Client.java | 20 +- .../RestClientStreamBodyRetryTests.java | 299 ++++++++++++++++++ 2 files changed, 314 insertions(+), 5 deletions(-) create mode 100644 rest5-client/src/test/java/co/elastic/clients/transport/rest5_client/low_level/RestClientStreamBodyRetryTests.java diff --git a/rest5-client/src/main/java/co/elastic/clients/transport/rest5_client/low_level/Rest5Client.java b/rest5-client/src/main/java/co/elastic/clients/transport/rest5_client/low_level/Rest5Client.java index 4d10b69569..695a7f3da5 100644 --- a/rest5-client/src/main/java/co/elastic/clients/transport/rest5_client/low_level/Rest5Client.java +++ b/rest5-client/src/main/java/co/elastic/clients/transport/rest5_client/low_level/Rest5Client.java @@ -101,7 +101,8 @@ * failed (the more failures, the later they will be retried). In case of failures all of the alive nodes * (or dead nodes that * deserve a retry) are retried until one responds or none of them does, in which case an - * {@link IOException} will be thrown. + * {@link IOException} will be thrown. Requests with a non-repeatable entity are not retried, as their body + * can only be sent once. *

* Requests can be either synchronous or asynchronous. The asynchronous variants all end with {@code Async}. *

@@ -308,7 +309,7 @@ private Response performRequest(final Iterator nodes, final InternalReques onFailure(context.node); Exception cause = extractAndWrapCause(e); addSuppressedException(previousException, cause); - if (isRetryableException(e) && nodes.hasNext()) { + if (isRetryableException(e) && request.canRetry() && nodes.hasNext()) { return performRequest(nodes, request, cause); } if (cause instanceof IOException) { @@ -326,7 +327,7 @@ private Response performRequest(final Iterator nodes, final InternalReques return responseOrResponseException.response; } addSuppressedException(previousException, responseOrResponseException.responseException); - if (nodes.hasNext()) { + if (request.canRetry() && nodes.hasNext()) { return performRequest(nodes, request, responseOrResponseException.responseException); } throw responseOrResponseException.responseException; @@ -422,7 +423,7 @@ public void completed(ClassicHttpResponse httpResponse) { if (responseOrResponseException.responseException == null) { listener.onSuccess(responseOrResponseException.response); } else { - if (nodes.hasNext()) { + if (request.canRetry() && nodes.hasNext()) { listener.trackFailure(responseOrResponseException.responseException); performRequestAsync(nodes, request, listener); } else { @@ -440,7 +441,7 @@ public void failed(Exception failure) { RequestLogger.logFailedRequest(logger, request.httpRequest, context.node, failure); onFailure(context.node); - if (isRetryableException(failure) && nodes.hasNext()) { + if (isRetryableException(failure) && request.canRetry() && nodes.hasNext()) { listener.trackFailure(failure); performRequestAsync(nodes, request, listener); } else { @@ -859,6 +860,15 @@ private void setRequestConfig(HttpUriRequestBase requestBase, RequestConfig requ } } + /** + * A non-repeatable entity (e.g. backed by an {@code InputStream}) is consumed by the first attempt, and + * sending it again would produce an empty or incomplete body. + */ + boolean canRetry() { + HttpEntity entity = httpRequest.getEntity(); + return entity == null || entity.isRepeatable(); + } + RequestContext createContextForNextAttempt(Node node) { this.httpRequest.reset(); return new RequestContext(this, node); diff --git a/rest5-client/src/test/java/co/elastic/clients/transport/rest5_client/low_level/RestClientStreamBodyRetryTests.java b/rest5-client/src/test/java/co/elastic/clients/transport/rest5_client/low_level/RestClientStreamBodyRetryTests.java new file mode 100644 index 0000000000..90b4eab5a3 --- /dev/null +++ b/rest5-client/src/test/java/co/elastic/clients/transport/rest5_client/low_level/RestClientStreamBodyRetryTests.java @@ -0,0 +1,299 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. licenses this file to you 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 co.elastic.clients.transport.rest5_client.low_level; + +import com.sun.net.httpserver.HttpHandler; +import com.sun.net.httpserver.HttpServer; +import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient; +import org.apache.hc.client5.http.protocol.HttpClientContext; +import org.apache.hc.core5.concurrent.FutureCallback; +import org.apache.hc.core5.http.ContentType; +import org.apache.hc.core5.http.Header; +import org.apache.hc.core5.http.HttpHost; +import org.apache.hc.core5.http.HttpResponse; +import org.apache.hc.core5.http.io.entity.InputStreamEntity; +import org.apache.hc.core5.http.message.BasicClassicHttpResponse; +import org.apache.hc.core5.http.nio.AsyncRequestProducer; +import org.apache.hc.core5.http.nio.AsyncResponseConsumer; +import org.elasticsearch.mocksocket.MockHttpServer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.mockito.stubbing.Answer; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static java.util.Arrays.asList; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.nullable; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Tests that {@link Rest5Client} does not retry requests when the body is a non-repeatable + * entity (e.g. backed by an InputStream), since the stream is consumed on the first attempt + * and cannot be re-read. + */ +public class RestClientStreamBodyRetryTests extends RestClientTestCase { + + private final ExecutorService exec = Executors.newFixedThreadPool(1); + + @AfterEach + public void shutdownExec() { + exec.shutdown(); + } + + /** + * A retryable status code (502/503/504) must not cause a stream body to be sent again. + */ + @Test + public void testNoRetryWithStreamBodyOnRetryableStatusCode() throws Exception { + AtomicInteger requestCount = new AtomicInteger(0); + CloseableHttpAsyncClient httpClient = mockHttpClientWithCounter(exec, requestCount, 503); + Rest5Client restClient = newRestClient(httpClient); + + ResponseException e = assertThrows(ResponseException.class, () -> restClient.performRequest(newStreamBodyRequest())); + + assertEquals(503, e.getResponse().getStatusCode()); + assertEquals(1, requestCount.get()); + } + + /** + * A retryable IOException must not cause a stream body to be sent again. + */ + @Test + public void testNoRetryWithStreamBodyOnIOException() throws Exception { + AtomicInteger requestCount = new AtomicInteger(0); + CloseableHttpAsyncClient httpClient = mockHttpClientThatThrows(exec, requestCount); + Rest5Client restClient = newRestClient(httpClient); + + assertThrows(IOException.class, () -> restClient.performRequest(newStreamBodyRequest())); + + assertEquals(1, requestCount.get()); + } + + /** + * Requests with a repeatable entity are still retried on all nodes. + */ + @Test + public void testRetryStillWorksWithRepeatableEntity() throws Exception { + AtomicInteger requestCount = new AtomicInteger(0); + CloseableHttpAsyncClient httpClient = mockHttpClientWithCounter(exec, requestCount, 503); + Rest5Client restClient = newRestClient(httpClient); + + Request request = new Request("POST", "/hello"); + request.setJsonEntity("{\"hello\":\"world\"}"); + + ResponseException e = assertThrows(ResponseException.class, () -> restClient.performRequest(request)); + + assertEquals(503, e.getResponse().getStatusCode()); + assertEquals(3, requestCount.get()); + } + + /** + * Same as {@link #testNoRetryWithStreamBodyOnRetryableStatusCode()} for the asynchronous API. + */ + @Test + public void testNoRetryWithStreamBodyOnRetryableStatusCodeAsync() throws Exception { + AtomicInteger requestCount = new AtomicInteger(0); + CloseableHttpAsyncClient httpClient = mockHttpClientWithCounter(exec, requestCount, 503); + + Exception failure = performAsyncAndGetFailure(newRestClient(httpClient), newStreamBodyRequest()); + + assertTrue(failure instanceof ResponseException); + assertEquals(503, ((ResponseException) failure).getResponse().getStatusCode()); + assertEquals(1, requestCount.get()); + } + + /** + * Same as {@link #testNoRetryWithStreamBodyOnIOException()} for the asynchronous API. + */ + @Test + public void testNoRetryWithStreamBodyOnIOExceptionAsync() throws Exception { + AtomicInteger requestCount = new AtomicInteger(0); + CloseableHttpAsyncClient httpClient = mockHttpClientThatThrows(exec, requestCount); + + Exception failure = performAsyncAndGetFailure(newRestClient(httpClient), newStreamBodyRequest()); + + assertTrue(failure instanceof IOException); + assertEquals(1, requestCount.get()); + } + + /** + * Against real http servers: the first node answers 503, and the stream must not be sent to the second + * one, as it would receive an empty body. + */ + @Test + public void testStreamBodyIsNotResentToNextNode() throws Exception { + List receivedBodies = Collections.synchronizedList(new ArrayList<>()); + HttpHandler handler = exchange -> { + receivedBodies.add(new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8)); + exchange.sendResponseHeaders(receivedBodies.size() == 1 ? 503 : 200, -1); + exchange.close(); + }; + HttpServer first = MockHttpServer.createHttp(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0); + HttpServer second = MockHttpServer.createHttp(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0); + first.createContext("/", handler); + second.createContext("/", handler); + first.start(); + second.start(); + + String body = "{\"hello\":\"world\"}"; + Request request = new Request("POST", "/hello"); + request.setEntity(new InputStreamEntity( + new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)), ContentType.APPLICATION_JSON + )); + + try (Rest5Client restClient = Rest5Client.builder( + new HttpHost(first.getAddress().getHostString(), first.getAddress().getPort()), + new HttpHost(second.getAddress().getHostString(), second.getAddress().getPort()) + ).build()) { + restClient.performRequest(request); + } catch (ResponseException e) { + assertEquals(503, e.getResponse().getStatusCode()); + } finally { + first.stop(0); + second.stop(0); + } + + assertEquals(List.of(body), receivedBodies); + } + + private static Rest5Client newRestClient(CloseableHttpAsyncClient httpClient) { + List nodes = asList( + new Node(new HttpHost("localhost", 9200)), + new Node(new HttpHost("localhost", 9201)), + new Node(new HttpHost("localhost", 9202)) + ); + return new Rest5Client( + httpClient, + new Header[0], + nodes, + null, + new HostsTrackingFailureListener(), + NodeSelector.ANY, + false, + false, + false + ); + } + + private static Request newStreamBodyRequest() { + byte[] body = "{\"hello\":\"world\"}".getBytes(StandardCharsets.UTF_8); + Request request = new Request("POST", "/hello"); + request.setEntity(new InputStreamEntity( + new ByteArrayInputStream(body), body.length, ContentType.APPLICATION_JSON + )); + return request; + } + + private static Exception performAsyncAndGetFailure(Rest5Client restClient, Request request) throws Exception { + CountDownLatch latch = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + restClient.performRequestAsync(request, new ResponseListener() { + @Override + public void onSuccess(Response response) { + latch.countDown(); + } + + @Override + public void onFailure(Exception exception) { + failure.set(exception); + latch.countDown(); + } + }); + assertTrue(latch.await(30, TimeUnit.SECONDS)); + assertNotNull(failure.get()); + return failure.get(); + } + + @SuppressWarnings("unchecked") + private static CloseableHttpAsyncClient mockHttpClientWithCounter( + ExecutorService exec, AtomicInteger requestCount, int statusCode + ) { + CloseableHttpAsyncClient httpClient = mock(CloseableHttpAsyncClient.class); + when( + httpClient.execute( + any(AsyncRequestProducer.class), + any(AsyncResponseConsumer.class), + any(HttpClientContext.class), + nullable(FutureCallback.class) + ) + ).thenAnswer((Answer>) invocationOnMock -> { + final FutureCallback futureCallback = + (FutureCallback) invocationOnMock.getArguments()[3]; + return exec.submit(() -> { + requestCount.incrementAndGet(); + BasicClassicHttpResponse response = new BasicClassicHttpResponse(statusCode, ""); + if (futureCallback != null) { + futureCallback.completed(response); + return null; + } + return (HttpResponse) response; + }); + }); + return httpClient; + } + + @SuppressWarnings("unchecked") + private static CloseableHttpAsyncClient mockHttpClientThatThrows( + ExecutorService exec, AtomicInteger requestCount + ) { + CloseableHttpAsyncClient httpClient = mock(CloseableHttpAsyncClient.class); + when( + httpClient.execute( + any(AsyncRequestProducer.class), + any(AsyncResponseConsumer.class), + any(HttpClientContext.class), + nullable(FutureCallback.class) + ) + ).thenAnswer((Answer>) invocationOnMock -> { + final FutureCallback futureCallback = + (FutureCallback) invocationOnMock.getArguments()[3]; + return exec.submit(() -> { + requestCount.incrementAndGet(); + IOException exception = new IOException("Connection failed"); + if (futureCallback != null) { + futureCallback.failed(exception); + return null; + } + throw exception; + }); + }); + return httpClient; + } +}