You have an event streaming application, and you want to make sure that it's robust in the face of unexpected errors. Depending on the situation, you'll want the application to either continue running or shut down. Using an implementation of the StreamsUncaughtExceptionHandler can provide this functionality.
To handle uncaught exceptions use the KafkaStreams.setUncaughtExceptionHandler method:
final StreamsUncaughtExceptionHandler exceptionHandler =
new MaxFailuresUncaughtExceptionHandler(3, 3600000);
kafkaStreams.setUncaughtExceptionHandler(exceptionHandler);
You can also use a lambda instead of a concrete implementation:
kafkaStreams.setUncaughtExceptionHander((exception) -> StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD);
The StreamsUncaughtExceptionHandler interface gives you an opportunity to respond to exceptions not handled by Kafka Streams. It has one method, handle, and it returns an enum of type StreamThreadExceptionResponse which provides you the opportunity to instruct Kafka Streams how to respond to the exception. There are three possible values: REPLACE_THREAD, SHUTDOWN_CLIENT, or SHUTDOWN_APPLICATION.
It's important to note that the exception handler is for errors not related to malformed records as when the error occurs, Kafka Streams will not commit, and when restarting a thread, it will encounter the bad record again.
The following steps use Confluent Cloud. To run the tutorial locally with Docker, skip to the Docker instructions section at the bottom.
git clone git@github.com:confluentinc/tutorials.git
cd tutorials
Login to your Confluent Cloud account:
confluent login --prompt --save
Install a CLI plugin that will streamline the creation of resources in Confluent Cloud:
confluent plugin install confluent-quickstart
Run the plugin from the top-level directory of the tutorials repository to create the Confluent Cloud resources needed for this tutorial. Note that you may specify a different cloud provider (gcp or azure) or region. You can find supported regions in a given cloud provider by running confluent kafka region list --cloud <CLOUD>.
confluent quickstart \
--environment-name kafka-streams-error-handling-env \
--kafka-cluster-name kafka-streams-error-handling-cluster \
--create-kafka-key \
--kafka-java-properties-file ./error-handling/kstreams/src/main/resources/cloud.properties
The plugin should complete in under a minute.
Create the input and output topics for the application:
confluent kafka topic create input_topic
confluent kafka topic create output_topic
Start a console producer:
confluent kafka topic produce input_topic
Enter a few strings:
apples
pears
tomatoes
bread
butter
milk
Enter Ctrl+C to exit the console producer.
Compile the application from the top-level tutorials repository directory:
./gradlew error-handling:kstreams:shadowJar
Navigate into the application's home directory:
cd error-handling/kstreams
Run the application, passing the Kafka client configuration file generated when you created Confluent Cloud resources:
java -cp ./build/libs/error-handling-standalone.jar \
io.confluent.developer.StreamsUncaughtExceptionHandling \
./src/main/resources/cloud.properties
Validate that the application continues to run and that there are uppercase strings in the output topic despite the fact that the application threw RuntimeExceptions.
confluent kafka topic consume output_topic -b
When you are finished, delete the kafka-streams-error-handling-env environment by first getting the environment ID of the form env-123456 corresponding to it:
confluent environment list
Delete the environment, including all resources created for this tutorial:
confluent environment delete <ENVIRONMENT ID>
git clone git@github.com:confluentinc/tutorials.git
cd tutorials
Start Kafka with the following command run from the top-level tutorials repository directory:
docker compose -f ./docker/docker-compose-kafka.yml up -d
Open a shell in the broker container:
docker exec -it broker /bin/bash
Create the input and output topics for the application:
kafka-topics --bootstrap-server localhost:9092 --create --topic input_topic
kafka-topics --bootstrap-server localhost:9092 --create --topic output_topic
Start a console producer:
kafka-console-producer --bootstrap-server localhost:9092 --topic input_topic
Enter a few strings:
apples
pears
tomatoes
bread
butter
milk
Enter Ctrl+C to exit the console producer.
On your local machine, compile the app:
./gradlew error-handling:kstreams:shadowJar
Navigate into the application's home directory:
cd error-handling/kstreams
Run the application, passing the local.properties Kafka client configuration file that points to the broker's bootstrap servers endpoint at localhost:9092:
java -cp ./build/libs/error-handling-standalone.jar \
io.confluent.developer.StreamsUncaughtExceptionHandling \
./src/main/resources/local.properties
Validate that the application continues to run and that there are uppercase strings in the output topic despite the fact that the application threw RuntimeExceptions.
kafka-console-consumer --bootstrap-server localhost:9092 --topic output_topic --from-beginning
From your local machine, stop the broker container:
docker compose -f ./docker/docker-compose-kafka.yml down