카프카 소비자 서비스 느린 로딩 문제 해결하기

dasd412·2023년 1월 1일

Spring boot Thread Dump consumer kafka o.h.e.t.j.p.i.JtaPlatformInitiator 느린 로딩

MSA 프로젝트

목록 보기

18/25

문제 상황

언제나처럼 코드를 작성한 후 JUnit으로 테스트를 실시하고 있었다.
그런데 run을 했는데도, 스프링 부트가 매우 오래 로딩되고 있었다!

로그는 다음과 같았다.

...
12:11:47.631 [main] INFO  o.h.e.t.j.p.i.JtaPlatformInitiator - HHH000490: Using JtaPlatform implementation: [org.hibernate.engine.transaction.jta.platform.internal.NoJtaPlatform]

그리고는 다음 로그가 진행되지 않았다.

좀 시간 지나고 나서야 테스트가 진행됬다. 너무 오래 걸리는 시간이었다.

원인 후보군 줄이기

로그 레벨을 올려도 아무 소용 없었다.

일단, 문제가 발생한 코드는 @SpringBootTest, @SpringBootApplication이 붙은 것들이었다.
반면 DataJpaTest는 아무런 문제가 없었다.

즉, JPA나 도메인 관련 문제가 아니라, 더 외부 설정에서의 문제라고 추측할 수 있었다.

그리고 내가 짠 로직에 진입하기도 전에 문제가 발생하는 것이므로 디버그는 사용하기 곤란했다.

스레드 덤프를 발견하다.

고민하던 중, intelliJ에서 사진기와 같은 아이콘을 발견했다.

마우스를 올려보니, thread dumps라고 나와 있었다.
dump? 시스템 프로그래밍 수업 때 레지스터 등을 분석할 때 나왔던 이름이었다.
그리고 사진기니까 스냅숏 같은 느낌인가 하고 클릭해봤다.

스레드 덤프 분석

클릭해보니, 다음과 같이 표시되었다.

그리고 친숙한 아이콘인 벌레(디버그)가 있어서 클릭해봤다.

클릭해보면, Object.wait()으로 대기하고 있는 것을 확인할 수 있다.
그리고 더 아래를 보면, kafka어쩌구가 있는 것을 볼 수 있다. 즉, 카프카와 관련된 문제라고 예측할 수 있었다.

문제의 코드

@EnableBinding(DiaryChannels.class)
public class DiaryChangeHandler {

    @StreamListener("inboundDiaryChanges")
    public void changeRelationWithDiary(DiaryChangeModel diaryChange) {
	...
	}
}

카프카 소비자 코드이다. 이 코드가 문제의 원인이라고 생각해서 임시 해결책을 시도 해봤다.

임시 해결책

//@EnableBinding(DiaryChannels.class)
public class DiaryChangeHandler {

    //@StreamListener("inboundDiaryChanges")
    public void changeRelationWithDiary(DiaryChangeModel diaryChange) {
	...
	}
}

카프카 어노테이션을 주석처리하고, 다시 테스트를 실행하니 로딩이 없었다. 역시 카프카 문제였다.

해결책

전에 작성했던 다음 글과 비슷하다.
https://velog.io/@dasd412/JUnit-%ED%85%8C%EC%8A%A4%ED%8A%B8%EC%97%90%EC%84%9C-%EC%B9%B4%ED%94%84%EC%B9%B4-%EC%9D%98%EC%A1%B4%EC%84%B1-%EC%A0%9C%EC%99%B8%ED%95%98%EA%B8%B0

1. handler에서 @EnableBinding 제거

public class DiaryChangeHandler {

    @StreamListener("inboundDiaryChanges")
    public void changeRelationWithDiary(DiaryChangeModel diaryChange) {
	...
	}
}

2. Config에서 책임 맡기

@EnableBinding(DiaryChannels.class)
@ConditionalOnProperty(value = "kafka.enabled", matchIfMissing = true)
@Configuration
public class KafkaConfiguration {
    /*
    카프카 관련은 @Component로 지정하지 말고 여기에 @Bean으로 따로 등록할 것.
    안 그러면 단위 테스트할 때 카프카 들어감.
     */

    private final UpdateWriterService updateWriterService;

    public KafkaConfiguration(UpdateWriterService updateWriterService) {
        this.updateWriterService = updateWriterService;
    }

    @Bean
    public DiaryChangeHandler diaryChangeHandler(){
        return new DiaryChangeHandler(updateWriterService);
    }
}

@EnableBinding(DiaryChannels.class)을 설정 컴포넌트로 배치시켰고, @ConditionalOnProperty(value = "kafka.enabled", matchIfMissing = true)을 이용해서 테스트 환경에서는 카프카 안쓰도록 변경했다.