Kafka ServletRequestAttributes NullPointer Exception

프로젝트에서 Kafka를 추가적으로 도입해서 상품의 재고를 수정하려던 와중 로직 구현을 완료하고 테스트를 진행하니
Cannot invoke "org.springframework.web.context.request.ServletRequestAttributes.getRequest()" because the return value of "org.springframework.web.context.request.RequestContextHolder.getRequestAttributes()" is null
위의 로그를 보이며, NullPointerException이 발생

추정 원인 1

• ServletRequestAttributes.getRequest()에서 발생하는 에러길래 생각해보다가 수정할때 updatedBy에 들어갈 Auditor 정보가 필요한데 kafka를 사용하는 로직에서 헤더에 해당 부분이 추가가 안되어있어 에러가 발생하는거로 추정되었음

시도 1

• 재고 수정 토픽 쪽으로 kafka message를 보낼 때 헤더에 X-USER-ID(Auditor를 위한 정보)를 추가

    Message<String> kafkaMessage = MessageBuilder
        .withPayload(EventSerializer.serialize(event)) // 직렬화
        .setHeader(KafkaHeaders.TOPIC, topic) // target Topic
        .setHeader("X-USER-ID", userId) // Auditor
        .build();
    kafkaTemplate.send(kafkaMessage);

시도 1 결과

• 동일한 에러 발생

---

추정 원인 2

• KafkaListener가 일반적인 HTTP 연결을 하는게 아니라서 ServletRequest를 사용하지 못하기때문에 Auditor 설정을 못해서 에러 발생하는 것으로 추정
• 우리의 친구 4o에 문의한 결과

  네, Kafka Listener와 같은 비동기 환경에서 HttpServletRequest를 사용할 수 없기 때문에, 
  RequestContextHolder를 사용하는 부분에서 오류가 발생하는 것입니다. 
  JPA의 AuditorAware는 보통 HTTP 요청과 관련된 정보를 통해 사용자 ID를 가져와서 감사 로그를 기록하는 데 사용되지만, 
  비동기 작업에서는 이를 직접 처리할 수 없습니다.

시도 2

• AuditorAwareImpl 기존 로직

  HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();

• ThreadLocal을 사용해 HttpRequest가 없을 경우를 위해 Setter로 Auditor를 주입하는 로직을 추가

  public class AuditorAwareImpl implements AuditorAware<Long> {
  
      // ThreadLocal을 사용해 비동기 환경에서 사용자 ID를 설정
      private static final ThreadLocal<String> auditor = new ThreadLocal<>();
  
      @Override
      public Optional<Long> getCurrentAuditor() {
          String updatedBy;
  
          ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
  
          if (attributes != null) {
              HttpServletRequest request = attributes.getRequest();
              updatedBy = request.getHeader("X-USER-ID");
  
              if (updatedBy == null) {
                  updatedBy = "-1"; // 기본 사용자 ID 설정
              }
          } else {
              // HTTP 요청이 없는 경우 (비동기 작업 등), ThreadLocal에서 사용자 ID를 가져옴
              updatedBy = auditor.get();
              if (updatedBy == null) {
                  updatedBy = "-1"; // 기본 사용자 ID 설정
              }
          }
  
          return Optional.of(Long.parseLong(updatedBy));
      }
  
      public static void setAuditor(String userId) {
          auditor.set(userId);
      }
  
      public static void clear() {
          auditor.remove();
      }
  
  }

시도 결과 2

• 에러 없이 정상 작동 완료

---

P.S.

• kafka 설정만 고려하다보니 Auditor부분을 미처 생각하지 못해서, 에러가 발생했을 때 원인을 한참 고민하게 되었음.
• 디버깅 모드로 한줄씩 확인하다가 Commit에서 딱 실패하는 부분을 보고 아! Auditor쪽 문제인가? 싶어서 위의 시도들을 하게 되었고, 무사히 해결하게 되었음