Để handle các exception có thể xảy ra trong các ứng dụng có sử dụng Spring for Apache Kafka Streams, các bạn có thể khai báo các cấu hình như sau:
| Configuration | Handles | Ví dụ |
|---|---|---|
DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG |
Handle các exception liên quan đến việc deserialization các message trong consumer | Chuỗi JSON không hợp lệ, sai đối tượng SerDer, … |
PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG |
Lỗi xảy ra khi process các message | Lỗi NullPointerException, ArithmeticException, lỗi xảy ra khi handle business logic trong các phương thức map(), join(), filter(), … |
PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG |
Lỗi xảy ra khi produce các output messages | Serialization bị lỗi, broker reject một message, … |
Cụ thể như thế nào? Chúng ta hãy cùng nhau tìm hiểu trong bài viết này, các bạn nhé!
Đầu tiên, mình sẽ tạo mới một Maven project, tương tự như trong ví dụ của bài viết trước, như sau:
Các dependency liên quan đến Spring for Apache Kafka Streams, mình sẽ khai báo như sau:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>4.1.0</version> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-streams</artifactId> <version>4.3.1</version> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>4.3.1</version> </dependency> <dependency> <groupId>tools.jackson.core</groupId> <artifactId>jackson-databind</artifactId> </dependency> |
Chúng ta cần khai báo dependency management cho các thư viện Jackson để các thư viện này tương thích với nhau, như sau:
|
1 2 3 4 5 6 7 8 9 10 11 |
<dependencyManagement> <dependencies> <dependency> <groupId>tools.jackson</groupId> <artifactId>jackson-bom</artifactId> <version>3.2.1</version> <type>pom</type> <scope>import</scope> </dependency> </dependencies> </dependencyManagement> |
Mình cũng khai báo dependency cho thư viện Slf4J và Logback để chúng ta có thể log tất cả những lỗi xảy ra như sau các bạn nhé:
|
1 2 3 4 5 6 7 8 9 10 11 |
<dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>2.0.18</version> </dependency> <dependency> <groupId>ch.qos.logback</groupId> <artifactId>logback-classic</artifactId> <version>1.5.38</version> </dependency> |
Tập tin cấu hình của thư viện Logback trong thư mục /src/main/resources sẽ có nội dung như sau:
|
1 2 3 4 5 6 7 8 9 10 11 |
<configuration> <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender"> <encoder> <pattern>%d{yyyy-MM-dd HH:mm:ss} [%thread] %-5level %logger{100}.%M:%L - %msg%n</pattern> </encoder> </appender> <root level="warn"> <appender-ref ref="CONSOLE"/> </root> </configuration> |
Các bạn cũng cấu hình thông tin Apache Kafka Server và enable kafka streams như sau các bạn nhé:
|
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 |
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.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()); return new KafkaStreamsConfiguration(props); } } |
Điều đầu tiên mình cần nói với các bạn là giá trị của các cấu hình DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG, PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG lần lượt là các implementation cho các interface là DeserializationExceptionHandler, ProcessingExceptionHandler và ProductionExceptionHandler, các bạn nhé!
DeserializationExceptionHandler
Như mình đã nói ở trên, DeserializationExceptionHandler sẽ handle các exception liên quan đến việc deserialization các message từ producer trong ứng dụng của chúng ta.
Cho phần implementation của interface DeserializationExceptionHandler, mặc định thì thư viện Apache Kafka Streams hỗ trợ 2 implementation là:
LogAndContinueExceptionHandlerLogAndFailExceptionHandler
còn thư viện Spring for Apache Kafka hỗ trợ thêm implementation:
RecoveringDeserializationExceptionHandler
các bạn nhé!
LogAndContinueExceptionHandler sẽ log lỗi deserialization, bỏ qua message bị lỗi và tiếp tục process các message khác, còn LogAndFailExceptionHandler cũng sẽ log lỗi deserialization nhưng sẽ stop application. Implementation RecoveringDeserializationExceptionHandler thì thay vì chỉ log lỗi và bỏ qua message bị lỗi, nó sẽ gửi message đó vào một Dead Letter Queue (DLQ) topic để các bạn có thể recover message này sau.
Ví dụ, mình định nghĩa 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 27 28 |
package com.huongdanjava.springkafka; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.kstream.Consumed; import org.apache.kafka.streams.kstream.KStream; 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", Consumed.with(Serdes.String(), new JacksonJsonSerde<>(User.class))); stream.peek((k, v) -> logger.info(v.toString())); return stream; } } |
Stream topology này sẽ consume topic users với phần serialization/ deserialization cho key là String, value là class User. Nội dung của class User như sau:
|
1 2 3 |
package com.huongdanjava.springkafka; public record User(String name) {} |
Class để chạy ứng dụng:
|
1 2 3 4 5 6 7 8 9 10 |
package com.huongdanjava.springkafka; import org.springframework.context.annotation.AnnotationConfigApplicationContext; public class Application { static void main(){ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(AppConfig.class, UserStreamTopology.class); } } |
Lúc này, nếu chạy ứng dụng và publish vào topic users một message với key là “001”, value là ‘{“name”:”. Các bạn sẽ thấy ứng dụng của chúng ta sẽ log lỗi deserialization và stop, sử dụng class deserialization LogAndFailExceptionHandler, như sau:
Nếu các bạn cấu hình exception handler cho phần deserialization này sử dụng class LogAndContinueExceptionHandler:
|
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 |
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.LogAndContinueExceptionHandler; 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()); return new KafkaStreamsConfiguration(props); } } |
và chạy lại ứng dụng, các bạn sẽ thấy class LogAndContinueExceptionHandler sẽ handle error:
Nó chỉ đơn giản là log lại error và ứng dụng của chúng ta sẽ tiếp tục chạy, các bạn nhé!
ProcessingExceptionHandler
Mặc định thì thư viện Apache Kafka Streams hỗ trợ 2 implementation cho interface ProcessingExceptionHandler là:
LogAndContinueProcessingExceptionHandlerLogAndFailProcessingExceptionHandler
còn thư viện Spring for Apache Kafka hỗ trợ thêm implementation:
RecoveringProcessingExceptionHandler
các bạn nhé!
Tương tự như DeserializationExceptionHandler, LogAndContinueProcessingExceptionHandler sẽ , LogAndFailProcessingExceptionHandler và RecoveringProcessingExceptionHandler, còn RecoveringProcessingExceptionHandler
Ví dụ như mình 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 27 28 29 30 31 32 33 34 |
package com.huongdanjava.springkafka; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.kstream.KStream; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class UserStreamTopology { private static final Logger logger = LoggerFactory.getLogger(UserStreamTopology.class); @Bean public KStream<String, String> users(StreamsBuilder builder) { KStream<String, String> stream = builder.stream("users"); stream .mapValues( value -> { logger.info("Received: {}", value); if (value.equals("ERROR")) { throw new RuntimeException("Something went wrong!"); } return value.toUpperCase(); }) .to("output"); return stream; } } |
Nếu bây giờ, các bạn chạy ứng dụng và publish một message với key là “001”, value là “ERROR”, các bạn sẽ thấy ứng dụng log lỗi sử dụng class LogAndFailProcessingExceptionHandler và stop như sau:
Các bạn có thể thay đổi class exception handler trong trường hợp này sử dụng cấu hình PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG 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 27 28 29 30 31 32 33 34 35 36 |
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.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()); return new KafkaStreamsConfiguration(props); } } |
Chạy lại ví dụ và publish lại message với nội dung như trên, các bạn sẽ thấy ứng dụng của chúng ta cũng sẽ log lỗi nhưng sẽ không stop nữa.
Các bạn có thể đọc tiếp phần 2 tại đây.
Các bạn có thể xem đầy đủ các exception handler trong video ở đây:
