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 0e6216d6ed fix: enforce single half-open circuit breaker probe (#7802)
0e6216d6ed is described below
commit 0e6216d6eda676025849ac38e70d7c144111a384
Author: IT-Althusser <[email protected]>
AuthorDate: Tue Oct 20 12:38:23 2026 +0800
fix: enforce single half-open circuit breaker probe (#7802)
* fix: enforce single half-open circuit breaker probe
* fix: restore per-upstream circuit breaker recovery
---
.../circuitbreaker/UpstreamCircuitBreaker.java | 44 +++++++--
.../circuitbreaker/UpstreamCircuitBreakerTest.java | 53 +++++++++++
.../apache/shenyu/plugin/divide/DividePlugin.java | 83 ++++++++++++-----
.../divide/DividePluginCircuitBreakerTest.java | 102 +++++++++++++++++++--
4 files changed, 247 insertions(+), 35 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreaker.java
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreaker.java
index 18c3c40918..89388ef6bc 100644
---
a/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreaker.java
+++
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreaker.java
@@ -100,8 +100,9 @@ public final class UpstreamCircuitBreaker {
/**
* Whether a request to the given upstream is currently allowed.
*
- * <p>Requests are allowed when the breaker is closed, half-open, or open
- * but already past its half-open wait window (i.e. eligible to be probed).
+ * <p>Requests are allowed only when the breaker is closed. An open breaker
+ * must be acquired through {@link #tryAcquireHalfOpenProbe(String)} after
+ * its wait window has elapsed.
*
* @param key the breaker key
* @return true if a request may be sent to the upstream
@@ -152,6 +153,18 @@ public final class UpstreamCircuitBreaker {
.recordFailure();
}
+ /**
+ * Re-open a breaker when its acquired half-open probe is cancelled.
+ *
+ * @param key the breaker key
+ */
+ public void recordProbeCancellation(final String key) {
+ CircuitBreakerState state = this.breakers.get(key);
+ if (Objects.nonNull(state)) {
+ state.recordProbeCancellation();
+ }
+ }
+
/**
* Whether the breaker currently blocks requests for the given key.
*
@@ -185,11 +198,27 @@ public final class UpstreamCircuitBreaker {
* @return the wrapped mono with outcome recording attached
*/
public static <T> Mono<T> recordOutcome(final Mono<T> source, final String
key) {
+ return recordOutcome(source, key, false);
+ }
+
+ /**
+ * Wrap a forwarded {@link Mono} and record its outcome, including
cancellation
+ * of a request that was granted the half-open probe.
+ *
+ * @param <T> the element type of the source mono
+ * @param source the mono executing the upstream call
+ * @param key the breaker key
+ * @param probeGranted whether this request acquired the half-open probe
+ * @return the wrapped mono with outcome recording attached
+ */
+ public static <T> Mono<T> recordOutcome(final Mono<T> source, final String
key, final boolean probeGranted) {
return source.doFinally(signal -> {
if (signal == SignalType.ON_COMPLETE) {
INSTANCE.recordSuccess(key);
} else if (signal == SignalType.ON_ERROR) {
INSTANCE.recordFailure(key);
+ } else if (signal == SignalType.CANCEL && probeGranted) {
+ INSTANCE.recordProbeCancellation(key);
}
});
}
@@ -231,10 +260,7 @@ public final class UpstreamCircuitBreaker {
}
synchronized boolean isRequestAllowed() {
- if (this.state == State.OPEN) {
- return System.currentTimeMillis() - this.openedAtMillis >=
this.halfOpenWaitMillis;
- }
- return true;
+ return this.state == State.CLOSED;
}
synchronized boolean isBlocking() {
@@ -266,6 +292,12 @@ public final class UpstreamCircuitBreaker {
}
}
+ synchronized void recordProbeCancellation() {
+ if (this.state == State.HALF_OPEN) {
+ open();
+ }
+ }
+
private void open() {
this.state = State.OPEN;
this.openedAtMillis = System.currentTimeMillis();
diff --git
a/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreakerTest.java
b/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreakerTest.java
index 35b3c718c3..1c29d1b063 100644
---
a/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreakerTest.java
+++
b/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreakerTest.java
@@ -22,7 +22,9 @@ import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
+import java.lang.reflect.Field;
import java.util.List;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -90,6 +92,7 @@ public final class UpstreamCircuitBreakerTest {
// within the wait window no probe is granted
assertFalse(breaker.tryAcquireHalfOpenProbe(KEY));
TimeUnit.MILLISECONDS.sleep(150L);
+ assertFalse(breaker.isRequestAllowed(KEY));
// the first caller is granted the single probe
assertTrue(breaker.tryAcquireHalfOpenProbe(KEY));
// no second probe while the first one is in flight
@@ -114,6 +117,56 @@ public final class UpstreamCircuitBreakerTest {
assertFalse(breaker.isBlocking(KEY));
}
+ @Test
+ public void testCancelledHalfOpenProbeReopensBreaker() throws Exception {
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.getInstance();
+ breaker.reset(KEY);
+ breaker.recordFailure(KEY);
+ breaker.recordFailure(KEY);
+ breaker.recordFailure(KEY);
+ elapseWaitWindow(breaker);
+ assertTrue(breaker.tryAcquireHalfOpenProbe(KEY));
+
+ StepVerifier.create(UpstreamCircuitBreaker.recordOutcome(Mono.never(),
KEY))
+ .thenCancel()
+ .verify();
+ assertFalse(breaker.tryAcquireHalfOpenProbe(KEY));
+
+ StepVerifier.create(UpstreamCircuitBreaker.recordOutcome(Mono.never(),
KEY, true))
+ .thenCancel()
+ .verify();
+
+ assertFalse(breaker.tryAcquireHalfOpenProbe(KEY));
+ elapseWaitWindow(breaker);
+ assertTrue(breaker.tryAcquireHalfOpenProbe(KEY));
+ breaker.reset(KEY);
+ }
+
+ @Test
+ public void testDefaultOutcomeRecordingIgnoresCancellation() {
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.getInstance();
+ breaker.reset(KEY);
+ breaker.recordFailure(KEY);
+
+ StepVerifier.create(UpstreamCircuitBreaker.recordOutcome(Mono.never(),
KEY))
+ .thenCancel()
+ .verify();
+ breaker.recordFailure(KEY);
+
+ assertTrue(breaker.isRequestAllowed(KEY));
+ breaker.reset(KEY);
+ }
+
+ private void elapseWaitWindow(final UpstreamCircuitBreaker breaker) throws
Exception {
+ Field breakersField =
UpstreamCircuitBreaker.class.getDeclaredField("breakers");
+ breakersField.setAccessible(true);
+ ConcurrentMap<String, ?> breakers = (ConcurrentMap<String, ?>)
breakersField.get(breaker);
+ Object state = breakers.get(KEY);
+ Field openedAtField =
state.getClass().getDeclaredField("openedAtMillis");
+ openedAtField.setAccessible(true);
+ openedAtField.setLong(state, System.currentTimeMillis() - 60000L);
+ }
+
@Test
public void testConcurrentFailuresOpenBreaker() throws
InterruptedException {
final int threads = 32;
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/DividePlugin.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/DividePlugin.java
index ef8deefe99..29df701bed 100644
---
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/DividePlugin.java
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/DividePlugin.java
@@ -99,18 +99,19 @@ public class DividePlugin extends AbstractShenyuPlugin {
Object error = ShenyuResultWrap.error(exchange,
ShenyuResultEnum.CANNOT_FIND_HEALTHY_UPSTREAM_URL);
return WebFluxResultUtils.result(exchange, error);
}
- final List<Upstream> breakeredUpstreamList =
filterByCircuitBreaker(selector.getId(), upstreamList);
+ List<String> specifyDomains =
exchange.getRequest().getHeaders().get(Constants.SPECIFY_DOMAIN);
+ String specifiedDomain = CollectionUtils.isNotEmpty(specifyDomains) ?
specifyDomains.get(0) : null;
+ CircuitBreakerSelection circuitBreakerSelection =
filterByCircuitBreaker(selector.getId(), upstreamList, specifiedDomain);
+ final List<Upstream> breakeredUpstreamList =
circuitBreakerSelection.getAllowedUpstreams();
if (CollectionUtils.isEmpty(breakeredUpstreamList)) {
LOG.warn("all upstreams of selector {} are blocked by the circuit
breaker, fail fast", selector.getId());
Object error = ShenyuResultWrap.error(exchange,
ShenyuResultEnum.CANNOT_FIND_HEALTHY_UPSTREAM_URL);
return WebFluxResultUtils.result(exchange, error);
}
- List<String> specifyDomains =
exchange.getRequest().getHeaders().get(Constants.SPECIFY_DOMAIN);
Upstream upstream;
if (CollectionUtils.isNotEmpty(specifyDomains)) {
- String requested = specifyDomains.get(0);
upstream = breakeredUpstreamList.stream()
- .filter(u -> u.getUrl().equals(requested))
+ .filter(u -> u.getUrl().equals(specifiedDomain))
.findFirst()
.map(u -> Upstream.builder()
.url(u.getUrl())
@@ -139,7 +140,8 @@ public class DividePlugin extends AbstractShenyuPlugin {
exchange.getAttributes().put(Constants.RETRY_STRATEGY,
StringUtils.defaultIfEmpty(ruleHandle.getRetryStrategy(),
RetryEnum.CURRENT.getName()));
exchange.getAttributes().put(Constants.LOAD_BALANCE,
StringUtils.defaultIfEmpty(ruleHandle.getLoadBalance(),
LoadBalanceEnum.RANDOM.getName()));
exchange.getAttributes().put(Constants.DIVIDE_SELECTOR_ID,
selector.getId());
- return forwardWithCircuitBreaker(exchange, chain, ruleHandle,
selector.getId(), upstream);
+ return forwardWithCircuitBreaker(exchange, chain, ruleHandle,
selector.getId(), upstream,
+ circuitBreakerSelection.isProbe(selector.getId(), upstream));
}
@Override
@@ -176,20 +178,22 @@ public class DividePlugin extends AbstractShenyuPlugin {
* @param ruleHandle the divide rule handle
* @param selectorId the selector id
* @param upstream the chosen upstream
+ * @param probeGranted whether this request acquired the half-open probe
* @return {@code Mono<Void>} to indicate when request processing is
complete
*/
private Mono<Void> forwardWithCircuitBreaker(final ServerWebExchange
exchange, final ShenyuPluginChain chain,
- final DivideRuleHandle
ruleHandle, final String selectorId, final Upstream upstream) {
+ final DivideRuleHandle
ruleHandle, final String selectorId, final Upstream upstream,
+ final boolean probeGranted) {
String breakerKey = UpstreamCircuitBreaker.buildKey(selectorId,
upstream);
if (ruleHandle.getLoadBalance().equals(P2C)) {
return UpstreamCircuitBreaker.recordOutcome(chain.execute(exchange)
- .doFinally(signalType -> responseTrigger(upstream)),
breakerKey);
+ .doFinally(signalType -> responseTrigger(upstream)),
breakerKey, probeGranted);
} else if (ruleHandle.getLoadBalance().equals(SHORTEST_RESPONSE)) {
long beginTime = System.currentTimeMillis();
return UpstreamCircuitBreaker.recordOutcome(chain.execute(exchange)
- .doOnSuccess(e -> successResponseTrigger(upstream,
beginTime)), breakerKey);
+ .doOnSuccess(e -> successResponseTrigger(upstream,
beginTime)), breakerKey, probeGranted);
}
- return UpstreamCircuitBreaker.recordOutcome(chain.execute(exchange),
breakerKey);
+ return UpstreamCircuitBreaker.recordOutcome(chain.execute(exchange),
breakerKey, probeGranted);
}
private DivideRuleHandle buildRuleHandle(final RuleData rule) {
@@ -199,30 +203,47 @@ public class DividePlugin extends AbstractShenyuPlugin {
/**
* Filter out upstreams blocked by the built-in circuit breaker.
*
- * <p>When every upstream of the selector is blocked, a single half-open
- * probe request is granted to one of them so the breaker can recover;
- * otherwise the caller should fail fast.
+ * <p>An upstream whose wait window has elapsed is selected alone for its
+ * single half-open probe, even when other upstreams remain available. This
+ * ensures the acquired probe is actually sent instead of being lost in
+ * load balancing. If no probe is ready, closed upstreams remain available.
+ * When a domain is explicitly specified, that upstream is preferred for
+ * probe acquisition. If it is unavailable and no closed upstream is
+ * available, an eligible upstream may acquire a probe as a fallback.
*
* @param selectorId the selector id the upstreams belong to
* @param upstreamList the healthy upstream list from the cache
+ * @param specifiedDomain the explicitly requested upstream URL, if present
* @return the upstreams allowed to receive requests, possibly empty
*/
- private List<Upstream> filterByCircuitBreaker(final String selectorId,
final List<Upstream> upstreamList) {
+ private CircuitBreakerSelection filterByCircuitBreaker(final String
selectorId, final List<Upstream> upstreamList,
+ final String
specifiedDomain) {
List<Upstream> allowed = new ArrayList<>(upstreamList.size());
+ Upstream probe = null;
+ String probeKey = null;
for (Upstream upstream : upstreamList) {
- if
(UpstreamCircuitBreaker.getInstance().isRequestAllowed(UpstreamCircuitBreaker.buildKey(selectorId,
upstream))) {
+ String key = UpstreamCircuitBreaker.buildKey(selectorId, upstream);
+ if (UpstreamCircuitBreaker.getInstance().isRequestAllowed(key)) {
allowed.add(upstream);
+ } else if (Objects.isNull(probe)
+ && (Objects.isNull(specifiedDomain) ||
upstream.getUrl().equals(specifiedDomain))
+ &&
UpstreamCircuitBreaker.getInstance().tryAcquireHalfOpenProbe(key)) {
+ probe = upstream;
+ probeKey = key;
}
}
- if (CollectionUtils.isNotEmpty(allowed)) {
- return allowed;
- }
- for (Upstream upstream : upstreamList) {
- if
(UpstreamCircuitBreaker.getInstance().tryAcquireHalfOpenProbe(UpstreamCircuitBreaker.buildKey(selectorId,
upstream))) {
- return Collections.singletonList(upstream);
+ if (CollectionUtils.isEmpty(allowed) && Objects.isNull(probe) &&
Objects.nonNull(specifiedDomain)) {
+ for (Upstream upstream : upstreamList) {
+ String key = UpstreamCircuitBreaker.buildKey(selectorId,
upstream);
+ if
(UpstreamCircuitBreaker.getInstance().tryAcquireHalfOpenProbe(key)) {
+ probe = upstream;
+ probeKey = key;
+ break;
+ }
}
}
- return Collections.emptyList();
+ List<Upstream> selected = Objects.nonNull(probe) ?
Collections.singletonList(probe) : allowed;
+ return new CircuitBreakerSelection(selected, probeKey);
}
private void responseTrigger(final Upstream upstream) {
@@ -253,4 +274,24 @@ public class DividePlugin extends AbstractShenyuPlugin {
upstream.getSucceeded().incrementAndGet();
}
+ private static final class CircuitBreakerSelection {
+
+ private final List<Upstream> allowedUpstreams;
+
+ private final String probeKey;
+
+ private CircuitBreakerSelection(final List<Upstream> allowedUpstreams,
final String probeKey) {
+ this.allowedUpstreams = allowedUpstreams;
+ this.probeKey = probeKey;
+ }
+
+ private List<Upstream> getAllowedUpstreams() {
+ return allowedUpstreams;
+ }
+
+ private boolean isProbe(final String selectorId, final Upstream
upstream) {
+ return Objects.equals(probeKey,
UpstreamCircuitBreaker.buildKey(selectorId, upstream));
+ }
+ }
+
}
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/test/java/org/apache/shenyu/plugin/divide/DividePluginCircuitBreakerTest.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/test/java/org/apache/shenyu/plugin/divide/DividePluginCircuitBreakerTest.java
index 69e67575e8..1eea3d546d 100644
---
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/test/java/org/apache/shenyu/plugin/divide/DividePluginCircuitBreakerTest.java
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/test/java/org/apache/shenyu/plugin/divide/DividePluginCircuitBreakerTest.java
@@ -47,6 +47,7 @@ import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
@@ -140,14 +141,7 @@ public final class DividePluginCircuitBreakerTest {
@Test
public void testHalfOpenProbeRecoversBreaker() throws Exception {
openBreaker();
- // rewind the open timestamp so the half-open wait window has elapsed
- Field breakersField =
UpstreamCircuitBreaker.class.getDeclaredField("breakers");
- breakersField.setAccessible(true);
- ConcurrentMap<String, ?> breakers = (ConcurrentMap<String, ?>)
breakersField.get(UpstreamCircuitBreaker.getInstance());
- Object state = breakers.get(breakerKey);
- Field openedAtField =
state.getClass().getDeclaredField("openedAtMillis");
- openedAtField.setAccessible(true);
- openedAtField.setLong(state, System.currentTimeMillis() - 60000L);
+ elapseWaitWindow(breakerKey);
// the probe request is forwarded and its success closes the breaker
when(chain.execute(exchange)).thenReturn(Mono.empty());
StepVerifier.create(dividePlugin.doExecute(exchange, chain,
selectorData, ruleData))
@@ -156,6 +150,88 @@ public final class DividePluginCircuitBreakerTest {
assertFalse(UpstreamCircuitBreaker.getInstance().isBlocking(breakerKey));
}
+ @Test
+ public void testHalfOpenProbeIsSelectedWhileAnotherUpstreamIsHealthy()
throws Exception {
+ openBreaker();
+ elapseWaitWindow(breakerKey);
+ Upstream healthy =
Upstream.builder().protocol("http://";).url("127.0.0.1:9091").build();
+ UpstreamCacheManager cacheManager = mock(UpstreamCacheManager.class);
+
when(cacheManager.findUpstreamListBySelectorId(SELECTOR_ID)).thenReturn(List.of(upstream,
healthy));
+
cacheMock.when(UpstreamCacheManager::getInstance).thenReturn(cacheManager);
+ when(chain.execute(exchange)).thenReturn(Mono.empty());
+
+ StepVerifier.create(dividePlugin.doExecute(exchange, chain,
selectorData, ruleData))
+ .verifyComplete();
+
+ assertEquals(upstream.buildDomain(),
exchange.getAttribute(Constants.HTTP_DOMAIN));
+
assertFalse(UpstreamCircuitBreaker.getInstance().isBlocking(breakerKey));
+ verify(chain, times(1)).execute(any());
+ }
+
+ @Test
+ public void testSpecifiedHealthyUpstreamIsNotOverriddenByAnotherProbe()
throws Exception {
+ openBreaker();
+ elapseWaitWindow(breakerKey);
+ Upstream healthy =
Upstream.builder().protocol("http://";).url("127.0.0.1:9091").build();
+ UpstreamCacheManager cacheManager = mock(UpstreamCacheManager.class);
+
when(cacheManager.findUpstreamListBySelectorId(SELECTOR_ID)).thenReturn(List.of(upstream,
healthy));
+
cacheMock.when(UpstreamCacheManager::getInstance).thenReturn(cacheManager);
+ this.exchange =
MockServerWebExchange.from(MockServerHttpRequest.get("http://localhost/test";)
+ .header(Constants.SPECIFY_DOMAIN, healthy.getUrl()).build());
+ exchange.getAttributes().put(Constants.CONTEXT,
mock(ShenyuContext.class));
+ when(chain.execute(exchange)).thenReturn(Mono.empty());
+
+ StepVerifier.create(dividePlugin.doExecute(exchange, chain,
selectorData, ruleData))
+ .verifyComplete();
+
+ assertEquals(healthy.buildDomain(),
exchange.getAttribute(Constants.HTTP_DOMAIN));
+
assertTrue(UpstreamCircuitBreaker.getInstance().isBlocking(breakerKey));
+ verify(chain, times(1)).execute(any());
+ }
+
+ @Test
+ public void testUnavailableSpecifiedUpstreamFallsBackToAvailableProbe()
throws Exception {
+ openBreaker();
+ elapseWaitWindow(breakerKey);
+ this.exchange =
MockServerWebExchange.from(MockServerHttpRequest.get("http://localhost/test";)
+ .header(Constants.SPECIFY_DOMAIN, "127.0.0.1:9091").build());
+ exchange.getAttributes().put(Constants.CONTEXT,
mock(ShenyuContext.class));
+ when(chain.execute(exchange)).thenReturn(Mono.empty());
+
+ StepVerifier.create(dividePlugin.doExecute(exchange, chain,
selectorData, ruleData))
+ .verifyComplete();
+
+ assertEquals(upstream.buildDomain(),
exchange.getAttribute(Constants.HTTP_DOMAIN));
+
assertFalse(UpstreamCircuitBreaker.getInstance().isBlocking(breakerKey));
+ verify(chain, times(1)).execute(any());
+ }
+
+ @Test
+ public void testCancelledHalfOpenProbeReopensBreaker() throws Exception {
+ openBreaker();
+ elapseWaitWindow(breakerKey);
+ when(chain.execute(exchange)).thenReturn(Mono.never());
+
+ StepVerifier.create(dividePlugin.doExecute(exchange, chain,
selectorData, ruleData))
+ .thenCancel()
+ .verify();
+
+
assertFalse(UpstreamCircuitBreaker.getInstance().tryAcquireHalfOpenProbe(breakerKey));
+ elapseWaitWindow(breakerKey);
+
assertTrue(UpstreamCircuitBreaker.getInstance().tryAcquireHalfOpenProbe(breakerKey));
+ }
+
+ @Test
+ public void testCancelledClosedRequestLeavesBreakerClosed() {
+ when(chain.execute(exchange)).thenReturn(Mono.never());
+
+ StepVerifier.create(dividePlugin.doExecute(exchange, chain,
selectorData, ruleData))
+ .thenCancel()
+ .verify();
+
+
assertTrue(UpstreamCircuitBreaker.getInstance().isRequestAllowed(breakerKey));
+ }
+
@Test
public void testHealthyUpstreamsBypassBreaker() {
Upstream healthy =
Upstream.builder().protocol("http://";).url("127.0.0.1:9091").build();
@@ -178,4 +254,14 @@ public final class DividePluginCircuitBreakerTest {
UpstreamCircuitBreaker.getInstance().recordFailure(breakerKey);
}
}
+
+ private void elapseWaitWindow(final String key) throws Exception {
+ Field breakersField =
UpstreamCircuitBreaker.class.getDeclaredField("breakers");
+ breakersField.setAccessible(true);
+ ConcurrentMap<String, ?> breakers = (ConcurrentMap<String, ?>)
breakersField.get(UpstreamCircuitBreaker.getInstance());
+ Object state = breakers.get(key);
+ Field openedAtField =
state.getClass().getDeclaredField("openedAtMillis");
+ openedAtField.setAccessible(true);
+ openedAtField.setLong(state, System.currentTimeMillis() - 60000L);
+ }
}
