How to Stop Application When Streams Goes into Error State?

Kafka Streams
1

Hi,
I have a Kafka stream java application. I used Kafka streams in a spring boot application. I want to stop the application when the stream goes into an error state. I used StreamsUncaughtExceptionHandler but streams go error state and the application doesn’t stop. I also want to stop the stream to gracefully shutting down. How can we achieve this?

public class StreamsUncaughtExceptionHandlerImpl implements StreamsUncaughtExceptionHandler {

    private final ShovelErrorStream shovelErrorStream;

    @Override
    public StreamThreadExceptionResponse handle(Throwable exception) {
        log.error("Kafka-Streams uncaught exception occurred. Stream will be closed", exception);
        shovelErrorStream.getStreamsList().forEach(KafkaStreams::close);
        Runtime.getRuntime().exit(0);
        return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION;
    }
}
2

Your handler implementation does not look right.

Why are you calling shovelErrorStream.getStreamsList().forEach(KafkaStreams::close);? – Do you have multiple KafkaStreams clients in the same application? Why? You should rather configure a single client with multiple threads. (Also note: calling close() might block).

This call does not sound right at all: Runtime.getRuntime().exit(0); – It will exit the JVM in a very hard way, and won’t give the application a chance to shut down. Btw: exit(0) means normal application shutdown; in case of error you should return a negative number, by default -1.

The return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION; looks correct. It will trigger a shutdown signal to all (remotely) running KafkaStreams instances of this application. Note, there is no guarantee that the signal will go through though. In your case, the signal is most likely not sent, because you call exit(0) and return won’t most likely even be executed.

3

Do you have multiple KafkaStreams clients in the same application → No, actually I have only one KafkaStreams client. The list has one item. I’ll change it list to one object.

You’re alright I should use exit(-1)

If I remove Runtime.getRuntime().exit(-1) streams closed but the spring application continues running. I want to stop the application when the stream goes into an error state.

4

If you have only once client, and you return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION you should be fine. You should call close() not inside the handler anyway. (cf below)

If I remove Runtime.getRuntime().exit(-1) streams closed but the spring application continues running. I want to stop the application when the stream goes into an error state.

You should do this “outside” of any handlers via a shutdown hook (cf. Writing a Streams Application | Confluent Documentation).