update: Updated Kafka Retry AOP

f-lab-edu · Jan 29, 2025 · 7b8d787 · 7b8d787
1 parent bf7bfe6
commit 7b8d787
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 2 deletions.
diff --git a/core/src/main/java/com/f_lab/joyeuse_planete/core/kafka/aspect/KafkaRetryAspect.java b/core/src/main/java/com/f_lab/joyeuse_planete/core/kafka/aspect/KafkaRetryAspect.java
@@ -4,6 +4,7 @@
 import com.f_lab.joyeuse_planete.core.exceptions.ErrorCode;
 import com.f_lab.joyeuse_planete.core.util.log.LogUtil;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.KafkaException;
 import org.aspectj.lang.ProceedingJoinPoint;
 import org.aspectj.lang.annotation.Around;
 import org.aspectj.lang.annotation.Aspect;
@@ -26,7 +27,7 @@ public Object kafkaRetry(ProceedingJoinPoint joinPoint) {
     while (attempts < MAX_RETRY) {
       try {
         return joinPoint.proceed();
-      } catch (Throwable e) {
+      } catch (KafkaException e) {
         LogUtil.retry(++attempts, joinPoint.getSignature().toString());
 
         try {
@@ -35,9 +36,13 @@ public Object kafkaRetry(ProceedingJoinPoint joinPoint) {
           LogUtil.exception(joinPoint.getSignature().toString(), ex);
           throw new JoyeusePlaneteApplicationException(ErrorCode.KAFKA_RETRY_FAIL_EXCEPTION, ex);
         }
+      } catch (Throwable e) {
+        LogUtil.exception(joinPoint.getSignature().toString(), e);
+        throw new JoyeusePlaneteApplicationException(ErrorCode.KAFKA_RETRY_FAIL_EXCEPTION, e);
       }
     }
 
     throw new JoyeusePlaneteApplicationException(ErrorCode.KAFKA_RETRY_FAIL_EXCEPTION);
   }
+
 }
diff --git a/core/src/main/java/com/f_lab/joyeuse_planete/core/kafka/config/KafkaConsumerConfig.java b/core/src/main/java/com/f_lab/joyeuse_planete/core/kafka/config/KafkaConsumerConfig.java
@@ -55,7 +55,7 @@ public abstract class KafkaConsumerConfig {
       LogUtil.exception("KafkaConsumerConfig.defaultDeadLetterTopicStrategy", ex);
 
       record.headers().add(KafkaHeaders.EXCEPTION_FQCN, ex.getClass().getName().getBytes());
-      record.headers().add(KafkaHeaders.EXCEPTION_MESSAGE, ex.getMessage().getBytes());
+      record.headers().add(KafkaHeaders.EXCEPTION_MESSAGE, ((ex.getMessage() != null) ? ex.getMessage() : "null").getBytes());
       record.headers().add(KafkaHeaders.ORIGINAL_TOPIC, record.topic().getBytes());
 
       return new TopicPartition(deadLetterTopic, -1);