Trong bài viết trước, chúng ta đã tìm hiểu chi tiết 2 loại exception handlers trong Spring for Apache Kafka Streams là DeserializationExceptionHandler và ProcessingExceptionHandler. Trong bài viết này, chúng ta hãy cùng nhau tìm hiểu exception handler còn lại là ProductionExceptionHandler, các bạn nhé!
Như mình đã nói ở bài viết trước, ProductionExceptionHandler sẽ handle lỗi xảy ra khi ứng dụng produce các output messages.
Cho interface ProductionExceptionHandler thì Apache Kafka Streams chỉ hỗ trợ 1 implementation là DefaultProductionExceptionHandler, còn Spring for Apache Kafka hỗ trợ thêm class RecoveringProductionExceptionHandler nha các bạn!
Ví dụ mình cũng có một stream topology như sau:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 |
package com.huongdanjava.springkafka; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Produced; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.support.serializer.JacksonJsonSerde; @Configuration public class UserStreamTopology { private static final Logger logger = LoggerFactory.getLogger(UserStreamTopology.class); @Bean public KStream<String, User> users(StreamsBuilder builder) { KStream<String, User> stream = builder.stream("users"); stream.to("output", Produced.with(Serdes.String(), new JacksonJsonSerde<>(User.class))); return stream; } } |
Cho stream topology này thì sau khi nhận message từ topic users, mình sẽ publish một message vào topic output. Message này sẽ được serialize từ một đối tượng của class User, các bạn nhé! Nếu message mà chúng ta publish vào topic users không đúng format của class User, các bạn sẽ thấy lỗi và ứng dụng stop như sau:
Mặc định thì Apache Kafka Stream sử dụng class DefaultProductionExceptionHandlerđể handle cho phần cấu hình PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG này.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 |
package com.huongdanjava.springkafka; import java.util.HashMap; import java.util.Map; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.errors.DefaultProductionExceptionHandler; import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler; import org.apache.kafka.streams.errors.LogAndContinueProcessingExceptionHandler; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.annotation.EnableKafkaStreams; import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration; import org.springframework.kafka.config.KafkaStreamsConfiguration; @Configuration @EnableKafkaStreams public class AppConfig { @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME) public KafkaStreamsConfiguration kafkaStreamsConfiguration() { Map<String, Object> props = new HashMap(); props.put(StreamsConfig.APPLICATION_ID_CONFIG, "spring-kafka-streams-example"); props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); props.put( StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class.getName()); props.put( StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueProcessingExceptionHandler.class.getName()); props.put( StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, DefaultProductionExceptionHandler.class.getName()); return new KafkaStreamsConfiguration(props); } } |
Và với lỗi này, thường ứng dụng của chúng ta sẽ không recover được, nên thường ứng dụng của chúng ta sẽ stop.
Nếu các bạn vẫn muốn ứng dụng chạy thì có thể định nghĩa một bean của class StreamsBuilderFactoryBeanConfigurer như sau các bạn nhé:
|
1 2 3 4 5 6 7 8 |
@Bean public StreamsBuilderFactoryBeanConfigurer streamsCustomizer() { return factoryBean -> factoryBean.setStreamsUncaughtExceptionHandler( exception -> { return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD; }); } |
Cấu hình này catch tất cả uncaught exceptions, không chỉ cho ProductionExceptionHandler mà còn nhiều loại exception khác và replace thread bị lỗi với thread mới để ứng dụng có thể chạy tiếp đó các bạn.
Có 2 cấu hình khác liên quan đến class StreamThreadExceptionResponse là SHUTDOWN_CLIENT và SHUTDOWN_APPLICATION. SHUTDOWN_CLIENT sẽ stop Kafka Stream instance đang consume các message. Chúng ta có thể có nhiều Kafka Stream instance đang consume các message, SHUTDOWN_CLIENT sẽ stop Kafka Stream instance hiện tại đang có lỗi. SHUTDOWN_APPLICATION sẽ stop toàn bộ application.
Các bạn có thể xem đầy đủ các exception handler trong video ở đây:
