This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new fda742c48e perf(httpclient): add lightweight execution path for
retry=0 (#7824) (#7835)
fda742c48e is described below
commit fda742c48e7f553c9964cf74b36732326d0fde00
Author: Harsh Raj <[email protected]>
AuthorDate: Wed Oct 21 08:23:50 2026 +0530
perf(httpclient): add lightweight execution path for retry=0 (#7824) (#7835)
Co-authored-by: Harsh <[email protected]>
---
.../httpclient/AbstractHttpClientPlugin.java | 29 ++-
.../httpclient/AbstractHttpClientPluginTest.java | 230 +++++++++++++++++++++
2 files changed, 248 insertions(+), 11 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/AbstractHttpClientPlugin.java
b/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/AbstractHttpClientPlugin.java
index d35c77ce67..212ca341a4 100644
---
a/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/AbstractHttpClientPlugin.java
+++
b/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/AbstractHttpClientPlugin.java
@@ -104,9 +104,23 @@ public abstract class AbstractHttpClientPlugin<R>
implements ShenyuPlugin {
final Mono<R> response = doRequest(exchange, httpMethod, uri,
requestBody)
.timeout(duration, Mono.error(() -> new
TimeoutException("Response took longer than timeout: " + duration)))
.doOnError(e -> LOG.error(e.getMessage(), e));
- RetryStrategy<R> strategy;
- String retryStrategyType = (String)
Optional.ofNullable(exchange.getAttribute(Constants.HTTP_RETRY_BACK_OFF_SPEC)).orElse(HttpRetryBackoffSpecEnum.getDefault());
- HttpRetryBackoffSpecEnum spec =
HttpRetryBackoffSpecEnum.acquireByName(retryStrategyType);
+ final Mono<R> retriedResponse = retryTimes <= 0
+ ? response
+ : executeWithRetry(response, exchange, duration, retryTimes);
+ return retriedResponse
+ .onErrorMap(ShenyuException.class, th -> new
ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
+
ShenyuResultEnum.CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER.getMsg(), th))
+ .onErrorMap(java.util.concurrent.TimeoutException.class, th ->
new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT, th.getMessage(), th))
+ .onErrorMap(DataBufferLimitException.class, th -> new
ResponseStatusException(HttpStatus.PAYLOAD_TOO_LARGE,
+ "Request body exceeds the maxInMemorySize limit and
cannot be cached for retry", th))
+ .flatMap((Function<Object, Mono<? extends Void>>) o ->
chain.execute(exchange));
+ }
+
+ private Mono<R> executeWithRetry(final Mono<R> response, final
ServerWebExchange exchange, final Duration duration, final int retryTimes) {
+ final RetryStrategy<R> strategy;
+ final String retryStrategyType = (String)
Optional.ofNullable(exchange.getAttribute(Constants.HTTP_RETRY_BACK_OFF_SPEC))
+ .orElse(HttpRetryBackoffSpecEnum.getDefault());
+ final HttpRetryBackoffSpecEnum spec =
HttpRetryBackoffSpecEnum.acquireByName(retryStrategyType);
switch (spec) {
case EXPONENTIAL_BACKOFF:
strategy = new ExponentialRetryBackoffStrategy<>(this);
@@ -117,14 +131,7 @@ public abstract class AbstractHttpClientPlugin<R>
implements ShenyuPlugin {
default:
strategy = new DefaultRetryStrategy<>(this);
}
- Mono<R> retriedResponse = strategy.execute(response, exchange,
duration, retryTimes);
- return retriedResponse
- .onErrorMap(ShenyuException.class, th -> new
ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
-
ShenyuResultEnum.CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER.getMsg(), th))
- .onErrorMap(java.util.concurrent.TimeoutException.class, th ->
new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT, th.getMessage(), th))
- .onErrorMap(DataBufferLimitException.class, th -> new
ResponseStatusException(HttpStatus.PAYLOAD_TOO_LARGE,
- "Request body exceeds the maxInMemorySize limit and
cannot be cached for retry", th))
- .flatMap((Function<Object, Mono<? extends Void>>) o ->
chain.execute(exchange));
+ return strategy.execute(response, exchange, duration, retryTimes);
}
/**
diff --git
a/shenyu-plugin/shenyu-plugin-httpclient/src/test/java/org/apache/shenyu/plugin/httpclient/AbstractHttpClientPluginTest.java
b/shenyu-plugin/shenyu-plugin-httpclient/src/test/java/org/apache/shenyu/plugin/httpclient/AbstractHttpClientPluginTest.java
new file mode 100644
index 0000000000..da5d1f063b
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-httpclient/src/test/java/org/apache/shenyu/plugin/httpclient/AbstractHttpClientPluginTest.java
@@ -0,0 +1,230 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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 org.apache.shenyu.plugin.httpclient;
+
+import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.common.enums.RetryEnum;
+import org.apache.shenyu.common.exception.ShenyuException;
+import org.apache.shenyu.plugin.api.ShenyuPluginChain;
+import org.apache.shenyu.plugin.api.context.ShenyuContext;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferLimitException;
+import org.springframework.http.HttpStatus;
+import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
+import org.springframework.mock.web.server.MockServerWebExchange;
+import org.springframework.web.server.ResponseStatusException;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import java.net.ConnectException;
+import java.net.URI;
+import java.util.Objects;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * The test case for {@link AbstractHttpClientPlugin}.
+ */
+public final class AbstractHttpClientPluginTest {
+
+ private ShenyuPluginChain chain;
+
+ @BeforeEach
+ public void setUp() {
+ chain = mock(ShenyuPluginChain.class);
+ when(chain.execute(any())).thenReturn(Mono.empty());
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, -1})
+ public void testZeroRetrySuccess(final int retryTimes) {
+ AtomicInteger attempts = new AtomicInteger();
+ TestHttpClientPlugin plugin = new TestHttpClientPlugin(exchange -> {
+ attempts.incrementAndGet();
+ return Mono.just("success");
+ });
+ ServerWebExchange exchange = createExchange(retryTimes, null, null);
+ StepVerifier.create(plugin.execute(exchange, chain))
+ .verifyComplete();
+ assertEquals(1, attempts.get());
+ }
+
+ @Test
+ public void testZeroRetryDefaultTimeoutMappedTo504() {
+ AtomicInteger attempts = new AtomicInteger();
+ TestHttpClientPlugin plugin = new TestHttpClientPlugin(exchange -> {
+ attempts.incrementAndGet();
+ return Mono.error(new TimeoutException("timeout"));
+ });
+ ServerWebExchange exchange = createExchange(0, null, null);
+ StepVerifier.create(plugin.execute(exchange, chain))
+ .expectErrorSatisfies(th -> {
+ ResponseStatusException statusException =
assertInstanceOf(ResponseStatusException.class, th);
+ assertEquals(HttpStatus.GATEWAY_TIMEOUT,
statusException.getStatusCode());
+ })
+ .verify();
+ assertEquals(1, attempts.get());
+ }
+
+ @Test
+ public void testZeroRetryDataBufferLimitExceptionMappedTo413() {
+ AtomicInteger attempts = new AtomicInteger();
+ TestHttpClientPlugin plugin = new TestHttpClientPlugin(exchange -> {
+ attempts.incrementAndGet();
+ return Mono.error(new DataBufferLimitException("too large"));
+ });
+ ServerWebExchange exchange = createExchange(0, null, null);
+ StepVerifier.create(plugin.execute(exchange, chain))
+ .expectErrorSatisfies(th -> {
+ ResponseStatusException statusException =
assertInstanceOf(ResponseStatusException.class, th);
+ assertEquals(HttpStatus.PAYLOAD_TOO_LARGE,
statusException.getStatusCode());
+ })
+ .verify();
+ assertEquals(1, attempts.get());
+ }
+
+ @Test
+ public void testZeroRetryShenyuExceptionMappedTo503() {
+ AtomicInteger attempts = new AtomicInteger();
+ TestHttpClientPlugin plugin = new TestHttpClientPlugin(exchange -> {
+ attempts.incrementAndGet();
+ return Mono.error(new ShenyuException("upstream unavailable"));
+ });
+ ServerWebExchange exchange = createExchange(0, null, null);
+ StepVerifier.create(plugin.execute(exchange, chain))
+ .expectErrorSatisfies(th -> {
+ ResponseStatusException statusException =
assertInstanceOf(ResponseStatusException.class, th);
+ assertEquals(HttpStatus.SERVICE_UNAVAILABLE,
statusException.getStatusCode());
+ })
+ .verify();
+ assertEquals(1, attempts.get());
+ }
+
+ @Test
+ public void testZeroRetryOtherExceptionPropagates() {
+ AtomicInteger attempts = new AtomicInteger();
+ ConnectException failure = new ConnectException("connection refused");
+ TestHttpClientPlugin plugin = new TestHttpClientPlugin(exchange -> {
+ attempts.incrementAndGet();
+ return Mono.error(failure);
+ });
+ ServerWebExchange exchange = createExchange(0, null, null);
+ StepVerifier.create(plugin.execute(exchange, chain))
+ .expectErrorMatches(th -> th == failure)
+ .verify();
+ assertEquals(1, attempts.get());
+ }
+
+ @Test
+ public void testZeroRetryFailoverStrategyTimeoutMappedTo504() {
+ AtomicInteger attempts = new AtomicInteger();
+ TestHttpClientPlugin plugin = new TestHttpClientPlugin(exchange -> {
+ attempts.incrementAndGet();
+ return Mono.error(new TimeoutException("timeout"));
+ });
+ ServerWebExchange exchange = createExchange(0,
RetryEnum.FAILOVER.getName(), null);
+ StepVerifier.create(plugin.execute(exchange, chain))
+ .expectErrorSatisfies(th -> {
+ ResponseStatusException statusException =
assertInstanceOf(ResponseStatusException.class, th);
+ assertEquals(HttpStatus.GATEWAY_TIMEOUT,
statusException.getStatusCode());
+ })
+ .verify();
+ assertEquals(1, attempts.get());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"fixed", "exponential"})
+ public void testZeroRetryNonDefaultSpecTimeoutMappedTo504(final String
spec) {
+ AtomicInteger attempts = new AtomicInteger();
+ TestHttpClientPlugin plugin = new TestHttpClientPlugin(exchange -> {
+ attempts.incrementAndGet();
+ return Mono.error(new TimeoutException("timeout"));
+ });
+ ServerWebExchange exchange = createExchange(0, null, spec);
+ StepVerifier.create(plugin.execute(exchange, chain))
+ .expectErrorSatisfies(th -> {
+ ResponseStatusException statusException =
assertInstanceOf(ResponseStatusException.class, th);
+ assertEquals(HttpStatus.GATEWAY_TIMEOUT,
statusException.getStatusCode());
+ })
+ .verify();
+ assertEquals(1, attempts.get());
+ }
+
+ @Test
+ public void testPositiveRetryExecutesStrategy() {
+ AtomicInteger attempts = new AtomicInteger();
+ TestHttpClientPlugin plugin = new TestHttpClientPlugin(exchange -> {
+ if (attempts.incrementAndGet() == 1) {
+ return Mono.error(new TimeoutException("transient timeout"));
+ }
+ return Mono.just("success");
+ });
+ ServerWebExchange exchange = createExchange(2, null, null);
+ StepVerifier.create(plugin.execute(exchange, chain))
+ .verifyComplete();
+ assertEquals(2, attempts.get());
+ }
+
+ private ServerWebExchange createExchange(final int retryTimes, final
String retryStrategy, final String retrySpec) {
+ ServerWebExchange exchange =
MockServerWebExchange.from(MockServerHttpRequest.get("/test").build());
+ exchange.getAttributes().put(Constants.CONTEXT,
mock(ShenyuContext.class));
+ exchange.getAttributes().put(Constants.HTTP_URI,
URI.create("http://localhost:8080/test";));
+ exchange.getAttributes().put(Constants.HTTP_RETRY, retryTimes);
+ exchange.getAttributes().put(Constants.HTTP_TIME_OUT, 3000L);
+ if (Objects.nonNull(retryStrategy)) {
+ exchange.getAttributes().put(Constants.RETRY_STRATEGY,
retryStrategy);
+ }
+ if (Objects.nonNull(retrySpec)) {
+ exchange.getAttributes().put(Constants.HTTP_RETRY_BACK_OFF_SPEC,
retrySpec);
+ }
+ return exchange;
+ }
+
+ private static final class TestHttpClientPlugin extends
AbstractHttpClientPlugin<String> {
+
+ private final Function<ServerWebExchange, Mono<String>>
responseFunction;
+
+ TestHttpClientPlugin(final Function<ServerWebExchange, Mono<String>>
responseFunction) {
+ super(Constants.BYTES_PER_MB);
+ this.responseFunction = responseFunction;
+ }
+
+ @Override
+ protected Mono<String> doRequest(final ServerWebExchange exchange,
final String httpMethod, final URI uri, final Flux<DataBuffer> body) {
+ return Mono.defer(() -> responseFunction.apply(exchange));
+ }
+
+ @Override
+ public int getOrder() {
+ return 0;
+ }
+ }
+}
