리승자이의 삽질 그리고 정리

회사의 서비스들이 여러개로 쪼개져있다.

그래서 우리는 주로 FeignClient를 사용하는데, 애를 먹었던 로깅레벨에 대해 포스팅한다.

업무중에 삽질을 진행했었으며, 해당 내용으로 자바스럽게 고쳤던 경험을 좀 풀어본다..

스프링 프레임워크를 사용하고 있기에 여기서 같이 제공해주는 Spring Cloud의 OpenFeign을 사용하였다.

아래는 사용하는 예시이다.

해당 예시처럼 나는 FeignClient를 구현했었다.

물론 @EnableFeignClients 는 별도의 Configuration 클래스 파일에 설정을 해주었었다!

여기까지는 일단 기본적인 설정이지만, 아래의 레벨 설명이 진짜다.

Feing Logging Level은 총 4단계로 이루어져있다.

• NONE:  로깅 없음(Default)
• BASIC:  요청 방법 및 URL, 응답 상태 코드 및 실행 시간 기록
• HEADERS:  요청 및 응답 헤더와 함께 기본 정보를 기록
• FULL : 요청과 응답 모두에 대한 헤더, 본문 및 메타데이터를 기록

기본적으로 BASIC, HEADERS에는 정상작동을 했던 우리의 소스였다.

일단 소스가 어떤 구성이냐면...

회사의 정보를 많이 노출할 수는 없기에 간략하게 설명한다.

일명 스크래핑이라고 하는 기술, 흔히 크롤링이라고도 할 수 있을것이다.

해당 작업을 외부 API로 연동하여 응답값을 받아 파싱해주는 작업이 있다.

근데 우리의 규칙상 해당 스크래핑의 원문 데이터라는 아예 크롤링의 전체에 해당하는 xml양식의 데이터도 같이 받아오는 방식을 정책으로 정했기 때문에 우리는 Google Cloud 를 사용하기에 Google Cloud Storage에 원문 데이터를 저장하고, 나머지 데이터를 데이터베이스에 저장하는 구조로 구성이 되어있다.

여기서 문제가 발생한다.

일단 원문 데이터가 포함이되기에 응답값의 길이가 상당히 길다.

길지않고 정상적인 응답을 받는 스크래핑 로직의 경우에는 이런 디버깅이 해당되지 않는다.

근데 글자가 길었고, 내가 핸드폰에서 슬랙으로 알림을 받았던 에러메시지는 다음과 같았다.

아무튼 이러한 예외가 발생했다.

에러 메시지를 보면 해당 응답값의 타입이 application/json이 아니라는데

디버깅을 찍어보면 정상 json응답값이 파싱이 되고 있었다.

아래부분이 좀 더 의심스러웠었는데, buffer Length : 8192 

그러니까... 버퍼길이가 8192 제한인데 길이가 더 커서 담기지 않아서 에러가 발생한다는 것이다.

이 부분이 왜 나왔는지를 보니까

해당 스크래핑 작업을 하는 로직에서 우리는 로그를 찍는다.

스크래핑 작업 전후로 시작했고 완료되었다는 외부의 로직 실행결과를 로그로 담고 있다.

그런데 이 담는 로직에서 예외를 이렇게 뱉는다는 것이다.

feign 의 기본 클라이언트는 어떤것을 쓰냐면 바로 ApacheHttpClient를 사용한다.

아래는 구현체의 내용 일부분이며, 빨간 네모의 asInputStream으로 body를 출력하고 있다.

httpResponse에서 받는 해당 entity의 content를 찍어내는 것인데, 이 부분에서 에러가 발생한다.

기본 maxContentLength 가 8192이기 때문에 정상적으로 스트림 반환을 하지 않고 저렇게 예외를 띄우게 되는 것이다!

이부분이 TCP 연결에서의 최대 길이이기 때문에 HttpClient를 사용하면 이런 예외가 발생된다.

아래는 ByteArrayBody이다.

해당 부분에서는 ByteArrayInputStream을 사용하기 때문에 TCP length 제한보다 더 쓸수 있는 것 같다.

아무든 이 두 asInputStream()을 보는 이유는 아래에서 설명하겠다.

아무튼 이 ByteArrayInputStream은 tcp max content length와 관계없이 스트림을 출력하기에 길이제한이 없다.

이 두개의 차이를 가르는 것은 Log 였다!!!

그래서 소스를 파고들었다.

해당 로그 부분이 BASIC레벨 보다 수준이 낮은 경우 수행되는 로직인데, 

그래서 FULL로직에서만 정상적으로 데이터 파싱이 된다.

왜? - response.toBuilder().body(bodyData).build() 를 통해 새로운 ByteArrayInputStream으로 생성되기 때문이다.

그래서 이 부분을 재정의를 해주어 BASIC 레벨에서도 길이가 아무리 길더라도 데이터를 파싱할 수 있게 설정해주었다.

로직이 정상적으로 수행되며 아래 로그부분에서 객체를 다시 만들어주게 되는걸 보고나서 넘 행복했다는것...😇😇😇

해결방안

해당 Feign의 Logger를 상속받아서 log찍는 부분을 override 시켜주었다.

그래서 기존 로직을 또 건드리고 싶지는 않았다. <- 로그는 이대로 찍어주는게 마음 편했으니까?

그래서 분기처리 하는 부분을 제거하고 정말로 필요없는 부분만을 제거해주고 나머지 로직은 유지했다.

아무튼 데이터가 길었기 때문에 로그레벨에 따라 값이 파싱이 되고 안되고 났던게 좀 신기했다.

길이가 짧은 응답을 사용했다면 오히려 이런것도 모르고 그냥 설정따라 파싱해주고 안해주고로 넘어갔을 것 같다!!!!

개발자 삶의 두번째 컨퍼런스인 NHN Forward에 다녀왔다.

인프콘 때와는 다르게 한 공간에서 쭉 들었다.

백엔드 세션만 들어야겠다 라고 생각했는데 그 세션들이 전부 한곳에 모여있었다 ㅋㅋㅋㅋ

들어가자마자 체크인을 QR로 진행하고 키노트를 진행하는데 계속 서서 컨퍼런스 장소까지 가니 다리가 아팠는데 또 내내 서있었다.

점심먹고 들었던 세션인 분산 시스템에서 데이터를 전달하는 효율적인 방법 세션에서

RDB, RabbitMQ, Kafka를 통한 서로 다른 마이크로 서비스들의 트랜잭션을 어떻게 관리할 것인가에 대해 되게 재밌게 봤다.

RDB로 사용하는 방법을 이번에 처음 보게되어 예전에 면접에서 받았던 질문이 이런걸 물어본 거였구나 하면서 아~ 하고 깨닫는 순간이었다.

메시지 큐를 사용하는 방법은 현재 내가 담당하고 있는 메시지 플랫폼에서의 데드레터 처리와 같았다.

자세하게 정리하는건 내 노션에서 해야겠다.

사람들이 엄청많았고, 아침에 점심 조금 늦게먹고 자리를 안잡았으면 밖에서 그냥 발만 동동 굴렀을 것 같다.

앉지 못하면 자리가 없어서 못들어간다고 밖에서 얘기하시는 스탭분을 봤다. 그래서 다른분들이 참여 못하시는걸 봤었다...

혼자만 뽑히게되어 아는사람 없이 혼자가서 얌전하게 세션들 다 듣고 바로 와버렸다.

중간에 다른 부스인 커리어리에서 앱 다운하면 뭔가 물품을 준다해서 이미 나는 앱이 있어서 저는 이미 앱이 있어요!

하니까 귀하신 회원님이라면서... ㅋㅋㅋ 바로 챙겨주셨다.

아무튼 너무 좋은 시간이었고, 지속적으로 이런 컨퍼런스도 참여하면서 많이 교류하고 정보도 얻어야겠다.

그리고 컨퍼런스를 다녀옴으로써 한번 더 자극을 받았고 다시 한번 열심히 달려봐야겠다!

오랜만에 포스팅하는데 회사에서 그동안 앱 2.0 버전을 출시한다고 이래저래 바빴던 나날을 보냈다.

결과적으로는 만족스러운 출시..? 였던것 같다 ㅋㅋㅋ

버그도 많았고, QA 엔지니어께서 고생을 많이 하셨을 수도 있고 내가 구현한 메시지 플랫폼도 테스트하기가 정말 까다로웠다.

각설하고..

해당 에러사항을 구현하는건 Kafka를 이용하지 않아도 되기 때문에 RestAPI로 구현했다. (+ 테스트코드로만)

모든 코드는 깃허브에 있다.

이번엔 무슨 버그였냐?

이런 에러가 쏟아져나왔다. 기존 레거시 푸시는 NHN Toast를 이용한 푸시서비스로 구성되어 있었다.

변경한다고 해도 과도기가 존재하기 때문에 바로 지울수는 없고 이전 앱을 사용하는 사용자들에게는 해당 푸시로 알림은 계속 받아야되기 때문이다.

왜 났던 에러였는지 더듬어봤더니

Async 설정

기본적으로 @Async 를 사용하기 위해서는 Configuration 클래스를 하나 만들어주고 비동기 실행기에 대한 설정을 해주어야 한다.

이런식으로 에러를 일부러 내기 위해서 큐 사이즈를 5개 기본 스레드 1, 최대 스레드 2, 그리고 비동기로 처리할 큐 사이즈를 5로 구성했다.

문제상황 1

nhn 푸시를 보내기 위해선 Kafka에 Producer를 통해 메세지를 넣어주면, 내가 구현한 메시지 플랫폼에서 해당 메시지를 Consume하여 동작하는 방식으로 구성되어있다.

그런데 나는 비동기로 구현하겠다고만 생각하고 사이즈를 제한해두지는 않았었다.

그래서 Consume은 계속해서 KafkaListener를 통해 무제한으로 하고 있는 와중에 이 컨슘 속도가 너무 빠르니까

위에서 설정한 비동기 처리 큐 사이즈를 200으로 잡아놓아도 예를 들어 1000건이 한방에 producer를 통해 적재된 후 바로 consume을 해버리니까

100개 큐를 훌쩍 넘어버려서도 막는게 아니라 지속적 컨슘이 일어나기에 큐 사이즈를 넘어서는 순간에도 이 메소드 실행을 밀어 넣는 오류였다.

그래서 맨위의 이미지처럼 Exception이 발생하게 되는 것이다.

에러 직접 구현

정확한 코드는 깃허브에서 보면 되지만 여기서는 간략하게 대충 구현하도록 하겠다.

JDK 17버전이기 때문에 record class를 사용함. (record를 사용하면 클래스는 final로 선언되며 equals & hashCode를 재정의하여 갖고있음)

/hello 를 호출하면 printAsync라는 비동기 메소드를 100번호출하게끔 만드는 로직이 있다.

HelloController.java

@RestController
public record HelloController(HelloService helloService) {

    @GetMapping("/hello")
    public ResponseEntity<?> hello() {

        int i = 0;
        while (true) {
            helloService.printAsync();
            i++;

            if (i == 100) {
                break;
            }
        }

        return ResponseEntity.ok(Map.of("data", "success"));
    }
}

HelloService.java

@Slf4j
@Service
public class HelloService {

    @Async("threadPoolTaskExecutor")
    public void printAsync() {
        try {
            Thread.sleep(5000);
            log.info(Thread.currentThread().getName());
            log.info("hello service print!!!");
        } catch (InterruptedException e) {
            log.error("error : {}", e.getMessage());
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

}

printAsync()는 비동기로 호출하며 5초를 기다리고 log를 찍게 되는데, 우리가 처음 전역적으로 설정해준 스레드 설정은

요청이 들어오면 maxQueue 사이즈인 5까지만 담을 수 있게 되어있고 작업이 느려서 큐 데이터를 하나 소진하지 못한다면 에러가 발생하는데

그 에러는 이렇다.

이미지로 한방에 찍어지지 않아 글자로 첨부한다.

Caused by: org.springframework.core.task.TaskRejectedException: Executor [java.util.concurrent.ThreadPoolExecutor@265c1a7c[Running, pool size = 2, active threads = 2, queued tasks = 5, completed tasks = 0]] did not accept task: org.springframework.aop.interceptor.AsyncExecutionInterceptor$$Lambda$967/0x00000008010a7dc0@740a0d5e
    at org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.submit(ThreadPoolTaskExecutor.java:391)
    at org.springframework.aop.interceptor.AsyncExecutionAspectSupport.doSubmit(AsyncExecutionAspectSupport.java:292)
    at org.springframework.aop.interceptor.AsyncExecutionInterceptor.invoke(AsyncExecutionInterceptor.java:129)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
    at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
    at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:708)
    at com.github.lsj8367.application.HelloService$$EnhancerBySpringCGLIB$$7321912.printAsync(<generated>)
    at com.github.lsj8367.presentation.HelloController.hello(HelloController.java:17)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:568)
    at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:205)
    at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:150)
    at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:117)
    at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:895)
    at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:808)
    at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87)
    at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1070)
    at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:963)
    at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1006)
    ... 87 more
Caused by: java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@29013ef2[Not completed, task = org.springframework.aop.interceptor.AsyncExecutionInterceptor$$Lambda$967/0x00000008010a7dc0@740a0d5e] rejected from java.util.concurrent.ThreadPoolExecutor@265c1a7c[Running, pool size = 2, active threads = 2, queued tasks = 5, completed tasks = 0]
    at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2065)
    at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:833)
    at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1365)
    at java.base/java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:145)
    at org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.submit(ThreadPoolTaskExecutor.java:388)
    ... 107 more

처음 봤던 이미지와 같은 예외가 발생하는걸 볼 수 있다.

그러니까 큐 공간이 가득차기 전에 계속해서 비동기를 호출하며 작업큐에 메소드를 태우려고 하다보니 이런 예외가 발생한 것이다.

해결은?

그래서 나는 일단 푸시 발송이 그렇게 오래걸리는 로직이 아니라서 일단은 동기 상태로 바꿔놓았지만,

지속적인 모니터링을 하여 트래픽이 많아지게 되면 그때는 batchListener를 적용하여 비동기로 여러개를 한방에 처리하는 방법을 고려중이다.

-> 이 부분은 카프카의 리밸런싱을 고려해야한다.

아무튼 내가 생각했던 여러 추측중에 하나가 걸리게 되니까 좀 재밌었던 것 같다.

그래도 오늘도 해결했다!

저번 포스팅의 마지막 마무리가 바로 이 에러를 찾지 못했던 것이다.

Kafka 에러를 고치게 된 시점

처음에 특정 Consumer만 consume을 못한다고 했었는데, 이는 당연 잘못된 것이었다!!!

그냥 로그를 좀 더 세세하게 찍고 검토를 더 열심히 했어야 했다!

우선 대략적으로 코드를 보면 아래와 같다.

@KafkaListener(topics = "topic", properties = {
        "spring.kafka.consumer.properties.spring.json.type.mapping=com.github.lsj8367.MessageReq"
    })
public void consumeSomething(final MessageReq req, Acknowledgment acknowledgment) {
    pushService.sendPush(req);
    acknowledgment.acknowledge();
}

일단 회사의 카프카 푸시 로직에서는 자동커밋을 사용하지 않는다.

spring kafka 에서는 enable.auto.commit=true 이라는 설정을 통해 자동으로 일정 시간이 지나면 커밋하게 만들 수 있다.

그 시간은 auto.commit.interval.ms라는 옵션을 통해 설정을 진행할 수 있다.

중복이 생기는 현상

max.poll.records로 15개를 가져온다고 가정한다. (왜냐면 내가 15개씩 가져와서 스레드 15개로 한번에 할당하고 있으니까 15개로 했다)

이제 한번의 컨슘을 통해 이 record들을 15개를 가져왔다.

잘 작업하다가 한 7개 작업을 했다(실질적인 푸시까지 완료) 아직 커밋은 일어나지 않았다. 커밋시간은 조금 더 뒤에 일어나는 상황

여기서

배치 서버에서 파티션 1개로 하니까 너무 느리게 적재되는 것 같아요 병렬처리 하게 파티션을 n개로 바꿀 필요성이 있습니다

라고 하면서 파티션을 늘리는 안이 통과되어 바로 파티션을 늘린다. (참고 - 파티션은 늘릴 수만 있고 줄일 수는 없다)

리밸런싱 시작....

이러면 8개가 아직 작업이 남았지만 컨슘이 멈추며 리밸런싱을 시작한다.

어? 커밋된게 없네 해서 이전 15개를 다시 읽어 재처리를 진행하게 된다.

지금 남은 이 8개는 중복이 아니지만, 앞서 작업했던 7개의 메시지는 다시 작업을 수행하게 된다.

이런식으로 중복이 발생한다고 한다.

그럼 넌 뭐가 문제였는데?

나는 애초에 auto commit이 아닌 수동커밋으로 작업을 진행했다.

푸시지만 중복으로 발생되면 안되는 그런 푸시 정보이기에 수동으로 작업을 진행했다.

일단 Spring kafka의 기본값이 수동 커밋인 enable.auto.commit=false 이다.

그리고 ackMode를 설정해줄 수 있는데

ackMode는 뭐냐면 KafkaListener의 offset에 대한 commit을 어떻게 줄것인가를 설정하는 모드이다.

Spring의 기본 AckMode는 BATCH이며, 나는 manual_immidiate로 작업했다.

둘의 차이는 이미지로도 적어놨지만...

• BATCH
  • poll() 메서드로 호출된 레코드가 모두 처리된 이후 커밋
  • 스프링 카프카 컨슈머의 AckMode 기본값
• MANUAL_IMMEDIATE
  • Acknowledgment가 승인되면 즉시 commit

이 두가지의 차이가 있다.

에러 현상은 아무런 오류도, 로그도 찍어지지 않는데, 카프카 Producer 메세지를 발행하면 LAG로 적재되었다.

특정 토픽은 정상적으로 컨슘이 되는게 확인이 된게 너무 억울했었다.

그래서 100개를 연속으로 발행해도 LAG로 메시지가 쌓여서 처리를 못하는 현상이 발생해서 2주를 머리쥐어짰다. 😅😅😅

다른 토픽으로 받아봐도 결과가 계속 같아서 너무 짜증이 더 났었다.

카프카도 새로 설치해달라고 데브옵스분께 요청했고 그래서 이래저래 너무 힘들었다. (운이 좋았던건 dev환경의 k8s설정이 좀 이상하게 되어 다시 구성해야 되는 상황에 요청드렸던 운이 좋았다😱)

에러난 지점은...

@Slf4j
@Service
public class PushService {

    public void sendPush(final MessageReq req, Acknowledgment acknowledgment) {
        try {
            //푸시로직이 작성되어있다...
            log.info("push Success : {}", req);
        } catch (SendFailureException e) {
            log.error("push failure : {}", e.getMessage());
            //DLT 처리
        }
        acknowledgment.acknowledge();  // 에러지점
    }
}

이런식으로 예시로 푸시로직이 있다고 가정하고 이렇게 작성하면 당연히 푸시를 보내고 ack를 통해 offset을 전진시켜주었다.

이게 초반엔 되게 잘 작동했는데 새로운 예외케이스만 모르고 이대로 문제가 없는줄 알았다.

이 부분에서 문제가 되었던건 SendFailureException이 발생하지 않고 NPE나 다른 예외가 뜨게 되면 이게 메세지 처리가 안된다.

그래서 예외가 났으니 offset을 전진 안시켰기 때문에 다시 메시지에 적재 되는것이다.

일정 시간이 지나고 다시 또 똑같은 메시지를 컨슘하는 이 부분이 무한 반복하는 컨슘이 시작되는 것이다.

심지어 로그도 안찍는 현상이다.

지나가며 보던 블로그의 글들을 보며 commit은 항상 필수적으로 해줘야 한다는게 있었는데 난 당연히 commit을 무조건 하는것으로 생각한 나의 실수였다.

그러면 LAG에 고정으로 쌓여있냐? 그건 아니라는 소리다. 수치로는 n개가 적재된거로 보이겠지만 0으로 갔다가 다시 쌓였다가 할것 같은 나의 생각이다.

해결 방법

정말 간단하다. 위에서 봤을때 문제점이 무엇이었는가?

바로 offset을 무조건 전진시켜주면 끝난다!!

@Slf4j
@Service
public class PushService {

    public void sendPush(final MessageReq req, Acknowledgment acknowledgment) {
        try {
            //푸시로직이 작성되어있다...
            log.info("push Success : {}", req);
        } catch (SendFailureException e) {
            log.error("push failure : {}", e.getMessage());
            //DLT 처리
        } catch (Exception e) {
            log.error("알 수 없는 오류 발생 : {}", e.getMessage());
        }finally {
            acknowledgment.acknowledge();  // 에러지점
        }
    }
}

이렇게 무조건 commit 처리하게 해주고 저런 예외가 터졌을 경우에는 DLT (Dead Letter Topic)에 적재해주고 후속 처리를 진행하면 되겠다.

정리

무턱대고 구현할 때 좀 더 꼼꼼하게 작성하고 좀 더 테스트를 빡세게 하고 테스트 코드도 좀 더 케이스를 많이 짜려고 노력을 좀 해야겠다.

그래도 이걸 2주동안 붙잡고 있던게 결국 이런 오류들은 어려운 오류가 아닌데에서 나오는 기간이다.

아무튼 기초를 좀 더 탄탄히 하고 항상 꼼꼼하게 하자 🔥🔥🔥🔥🔥

일전에 카프카 에러 포스팅 에서

Configuration 설정을 자바 클래스에서 해주었다. 그것도 클래스별로!!

에러 포스팅과 더불어

This error handler cannot process 'SerializationException's directly; please consider configuring an 'ErrorHandlingDeserializer' in the value and/or key deserializer

이런 문구도 출력해줬었다.

근데 저 에러 포스팅을 보면서 좀 더 공통화할 수 없을까에서 찾아보다가 ErrorHandlingDeserializer 관련 검색을 해보다가 문서에서 찾게 되었던게 있는데,

기존의 설정을 이미지로 한번 가져와봤다.

그런데 이 방법을 Listener가 늘어나면 늘어날 수록 고수할 수가 없다는 생각이 들었다.

세부적인 사항 외에는 조금 다 yaml로 공통화를 할 수 있지 않을까?? 나는 SpringBoot를 사용하는데!

그래서 문서를 찾아본 결과

yaml 설정

spring:
  kafka:
    consumer:
      group-id: test
      auto-offset-reset: latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring:
          json:
            use:
              type:
                headers: false
            trusted:
              packages: '*'
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

    listener:
      ack-mode: manual_immediate

    retry:
      topic:
        attempts: 2

일단 헤더로서 값을 받지 않겠다 라는것을 선언해주고

문서처럼
spring.kafka.consumer.properties.spring.json.type.mapping=메시지를 수신할 dto 풀경로

로 전역에서 여러개를 설정해주어도 되지만, 이 부분은 리스너마다 다를것이라고 생각해서 @KafkaListener마다 설정해주기로 했다.

@KafkaListener(properties={
    "max.poll.records=15", // 한 작업에 15개의 레코드를 컨슘한다.
    "spring.kafka.consumer.properties.spring.json.type.mapping=com.github.lsj8367.MessageReq"
})
public void consumeMessage(final MessageReq request, Acknowledgement ack) {
    // 메세지 처리
}

이렇게 완성하여 ConsumerConfiguration을 제거하게 되었다.

나는 Spring을 쓰는 것이 아니라 이 설정을 한번 더 추상화하여 간편하게 쓰게 해주는 Boot를 쓰게해주는데

너무 Spring처럼 쓰는게 아닌가 싶었는데, 원하는대로 잘 바꾸어 준 것 같다.

나머지로 더 전역적으로 쓸 수 있는것은 공통으로 충분히 더 빼주고 특이한 케이스만이 별도의 Configuration 클래스로 빠져서 설정해주어야 하지 않을까 싶다.

아직 해결하지 못한 이슈가 있는데, 특정 Topic만 특이하게 한건을 consume하고 나서부터는 consume을 더이상 하지 않는 오류가 있다.

이 부분을 얼른 해결하고 싶다.

현재 회사에서 spring-kafka 를 이용해서 특정 서비스들의 푸시 메세지 이벤트를 받아서

전송해주는 서버를 구현하고 있다. (오늘 쿠버네티스에 배포까지 했다!! 모르는게 너무많은...)

로컬에서 내 맘대로 메세지 토픽을 발행해서 쏴도 잘 맞게 역직렬화를 수행을 해주길래 그냥 그런가보다.

하고 잘 넘어갔던 찰나에!!!

Class Not Found Exception

왜? 이 에러가 났을까?

일단 나는 구독하는쪽 그러니까 Kafka에서는 Consumer 쪽 만을 구현해주었다.

내 지식이 부족했던 탓인지는 모르겠지만, 어쨌든 같은 JSON 형태라고 생각해서 클래스가 무엇이던 간에

JSON형식만 같다면 Consume해도 괜찮을 것이라고 처음 생각했었다.

그래서 발행 모델인 Producer쪽에서는 예를 들면 MessageReq로 Producer가 보내고

받은 부분인 Consumer에서는 PushReq라고 받는다고 하고 데이터는 둘다 똑같은 형식으로 매칭이 되어있다고 가정한다.

이랬을 때 Consumer 서버를 키면???!!!

바로 Class Not Found Exception이 떠버린다.

Caused by: org.springframework.messaging.converter.MessageConversionException: failed to resolve class name. Class not found [com.github.lsj8367.message.PushReq]; nested exception is java.lang.ClassNotFoundException: com.github.lsj8367.message.MessageReq
  at org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper.getClassIdType(DefaultJackson2JavaTypeMapper.java:138)
  at org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper.toJavaType(DefaultJackson2JavaTypeMapper.java:99)
  at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:342)
  at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:1030)
  at org.apache.kafka.clients.consumer.internals.Fetcher.access$3300(Fetcher.java:110)
  at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.fetchRecords(Fetcher.java:1250)
  at org.apache.kafka.clients.consumer.internals.Fetcher$PartitionRecords.access$1400(Fetcher.java:1099)
  at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:545)
  at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:506)
  at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1269)
  at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1200)
  at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1176)
  at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:741)
  at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:698)
  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
  at java.util.concurrent.FutureTask.run(FutureTask.java:266)
  at java.lang.Thread.run(Thread.java:748)

예제코드는 깃허브에 있지만, producer와 consumer는 각기 다른 서버로 만들어서 진행시켜주어야 한다!! 안그러면 에러 안나고 받지도 않는다.

왜? Consumer쪽엔 직렬화한 MessageReq클래스가 없기 때문이다.

그래서 보내는 쪽에서 미리 받는 Consumer쪽의 객체 형식을 Header에 바인딩 해주거나,

받는 입장에서 Header가 아닌 Method로 받게 설정해줄 수 있다.

해결책

그래서 해결책이 무엇이냐!

위에서 잠깐 말했지만 producer에서 header를 설정해주거나 consumer에서 설정을 해주는 방법이 있다고 했다.

물론 서로 정의가 잘 되어있고, 동시에 구현했다면 나는 전자의 방법을 택했을 것이다.

그렇지만, 각자 개발하는 속도가 있었고, 또 다른 작업들을 계속해서 진행해야 하고 이미 돌아가고 있던 배치서버에서 그걸 바꿔서

다시 배포하기엔 쉽지 않았었다. (사실 말씀드리기도 좀 그랬다.)

디버깅

그래서

맨끝부분에 false를 넣어주었는데, 이는 들어가보면...

생성자를 오버로딩하기에 쭉 들어오니 이 생성자가 나온다.

initialize에서 해당 boolean값을 사용하고 있기 때문에

boolean값에 의해서 어떤 타입으로 결정하는지가 나온다.

그 설명은

위에서 앞서 설명했던 것과 똑같은 설명이 자바독으로 쓰여져있다.

공부해야될거 참 많다...

정리

그래도 내가 혼자 힘으로 이 메세지 서버를 만들면서 정말 재밌게 개발했다.

물론 오늘 배포하는데 쿠버네티스 지식이 없어서 애를 먹었다. ~~ㅋㅋㅋㅋㅋㅋ~~

그래도 백엔드 개발자분들이 도와주셔서 꿀 플러그인도 전수받고 기본 개념을 정리해가는 것 같다.

뭐하나 건들면 바로 공부해야되는게 너무 어렵지만 재밌어서 좋다.

조만간 카프카 기본개념과 쿠버네티스도 정리해야겠다 🔥🔥🔥🔥🔥

ModelAttribute와 RequestBody의 커맨드 객체 파싱이 다른것을 확인했다.
한번 알아보자! ModelAttribute 동작과정은 덤이다.

ModelAttribute

이 포스팅을 하는 이유는 인자가 많을 경우에 post방식으로 조회를 하는 식으로 구성을 했었는데,

코드리뷰중에 이런말이 나왔었다.

get방식으로 다른 객체로 묶어서 한번에 받아보는건 어떤가요?

변수가 많아지면 많아질 수록 수정점이 늘어날것 같아요! 라고 받았다.

그래서 무의식적으로 평소에 하던방식처럼 post로 수정하여 커밋하고 수정했었다.

근데 post로 안바꾸고 get에서 @ModelAttribute 사용하면 객체로 파싱이 된다는것을 듣고 내가 부족했구나 싶었다.

이 글은 그 부분에서 나와 집에와서 따로 정리하여 포스팅한다.

코드는 깃허브에 있다.

컨트롤러와 dto 그리고 컨트롤러 테스트 코드를 간략하게 작성했다.

Debugging

그리고 디버깅을 돌리면 이런 순서로 진행이 된다.

무조건 DisPatcherServlet이 모든 작업을 분산하여 처리 위임을 진행해주는데,

여기서의 핵심은 이 아랫부분이다.

RequestMappingHandlerAdapter

RequestMappingHandlerAdapter로 시작

핸들러 메소드를 처리할 수 있는 ArgumentResolver 들과
returnValueHandler들을 넣어준다.

이렇게 넣어준다!

ServletInvocableHandlerMethod

다음은

ServletInvocableHandlerMethod가 returnValue를 할 수 있는 애로 채택되는데

InvocableHandlerMethod

InvocableHandlerMethod를 통해 HandlerMethodArgumentResolver를 찾는다.

쭉 반복해서 알맞는 것을 탐색중....

완료되면? 여기서 찾게되는 인스턴스가 바로 ModelAttributeMethodProcessor

ModelAttributeMethodProcessor가 처리해주게끔 반환해준다.

ModelAttributeMethodProcessor

ModelAttribute어노테이션을 읽어들이고

ServletModelAttributeMethodProcessor가
어노테이션이 붙은 변수의 타입이 url 변수와 일치하는 속성이 있는지

이렇게 찾는다.

찾고 없으면 null을 반환하여 객체 생성부분으로 간주하고

같은 객체내에 오버로딩된 메소드에서 리플렉션을 이용해서
해당 ModelAttribute 어노테이션이 붙은 객체의 타입을 가져온다.

BeanUtils 클래스를 이용하여
이때 이 타입의 생성자를 찾는데
제일 먼저 기본생성자를 찾고 있다면 기본생성자를 반환하고 아니라면 구현된 생성자를 전부 가져오는데
이때 길이가 1개이면 해당 생성자를 생성해준다.

이후에 ModelAttributeMethodProcessor에서 파라미터와 매칭되는 것을 넣어준다.

커맨드 객체의 차이?

개인적으로 생각했을 때 @ModelAttribute와 @RequestBody를 읽는 커맨드 객체의 차이점은

RequestBody는 Jackson 라이브러리를 통해서 읽어와 ObjectMapper가 사용되서 조금 생성부분에서 차이가 나지않나 생각한다.

@RequestBody

전에 RequestBody 동작이 어떻게 되는지를 보며 작성한 메세지 컨버터 정리

중간 부분에 보면 해당 컨버팅 작업을 MappingJackson2HttpMessageConverter가 수행해주는데

read() 메소드에서 ObjectMapper를 사용하여 매핑해준다.

그래서 @RequestBody는 getter메소드와 기본생성자가 있어도 ObjectMapper가 해주기에 setter를 쓰거나 해주지 않아도 주입된다.

@ModelAttribute

ModelAttribute는 이와는 다르게 커스텀하게 어노테이션을 만들어서 우리가 MethodArgumentResolver를 구현해서 파라미터에 값을 넣어주듯,

어노테이션 + 리플렉션을 이용해서 값을 넣어주는 것이다. 물론 ServletRequest에서 읽어와서 값들을 가지고 있는건 둘다 공통이지만 말이다.

그래서 기본생성자와 getter만 있으면 값을 넣어줄 수가 없기에

기본생성자에 setter를 전부 생성해주거나 전체 필드를 할당할 수 있는 생성자를 만들어 주어야 한다!!

이펙티브 자바를 읽다가 약한참조에 대한 이야기가 나와서 포스팅한다.

참조에는 아래 4가지가 존재한다.

• Strong References (강한 참조)
• Soft References (소프트 참조)
• Weak References (약한 참조)
• Phantom References (팬텀 참조)

이 해당 참조 유형에 따라 GC 실행 대상여부, 시점이 달라진다.

강한참조

new 연산자를 사용하여 객체를 인스턴스화 하고 참조하는 방식.

참조가 해제되지 않으면 GC의 대상이 되지 않는다.

Test test = new Test();

해당 test라는 변수가 참조를 가지고 있다면 GC의 대상이 되지 않는다.

test = null이 되는 순간 GC의 대상이 된다.

소프트참조

대상 객체의 참조가 SoftReference만 있다면 GC의 대상이 된다.

단, JVM 메모리가 부족한 경우에만 Heap에서 제거된다.

메모리가 부족하지 않은경우에는 제거하지 않는다.

public static void main(String[] args) {
        String ss = "문자열";
        SoftReference<String> reference = new SoftReference<>(ss);

        // 이 시점에 GC의 실행 대상이 가능
        ss = null;

        System.gc();

        // JVM의 메모리가 부족하지 않아서 GC 실행 대상이 되지 않은 경우
        // 그대로 유지한다.
        ss = reference.get();
        System.out.println(ss);
}

약한참조

위에서 봤던 소프트참조와 비슷하게 

대상 객체의 참조가 WeakReference만 있다면 GC의 대상이 된다.

다른점은, 메모리가 부족한경우가 아니라 다음 GC가 일어나게 되면 바로 힙에서 제거된다.

public static void main(String[] args) {
        String ss = "문자열";
        WeakReference<String> reference = new WeakReference<>(ss);

        // 이 시점에 GC의 실행 대상이 가능
        ss = null;

        System.gc();

        // gc를 명시적으로 호출했지만 컬렉션이 동작하지 않을수도 있음
        // 그래도 무조건 동작한다고 가정
        ss = reference.get();

        // null 로 비어있게 된다.
        System.out.println(ss);
}

팬텀참조

생성시 ReferenceQueue가 필요하며, PhantomReference의 참조값을 수동으로 clear() 메서드를 실행해야 하고, PhantomReference.get() 메서드는 항상 null을 반환한다는 특징이 있다.

PhantomReference는 객체 내부의 참조를 null로 설정하지 않고 참조된 객체를 phantomly reachable 객체로 만든 이후에 ReferenceQueue에 enqueue 된다.

두가지에서 사용한다.

1. 자원 정리 (finalizer 보다는 조금 나은 방법) 그렇지만 try-with-resources를 사용하자.
2. 생성 비용이 비싼 객체가 언제 메모리에서 해제되는지 알 수 있음.

깃허브 바로가기

예외를 알아보기 전에

에러와 예외의 차이부터 알아보도록 하자.

에러

일단 에러(Error)는 시스템이 비정상적인 상황에 발생하게 된다.

수습할 수 없는 상황에 놓이게되어 개발자가 예측하지 못한경우이다.

예외

예외는 개발자가 구현한 로직에서 발생된 실수나 사용자의 영향이 미쳐 발생하게 되는 것이다.

그렇기에 미리 예측해서 방지할 수가 있다.

이펙티브 자바를 회독하며 스터디를 진행하면서 2장에

IllegalArgumentException이 나오게 되어 이 예외 부분을 정리하게 됐다.

Checked, Unchecked Exception

기본적으로 오류, 예외는 Object를 상속받는 Throwable클래스를 상속받아 구현이 되어있다.

이미지 출처 - https://www.programcreek.com/2009/02/diagram-for-hierarchy-of-exception-classes/

그림을보면 Exception을 포함해 아래 뿌리로 빨간색으로 표시된 예외들은 모두

Checked Exception이다.

푸른색은 전부 Unchecked Exception이다.

RuntimeException을 상속받는 클래스들이면 하나같이 Unchecked Exception이라 할 수 있겠다.

보면 알겠지만, 프로그램이 정상 작동중에 실행될 수 있는 예외라는 뜻이다.

차이점

둘의 큰 차이점은 check, uncheck 둘다 throws로 Exception을 처리하게끔 메소드에 달아주어도

명시적으로 컴파일에서부터 에러를 띄워주는것은 Checked Exception이고 throws로 던져지는 예외는 모두 반드시 처리해주어야 한다.
(상위로 던지거나 자신의 위치에서 try~catch로 처리해주거나)

RuntimeException하위 예외들도 throws에 넣어줄 수 있겠지만, 처리해주어야 할 필요는 없을 수 있다.

같이 throws에 넣어주게 되면 이러이러한 예외가 발생한다~ 정도를 나타내는 의미정도로 생각하면 되겠다.

근데 이마저도 너무 많은 예외를 throws에 같이 넣어주면 가독성을 반대로 해칠수도 있다.

예외로 의미있는 어떤 무언가를 반드시 처리해줄 수 있는 로직이라면 Checked Exception을

예외 상황이나 문제를 해결할 수 없다면 Unchecked Exception을 활용해 볼 수 있다.

Checked Exception을 던지게 되었을 때 이 예외를 처리하는 무언가의 핸들러까지 던져지게 될텐데 (무지성 throws)

이럴때 그냥 try~catch해주면서 catch부분에 해당 Checked Exception을 받아서 Unchecked Exception으로 바꾸어 던져주는 방법도 있을 것이다.

트랜잭션 처리에 관하여...

많은 글들에서 트랜잭션 처리 시

• Checked Exception은 Rollback 하지않음
• Unchecked Exception은 Rollback 한다
  라고 되어있는데

어떤 트랜잭션인지 명시도 되어있지 않고 무작정 Rollback을 한다 안한다의 여부는 조금 잘못되었다고 백기선님이 말씀하시는걸 보았다.

좀 당연시하게 Spring Framework에서 @Transactional이라고 무작정 생각하고 보면 맞다고 느꼈었는데,

Kafka의 트랜잭션, DB의 트랜잭션 등등 트랜잭션은 정말 많다.

이 문서만 잘 읽어봐도 오해라는거다.

범용적으로 사용하는 Spring Framework에서 사용되고 있는 @Transactional 어노테이션의 롤백 기준은

바로 Unchecked Exception 종류는 기본 Rollback을 진행한다. 반대로 Checked Exception 종류는 Rollback을 하지 않는다.

기본값으로 두개를 나눠서 Check, Uncheck에 대해 트랜잭션 전략을 나누고는 있지만,

어디까지나 기본값이라는거고, Check, Uncheck상관없이 언제든 내가 원하면 해당 예외를 통해 Rollback전략을 매겨줄 수 있다는 것이다.

그러니 Check와 Uncheck라고 해서 트랜잭션을 롤백한다 안한다의 이분법적 사고는 잘못된 사고라고 이해를 하게 되었다.