refactor: Refactored core Kafka Configs integrating DeadLetter strategy

core/src/main/java/com/f_lab/joyeuse_planete/core/kafka/config/KafkaConsumerConfig.java

@@ -1,53 +1,75 @@
 package com.f_lab.joyeuse_planete.core.kafka.config;
 
+import com.f_lab.joyeuse_planete.core.exceptions.JoyeusePlaneteApplicationException;
 import com.f_lab.joyeuse_planete.core.kafka.exceptions.NonRetryableException;
 import com.f_lab.joyeuse_planete.core.kafka.exceptions.RetryableException;
+import com.f_lab.joyeuse_planete.core.kafka.util.ExceptionUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.TopicPartition;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
 import org.springframework.kafka.listener.DefaultErrorHandler;
-import org.springframework.util.backoff.FixedBackOff;
+import org.springframework.kafka.support.ExponentialBackOffWithMaxRetries;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.util.backoff.BackOff;
 
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
 import java.util.Map;
+import java.util.function.BiFunction;
 
+@Slf4j
 public abstract class KafkaConsumerConfig {
 
-  @Value("${spring.kafka.bootstrap-servers}")
+  @Value("${spring.kafka.bootstrap-servers:localhost:9092}")
   protected String BOOTSTRAP_SERVERS;
 
-  @Value("${spring.kafka.consumer.enable-auto-commit}")
+  @Value("${spring.kafka.consumer.enable-auto-commit:false}")
   protected boolean AUTO_COMMIT;
 
-  @Value("${kafka.container.concurrency}")
+  @Value("${kafka.container.concurrency:3}")
   protected int CONCURRENCY;
 
-  @Value("${spring.kafka.consumer.properties.spring.json.trusted.packages}")
+  @Value("${spring.kafka.consumer.properties.spring.json.trusted.packages:com.f_lab.joyeuse_planete.core.*}")
   protected String TRUSTED_PACKAGES;
 
-  @Value("${spring.kafka.consumer.isolation-level}")
+  @Value("${spring.kafka.consumer.isolation-level:read_committed}")
   protected String ISOLATION_LEVEL;
 
-
   abstract public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory();
-
   abstract protected Map<String, Object> consumerConfig();
+  abstract protected String deadLetterTopicName();
+  protected BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> deadLetterTopicStrategy() {
+    return defaultDeadLetterTopicStrategy(deadLetterTopicName());
+  }
+  protected BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> defaultDeadLetterTopicStrategy(String deadLetterTopic) {
+    return (record, ex) -> {
+      ex = ExceptionUtil.unwrap(ex);
+      record.headers().add(KafkaHeaders.EXCEPTION_MESSAGE, ex.getMessage().getBytes(StandardCharsets.UTF_8));
+      record.headers().add(KafkaHeaders.ORIGINAL_TOPIC, record.topic().getBytes(StandardCharsets.UTF_8));
+
+      return new TopicPartition(deadLetterTopic, -1);
+    };
+  }
 
   protected DeadLetterPublishingRecoverer deadLetterPublishingRecoverer() {
     return null;
   }
 
-  public FixedBackOff defaultBackOffStrategy() {
-    return new FixedBackOff(1500L, 10);
+  public BackOff defaultBackOffStrategy() {
+    return new ExponentialBackOffWithMaxRetries(5);
   }
 
   public DefaultErrorHandler defaultErrorHandler() {
     DefaultErrorHandler errorHandler = new DefaultErrorHandler(
         deadLetterPublishingRecoverer(),
         defaultBackOffStrategy());
 
-    errorHandler.addNotRetryableExceptions(NonRetryableException.class);
+    errorHandler.addNotRetryableExceptions(NonRetryableException.class, JoyeusePlaneteApplicationException.class);
     errorHandler.addRetryableExceptions(RetryableException.class);
 
     return errorHandler;

core/src/main/java/com/f_lab/joyeuse_planete/core/kafka/config/KafkaProducerConfig.java

@@ -12,7 +12,7 @@
 
 public abstract class KafkaProducerConfig {
 
-  @Value("${spring.kafka.bootstrap-servers}")
+  @Value("${spring.kafka.bootstrap-servers:localhost:9092}")
   protected String BOOTSTRAP_SERVERS;
 
   @Value("${spring.kafka.producer.ack:all}")

core/src/main/resources/application.yml

@@ -1,6 +1,6 @@
 spring:
   kafka:
-    bootstrap-servers: localhost:9094
+    bootstrap-servers: localhost:9092
     listener:
       ack-mode: manual
     producer:

foods/src/main/resources/application.yml

@@ -4,19 +4,26 @@ spring:
   application:
     name: foods
 
-  datasource:
-    url: jdbc:h2:tcp://localhost/~/Desktop/software/h2/h2-datafile/kafka
-    username: sa
-    password:
-    driver-class-name: org.h2.Driver
+#  datasource:
+#    url: jdbc:h2:tcp://localhost/~/Desktop/software/h2/h2-datafile/kafka
+#    username: sa
+#    password:
+#    driver-class-name: org.h2.Driver
   jpa:
-    database-platform: org.hibernate.dialect.H2Dialect
-    hibernate:
-      ddl-auto: validate
+#    database-platform: org.hibernate.dialect.H2Dialect
+#    hibernate:
+#      ddl-auto: validate
     properties:
       hibernate:
         show_sql: true
 
+  kafka:
+    producer:
+      transaction-id-prefix: foods-tx
+
+    consumer:
+      group-id: foods
+
 orders:
   events:
     topic:
@@ -28,10 +35,8 @@ foods:
       name: foods.order-created-event
       fail: foods.foods-reservation-fail-event
 
-payment:
-  events:
-    topic:
-      name: payments.payment-process-command
+  dead-letter-topic:
+    name: foods.dead-letter-topic
 
 kafka:
   topic:
@@ -40,9 +45,14 @@ kafka:
   container:
     concurrency: 3
 
+non-retryable-errors:
+  - "상품이 존재하지 않습니다."
+  - "상품의 수량이 부족합니다"
+  - "현재 너무 많은 요청을 처리하고 있습니다. 다시 시도해주세요 (락)"
+
 logging:
   level:
     org.hibernate.sql: TRACE
-
-#    org.springframework.kafka: DEBUG
-#    org.apache.kafka: DEBUG
+#
+#    org.springframework.kafka: TRACE
+#    org.apache.kafka: TRACE

orders/src/main/resources/application.yml

@@ -3,25 +3,21 @@ server:
 
 spring:
   kafka:
-    bootstrap-servers: localhost:9094
     producer:
-      acks: all
-      enable:
-        idempotence: true
+      transaction-id-prefix: orders-tx
 
     consumer:
       group-id: orders
-      enable-auto-commit: false
-      properties:
-        spring.json.trusted.packages: com.f_lab.joyeuse_planete.core.*
-      isolation-level: read_committed
 
 orders:
   events:
     topic:
       name: orders.order-created-event
       fail: orders.orders-creation-failed-event
 
+  dead-letter-topic:
+    name: orders.dead-letter-topic
+
 kafka:
   topic:
     partitions: 3
@@ -34,7 +30,8 @@ logging:
   level:
     org:
       apache:
-        kafka: error
+#        kafka: TRACE
         orm:
           jpa:
-            JpaTransactionManager: trace
+            JpaTransactionManager: TRACE
+      hibernate.sql: TRACE