spring-cloud-stream: Error Handling Clarity and Intended Workflows

Describe the issue Missing samples or documentation on intended workflows when it comes to error handling and best practices deeper in streams workflows.

Practically all of the documentation around error handling refers to serialization or deserialization errors and handling exceptions that occur at these integration points.

However, any exceptions that are thrown from Transformers, Processors and KeyValueMappers, etc. methods result in Kafka consumers being destroyed.

There are some stack overflow questions around error handling, but the recommendations seem far from ideal.

For example, given the following workflow:

Consumer<KStream<String, SomeObject>>() {
    return input -> input.filter(someInitialFilter)
                    .transform(someExceptionThrowingTransformer)
                    .process(someExceptionThrowingProcessor);
}

The recommendation seems to be to add a try-catch inside the transformer and processor, but returning a value becomes difficult to do cleanly.

One can choose to return null, and then add a filter for a null value. Or a specially configured object, which is then checked and filtered out, OR the processor needs to account for a success or error case from the transformer. We’ve also seen a branching approach with a global error object. Nearly every one of our transform and process methods have risk of throwing an exception, so these strategies are all very intrusive.

Each of these methods are clunky in my team’s opinion, and in my company’s usage, we generally want to send the result to a DLQ or throw it away entirely (along with logging).

We’ve seen recommendations to perform actions that could result in exceptions in filters. But then we are unable to return values when making a “transformation” from a rest call for example.

Are we using these components incorrectly?

The request and use-case for this issue is the following:

Escape hatch handler that can be overridden for any given stream component; Filter/Transformer/Processer/KeyValueMapper/etc. If the exception is swallowed and not re-thrown, the message is removed the same as if it was filtered out.

The idea is to enable developers to encapsulate a transformation for example, and it’s error handling outside the stream configuration.

Existing repository that contains some of the newer capabilities documented, but do not address mid-stream errors.

https://github.com/StevenPG/scs-experimentation/blob/main/scs4-error-handling/error-handling/src/main/java/com/stevenpg/errorhandling/ErrorHandlingApplication.java