Skip to content
Permalink
Browse files
GERONIMO-6585 - Implement the combination of timeout + retry + circui…
…tbreaker.
  • Loading branch information
johnament committed Oct 8, 2017
1 parent 0f811c3 commit b70e077a274be5c426e41bfdb94509dc9911ef55
Showing 5 changed files with 53 additions and 14 deletions.
@@ -53,7 +53,8 @@ public <T> T execute(Callable<T> callable, InvocationContext invocationContext)
return asyncFailsafe.get(callable).get();
} else {
return asyncFailsafe
.get(new TimeoutWrappedCallable<>(callable, executorService, timeout))
.get(new TimeoutWrappedCallable<>(callable, executorService, timeout,
super.failsafeCircuitBreaker))
.get();
}
} catch (CircuitBreakerOpenException e) {
@@ -33,7 +33,7 @@ class MicroprofileAnnotationMapper {
static FailsafeRetryDefinition mapRetry(Retry retry, FailsafeRetryBuilder retryBuilder) {
retryBuilder.withMaxRetries(retry.maxRetries())
.withRetryOn(retry.retryOn())
.withRetryOn(TimeoutException.class)
.withRetryOn(TimeoutException.class, org.eclipse.microprofile.faulttolerance.exceptions.TimeoutException.class)
.withAbortOn(retry.abortOn());
if (retry.delay() > 0L) {
retryBuilder.withDelay(Duration.of(retry.delay(), retry.delayUnit()));
@@ -52,7 +52,7 @@ static FailsafeCircuitBreakerDefinition mapCircuitBreaker(CircuitBreaker circuit
int failureCount = (int) (circuitBreaker.failureRatio() * circuitBreaker.requestVolumeThreshold());
FailsafeCircuitBreakerBuilder failsafeCircuitBreakerBuilder = builder
.withFailOn(circuitBreaker.failOn())
.withFailOn(TimeoutException.class)
.withFailOn(TimeoutException.class, org.eclipse.microprofile.faulttolerance.exceptions.TimeoutException.class)
.withDelay(Duration.of(circuitBreaker.delay(), circuitBreaker.delayUnit()))
.withSuccessCount(circuitBreaker.successThreshold());
if (failureCount > 0) {
@@ -28,11 +28,10 @@

import javax.interceptor.InvocationContext;
import java.util.concurrent.Callable;
import java.util.function.Function;

public class SyncFailsafeExecutionPlan implements ExecutionPlan {
private final FailsafeRetryDefinition retryDefinition;
private final FailsafeCircuitBreaker failsafeCircuitBreaker;
protected final FailsafeRetryDefinition retryDefinition;
protected final FailsafeCircuitBreaker failsafeCircuitBreaker;
private final FallbackRunner fallback;

SyncFailsafeExecutionPlan(FailsafeRetryDefinition retryDefinition, FailsafeCircuitBreaker failsafeCircuitBreaker, FallbackRunner fallback) {
@@ -19,30 +19,70 @@

package org.apache.safeguard.impl.executionPlans;

import org.apache.safeguard.api.circuitbreaker.CircuitBreakerState;
import org.apache.safeguard.impl.circuitbreaker.FailsafeCircuitBreaker;
import org.eclipse.microprofile.faulttolerance.exceptions.CircuitBreakerOpenException;
import org.eclipse.microprofile.faulttolerance.exceptions.TimeoutException;

import java.time.Duration;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

class TimeoutWrappedCallable<T> implements Callable<T> {
private final Callable<T> delegate;
private final ExecutorService executorService;
private final ScheduledExecutorService executorService;
private final Duration timeout;
private final FailsafeCircuitBreaker failsafeCircuitBreaker;
private boolean timedout = false;

TimeoutWrappedCallable(Callable<T> delegate, ExecutorService executorService, Duration timeout) {
TimeoutWrappedCallable(Callable<T> delegate, ScheduledExecutorService executorService, Duration timeout, FailsafeCircuitBreaker failsafeCircuitBreaker) {
this.delegate = delegate;
this.executorService = executorService;
this.timeout = timeout;
this.failsafeCircuitBreaker = failsafeCircuitBreaker;
}

@Override
public T call() throws Exception {
boolean circuitBreakerOpen = failsafeCircuitBreaker != null && failsafeCircuitBreaker.getState() == CircuitBreakerState.OPEN;
if(circuitBreakerOpen) {
throw new CircuitBreakerOpenException();
}
ScheduledFuture<?> scheduledFuture = executorService.schedule(new TimerRunnable(Thread.currentThread(), this),
timeout.toMillis(), TimeUnit.MILLISECONDS);

T result;
try {
return executorService.submit(delegate).get(timeout.toMillis(), TimeUnit.MILLISECONDS);
result = delegate.call();
} catch (Exception e) {
throw e;
} finally {
scheduledFuture.cancel(true);
}
catch (TimeoutException e) {
throw new org.eclipse.microprofile.faulttolerance.exceptions.TimeoutException(e);
if(timedout) {
throw new TimeoutException("Execution timed out after " + timeout);
}
return result;
}

private class TimerRunnable implements Runnable {
private final Thread targetThread;
private final TimeoutWrappedCallable task;

private boolean doInterrupt = true;
private TimerRunnable(Thread targetThread, TimeoutWrappedCallable task) {
this.targetThread = targetThread;
this.task = task;
}

@Override
public void run() {
if(doInterrupt) {
task.timedout = true;
targetThread.interrupt();
}
}
}
}
@@ -67,7 +67,6 @@
<dependency>org.eclipse.microprofile.fault-tolerance:microprofile-fault-tolerance-tck</dependency>
</dependenciesToScan>
<excludes>
<exclude>org.eclipse.microprofile.fault.tolerance.tck.CircuitBreakerRetryTest</exclude>
<exclude>org.eclipse.microprofile.fault.tolerance.tck.ConfigTest</exclude>
<exclude>org.eclipse.microprofile.fault.tolerance.tck.bulkhead.BulkheadAsynchRetryTest</exclude>
<exclude>org.eclipse.microprofile.fault.tolerance.tck.bulkhead.BulkheadAsynchTest</exclude>

0 comments on commit b70e077

Please sign in to comment.