I am getting the above exception when executing a wrapped policy including: retry, circuit breaker and bulk head.
I have the following policies:
var sharedBulkhead = Policy.BulkheadAsync(
maxParallelization: maxParallelizations,
maxQueuingActions: maxQueuingActions,
onBulkheadRejectedAsync: (context) =>
{
Log.Info($"Bulk head rejected => Policy Wrap: {context.PolicyWrapKey}, Policy: {context.PolicyKey}, Endpoint: {context.OperationKey}");
return TaskHelper.EmptyTask;
}
);
var retryPolicy = Policy.Handle<Exception>(e => (e is HttpRequestException)).WaitAndRetryAsync(
retryCount: maxRetryCount,
sleepDurationProvider: attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)),
onRetryAsync: (exception, calculatedWaitDuration, retryCount, context) =>
{
Log.Error($"Retry => Count: {retryCount}, Wait duration: {calculatedWaitDuration}, Policy Wrap: {context.PolicyWrapKey}, Policy: {context.PolicyKey}, Endpoint: {context.OperationKey}, Exception: {exception}.");
return TaskHelper.EmptyTask;
});
var circuitBreaker = Policy.Handle<Exception>(e => (e is HttpRequestException)).CircuitBreakerAsync(
exceptionsAllowedBeforeBreaking: maxExceptionsBeforeBreaking,
durationOfBreak: TimeSpan.FromSeconds(circuitBreakDurationSeconds),
onBreak: (exception, timespan, context) =>
{
Log.Error($"Circuit broken => Policy Wrap: {context.PolicyWrapKey}, Policy: {context.PolicyKey}, Endpoint: {context.OperationKey}, Exception: {exception}");
},
onReset: (context) =>
{
Log.Info($"Circuit reset => Policy Wrap: {context.PolicyWrapKey}, Policy: {context.PolicyKey}, Endpoint: {context.OperationKey}");
}
);
var fallbackForCircuitBreaker = Policy<bool>
.Handle<BrokenCircuitException>()
.FallbackAsync(
fallbackValue: false,
onFallbackAsync: (b, context) =>
{
Log.Error($"Operation attempted on broken circuit => Policy Wrap: {context.PolicyWrapKey}, Policy: {context.PolicyKey}, Endpoint: {context.OperationKey}");
return TaskHelper.EmptyTask;
}
);
var fallbackForAnyException = Policy<bool>
.Handle<Exception>()
.FallbackAsync(
fallbackAction: (ct, context) => { return Task.FromResult(false); },
onFallbackAsync: (e, context) =>
{
Log.Error($"An unexpected error occured => Policy Wrap: {context.PolicyWrapKey}, Policy: {context.PolicyKey}, Endpoint: {context.OperationKey}");
return TaskHelper.EmptyTask;
}
);
var resilienceStrategy = Policy.WrapAsync(retryPolicy, circuitBreaker, sharedBulkhead);
var policyWrap = fallbackForAnyException.WrapAsync(fallbackForCircuitBreaker.Wrap(resilienceStrategy));
I execute the policy like so:
Task.Run(() =>
{
foreach (var changeMessage in changeMessages)
{
policyWrap.ExecuteAsync((context) => CallApi(changeMessage), new Context(endPoint));
}
});
This is producing the said exception: "Please use asynchronous-defined policies when calling asynchronous ExecuteAsync (and similar) methods." inside the fallbackForAnyException
. What am I doing wrong?
fallbackForCircuitBreaker
. I want the retries to happen as per the wait duration regardless of the state the circuit is in. Why doesn't this work? – Observable