CDI and fallback handling: Microprofile…​or not

With the microservices trend you have to face the need to handle fallback logic more and more often. Concretely it means that under some circumstances you will use an alternative implementation for a particular piece of your application.

To illustrate that we will use that service:

@ApplicationScoped
public class DefaultService {
    public String findRemoteData(final long id) {
        if (id > 1000) {
            throw new IllegalStateException("this simulates a failure (#" + id + ")");
        }
        return "remote success :)";
    }
}

This service is pretty simple: it returns a constant and if the parameter is higher than 1000 it will throw an exception. This last case will simulate a remote failure for the context of this post.

What we want to achieve is to replace the exception but an alternative implementation in case of an error, for instance:

public String fallbackData(final long id) {
    return "local success :/";
}

In this case the two following assertions would be true:

assert service.findRemoteData(1).equals("remote success :)")
assert service.findRemoteData(1001).equals("local success :/")

The most trivial implementation is to replace the throw by a call to the fallback logic but this is rarely what you want because the fallback logic should be pluggable depending the instance or environment and decoupled from the main service - otherwise it is part of the main service. This simple rule will ensure your code stay maintenable and simple to test and update.

The Microprofile way

Microprofile Fault Tolerance provides a way to do it. It looks like:

@ApplicationScoped
public class DefaultService {
    @Fallback(fallbackMethod = "fallbackData")
    public String findRemoteData(final long id) {
        if (id > 1000) {
            throw new IllegalStateException("this simulates a failure (#" + id + ")");
        }
        return "remote success :)";
    }

    public String fallbackData(final long id) {
        return "local success :/";
    }
}

It is exactly the previous code but we added a @Fallback annotation on our entry method which references the fallback method. This is not bad but this doesn't respect the decoupling of both methods.

Still using microprofile, you can decouple the fallback implementation using value instead of fallbackMethod in @Fallback:

@Fallback(DataFallbackHandler.class)
public String findRemoteData(final long id) {
    // same as before
}

And the fallback implementation:

@ApplicationScoped
public class DataFallbackHandler implements FallbackHandler<String> {
    @Override
    public String handle(final ExecutionContext context) {
        return "local success :/ (#" + context.getParameters()[0] + ")";
    }
}

It is nice because you have access to the parameters and method so you can implement easily some generic handlers. However you still couple the default implementation to its fallback and changing that requires to write a CDI extension changing the annotated type to change the actual service model. Doable but not user friendly.

To go one step further you need to drop this part of the microprofile - don't drop it all since you will likely want to combine the coming implementation to the microprofile retry/timeout/bulkhead logics.

The custom but loosely coupled way

The alternative is to define an annotation to mark a method as being an implementation of a service. Let's use this one as an example:

@Target(METHOD)
@Retention(RUNTIME)
public @interface ServiceDefinition {
    String name() default "";
    int order() default 0;
}

Then we add it on our main implementation:

@ServiceDefinition(name = "DefaultService.findRemoteData")
public String findRemoteData(final long id) {
    if (id > 1000) {
        throw new IllegalStateException("this simulates a failure (#" + id + ")");
    }
    return "remote success :)";
}

And our fallback extracted in another bean:

@ApplicationScoped
public class FallbackService {
    @ServiceDefinition(name = "DefaultService.findRemoteData", order = -1)
    public String fallbackFindRemoteData(final long id) {
        return "local success :/";
    }
}

This looks close to the microprofile handler implementation but the main difference is the relationship is handled based on a name and not hardcoded between both beans.

Note: you will also note that the default order being 0 you can use as convention that 0 is the default, negative values define the fallbacks and positive ones the overrides.

But how to make it working? The overall idea is make the @ServiceDefinition an interceptor binding:

@Target({METHOD, TYPE})
@Retention(RUNTIME)
@InterceptorBinding
public @interface ServiceDefinition {
    @Nonbinding
    String name() default "";

    @Nonbinding
    int order() default 0;
}

We just added @InterceptorBinding to the annotation, made it settable on a TYPE and made the attributes non binding. An alternative is to use another internal annotation in an extension but this is too long to explain in the context of this post.

Then we need an extension capturing all the services checking if they have @ServiceDefinition on some methods and once the container is started, the extension sorts the services respecting their order:

public class FallbackExtension implements Extension {
    private final Map<String, List<MethodInfo>> services = new HashMap<>();

    public List<MethodInfo> getService(final String name) {
        return services.get(name);
    }

    <T> void captureServices(@Observes final ProcessBean<T> bean,
                             final BeanManager beanManager) {
        final Annotated annotated = bean.getAnnotated();
        if (AnnotatedType.class.isInstance(annotated) && beanManager.isNormalScope(bean.getBean().getScope())) {
            filterServiceMethods((AnnotatedType<T>) annotated)
                    .forEach(m -> {
                        final ServiceDefinition definition = m.getAnnotation(ServiceDefinition.class);
                        services.computeIfAbsent(definition.name(), n -> new ArrayList<>())
                                .add(new MethodInfo(definition.order(), m, bean.getBean()));
                    });
        } // else not handled
    }

    void prepareRuntime(@Observes final AfterDeploymentValidation afterDeploymentValidation,
                        final BeanManager beanManager) {
        services.values().forEach(l -> {
            final Comparator<MethodInfo> comparing = comparing(m -> m.order);
            l.sort(comparing.reversed());
            // we only use normal scoped instances so we can look them up eagerly
            l.forEach(mi -> mi.instance = beanManager.getReference(
                    mi.bean, mi.bean.getBeanClass(), beanManager.createCreationalContext(null)));
        });
    }

    private <T> Stream<AnnotatedMethod<? super T>> filterServiceMethods(final AnnotatedType<T> annotatedType) {
        return annotatedType.getMethods().stream()
                .filter(m -> m.isAnnotationPresent(ServiceDefinition.class));
    }

    static final class MethodInfo implements Function<Object[], Object> {
        private final int order;
        private final AnnotatedMethod<?> method;
        private Bean<?> bean;
        private Object instance;

        private MethodInfo(final int order,
                           final AnnotatedMethod<?> method,
                           final Bean<?> bean) {
            this.order = order;
            this.method = method;
            this.bean = bean;
        }

        @Override
        public Object apply(final Object[] parameters) {
            try {
                return method.getJavaMember().invoke(instance, parameters);
            } catch (final IllegalAccessException e) {
                throw new ContinueException(e);
            } catch (final InvocationTargetException e) {
                throw new ContinueException(e.getTargetException());
            }
        }
    }
}

Don't forget to register the extension in META-INF/services/javax.enterprise.inject.spi.Extension. captureServices will create the registry of services with the N implementations available and prepareRuntime will sort the implementations and lookup the bean instances for the runtime - we assumed in this implementation all beans are normal scoped so we can do it eagerly. Finally MethodInfo implements an invocation of the service it represents through reflection and the particularity of this implementation is to throw a ContinueException  - which is just a custom RuntimeException - to notify the runtime the call failed.

The last missing part is the interceptor implementation. We will bind it between LIBRARY_BEFORE and APPLICATION to let it be exected after potential security checks but before the actual method implementation since we want to replace it. To make it as late as possible we will therefore use APPLICATION-1.

The first thing the interceptor needs to do is to lookup its service name. To do that we will use a small workaround - but a cleaner - but more complex - implementation would use the name as being binding in the @ServiceDefinition binding. The workaround is to inject the BeanManager and the @Intercepted Bean to find back the called AnnotatedMethod and read the @ServiceDefinition from here:

@Interceptor
@ServiceDefinition
@Priority(Interceptor.Priority.APPLICATION - 1)
public class FallbackInterceptor implements Serializable {
    @Inject
    private BeanManager beanManager;

    @Inject
    @Intercepted
    private Bean<?> bean;

    private volatile String service;

    @AroundInvoke
    public Object invoke(final InvocationContext context) throws Exception {
        if (service == null) {
            synchronized (this) {
                if (service == null) {
                    service = beanManager.createAnnotatedType(bean.getBeanClass()).getMethods().stream()
                            .filter(m -> m.getJavaMember().equals(context.getMethod()))
                            .findAny()
                            .map(m -> m.getAnnotation(ServiceDefinition.class))
                            .get()
                            .name();
                }
            }
        }

        // rest of the impl
    }
}

Now we have the service we can lookup the implementations in the extension and call them in chain until one passes:

@Interceptor
@ServiceDefinition
@Priority(Interceptor.Priority.APPLICATION - 1)
public class FallbackInterceptor implements Serializable {
    @Inject
    private FallbackExtension extension;


    @AroundInvoke
    public Object invoke(final InvocationContext context) throws Exception {
        // service lookup

        final List<FallbackExtension.MethodInfo> impls = extension.getService(this.service);
        if (impls == null) {
            return context.proceed();
        }
        Exception error = null;
        for (final FallbackExtension.MethodInfo mi : impls) {
            try {
                return mi.apply(context.getParameters());
            } catch (final ContinueException ce) {
                if (error == null) {
                    final Throwable cause = ce.getCause();
                    error = Exception.class.isInstance(cause) ? Exception.class.cast(cause) : new IllegalStateException(cause);
                } // else reuse previous one
            }
        }
        throw error;
    }
}

Combining both parts (service name lookup and this delegation chain) doesn't work because the MethodInfo call will go through the interceptor again and will loop. To solve that we will add a state which will prevent this loop. The state will just be a ThreadLocal - we avoid @RequestScoped since there are some contexts where it is not usable. The ThreadLocal will store a set of the services to skip. The state will also provide a way to do an invocation chain (our delegation chain in previous snippet) and handle the state around this invocation (set the service interceptor to be skipped before and set it to be used back after):

@ApplicationScoped
static class State {
    // request scoped would work most of the time but not everywhere so just use a threadLocal
    private final ThreadLocal<Set<String>> state = new ThreadLocal<>();

    boolean getSkip(final String key) {
        final Set<String> value = state.get();
        if (value == null) {
            state.remove();
            return false;
        }
        return value.contains(key);
    }

    Object execute(final String key, final Callable<Object> o) throws Exception {
        Set<String> map = state.get();
        if (map == null) {
            map = new HashSet<>();
            state.set(map);
        }
        map.add(key);
        try {
            return o.call();
        } finally {
            map.remove(key);
            if (map.isEmpty()) {
                state.remove();
            }
        }
    }
}

This state will be injected into the interceptor which will be modified to use it:

final boolean doSkip = state.getSkip(service);
if (doSkip) {
    return context.getMethod().invoke(context.getTarget(), context.getParameters());
}

return state.execute(service, () -> {
    final List<FallbackExtension.MethodInfo> impls = extension.getService(this.service);
    if (impls == null) {
        return context.proceed();
    }
    Exception error = null;
    for (final FallbackExtension.MethodInfo mi : impls) {
        try {
            return mi.apply(context.getParameters());
        } catch (final ContinueException ce) {
            if (error == null) {
                final Throwable cause = ce.getCause();
                error = Exception.class.isInstance(cause) ? Exception.class.cast(cause) : new IllegalStateException(cause);
            } // else reuse previous one
        }
    }
    throw error;
});

With this implementation you can inject any of the service implementation - or through an interface if you have one - and the interceptor will handle itself the fallbacks as needed:

public class DefaultServiceTest {
    @ClassRule
    public static final MeecrowaveRule RULE = new MeecrowaveRule();

    @Inject
    private DefaultService service;

    @Test
    public void run() {
        RULE.inject(this);

        assertEquals("remote success :)", service.findRemoteData(1));
        assertEquals("local success :/", service.findRemoteData(1001));
    }
}

In this test - using Apache Meecrowave - we used our DefaultService but our FallbackService is used automatically when the DefaultService fails. Neat, right?

Going further

This first implementation is quite simple and can be enhanced through several vectors:

1. You maybe didn't notice but you now have a service registry (our extension) and therefore can use that to expose the services or register them in a real runtime registry usable by a cluster to distribute the load dynamically! You can also dump this registry at startup to log the fallback chains the application uses (discovered).
2. You can make the interceptor implementation async friendly. Instead of doing a loop you use a CompletionFuture to set the exceptionn handler the next callback call etc...
3. For now we assumed all beans use the same signatures (parameters and return type) but you can easily put in place a "parameter mapper" allowing to support methods with different parameters.
4. All the services must be normal scoped which prevents to use dependent beans for instance. This is easy to solve (you need to lookup the bean at each invocation and release the creationalcontext once called) but requires some changes to previous snippets.
5. The service name lookup in the interceptor can be enhanced.
6. The state should be handled through 2 interceptors: the one we saw where the state will be set and another one bound to PLATFORM_BEFORE where the state will be read only and bypass the interceptor chain completely - it is rarely an issue but depends the interceptors you have in your chain.

However, even if it can be enhanced, it already allows to provide N implementations for the same bean without any coupling between them except the service name used to register the implementations which is not an issue compared to couple the implementations all together.