개요

회사의 기술스택에서 카프카를 사용하고 있다.
내가 이번에 정리하는 글은 막연하게 ack을 날리는 부분에 있어서 간과했던 로직 때문에 벌어진 이슈를 정리한다.

어떻게 로직이 생겼었는가?

우선 카프카를 사용하기 위해선 스프링 부트에서 Spring for Apache Kafka 를 사용해야 한다.
나는 로직에서 try-catch-finally 를 붙여 사용했는데 동료 개발자분께서 의도하신 것인지는 모르겠으나 catch절을 빼고 로직을 구성했다.

@KafkaListener
public void consumeExample(final ScrapeUpdateDto dto, final Acknowledgement ack) {
    try {
        // do something
    } finally {
        ack.acknowledge();
    }
}

이러한 방식으로 구현을 했었고, try 내부에서는 throw를 하는 부분이 군데군데 적용이 되어있었다.

에러는?

샘플 코드에서 do something에 해당하는 로직에서 문제가 발생하게 되는데, 그것이 바로 재시도 로직은 어디에도 존재하지 않았다.
하물며 Spring측에서 제공하는 @Retryable 이나 Spring Cloud Openfeign에서 설정하는 Retry 를 어디에도 구성하지 않았는데 실제 호출은 원본 호출까지 합쳐 10번을 수행했다.

디버깅 시 당연하다고 생각한 부분

일단 디버깅을 진행했을 때 아 그럼 예외 핸들러가 뭐 처리를 했겠지 라는 생각으로 대충 ErrorHandler를 검색했다.
검색하고 보니까 뭔가 힌트는 얻었다.

그것이 바로 DefaultErrorHandler
근데 이 클래스가 10회가 발생하고 백오프 9번 재시도 한다고 하는데 이 부분을 생성해주는 곳은 어디에도 없었으며, 변수로 명시한 setter 메소드에도 디버깅 포인트를 집어놓아도 null만을 주입하고 있었다.

이것이 바로 그 에러 핸들러의 생성자 부분이다.

아직 잘 모르겠으니 차례대로 디버깅을 수행해보도록 하자!

디버깅

우선 KafkaAnnotationDrivenConfiguration 을 통해 ConcurrentKafkaListenerContainerFactoryConfigurer 를 생성한다.

모든 값이 default 설정이라 그런가 전부 errorHandler 부분이 null을 가지고 있다.
여기서 DefaultErrorHandler는 절대 사용되지 않는다라고 확정지었다.

이후에 ConcurrentKafkaListenerContainerFactory를 커스텀한 설정을 넣지 않는다면 자동생성 빈으로 구성해준다.

죄다 이런 default 클래스들로 구성을 해준다.

이렇게 기본 팩토리 구성을 끝마친다.

팩토리 생성 이후

팩토리가 만들어졌다면 당연히 Listener들을 생성시켜주어야 하는데, 이 부분을 createListenerContainer 메소드로 생성해주고 있다.

아래 로직은 ConcurrentKafkaListenerContainerFactory 가 상속한 AbstractKafkaListenerContainerFactory 에 구성된 로직을 오버라이딩한 메소드이다.

createListenerContainerInstance 로 리스너에 대한 부분을 생성해주게 된다.

여기서 추가적으로 알게된 것은 카프카를 설정하기 위해 yaml파일에 구성하는 concurrency 값을 토대로 리스너 컨테이너를 생성한다.

어 잠깐만! 근데 리스너 컨테이너 생성하면 리스너가 여기서 설정되는거 아냐?

위의 리스너 컨테이너도 마찬가지로 추상 컨테이너를 상속받고 있는데, 이 부분을 살펴봤다.

왜?

당연히 리스너면 컨슈머에 대한 설정 부분이 어느정도 구현이 되어있을 것이라고 추측하고 들어가보았다.

변수에는 롤백 프로세서 변수가 default 설정이 되어있는 것을 볼 수 있다.

아래 코드는 AbstractMessageListenerContainer의 변수로 선언된 일부를 가져왔다.

private AfterRollbackProcessor<? super K, ? super V> afterRollbackProcessor = new DefaultAfterRollbackProcessor<>();

DefaultAfterRollbackProcessor를 따라가보면 정답이 나올 것 같아서 바로 들어갔다~~

아래와 같은 설정 구성이 들어가있었고, DefaultErrorHandler와 같이 최초 1회 수행한 후 로직내부의 별도의 Exception을 처리해주는 부분이 없다면 9회를 재시도하도록 구성이 되어있었다.

디버깅 끝낸 후

이 부분을 끝내고 난 뒤 공식문서를 봤는데,

공식문서에서 deafult error handler 설정 값이 어떻게 들어가있는지 알려주고 있다.

공식문서 바로가기

예외를 처리해줄 핸들러를 구현할 때 참고할 점

추가적으로 코드에서 설명해주고 있는 부분은

AbstractKafkaListenerContainerFactory내부에 있는데,

deprecated 처리된 메소드들

1. 2.2버전 이상
   1. setErrorHandler
   2. setBatchHandler
2. 2.8버전 이상
   1. setRetryTemplate
   2. setRecoveryCallback

전부 setCommonErrorHandler 와 setAfterRollbackProcessor 메소드를 사용하도록 추천하고 있다.

결론

결론은 무조건적으로 ack을 날리더라도 dead letter를 바로바로 처리해줄 수 있는 또 하나의 프로듀서, 컨슈머 한쌍을 만들어 두던지, 아니면 db에 따로 오류난 메시지들을 적재하고 배치로 수행해주던지 구성이 필요할 것 같고,

에러를 명확하게 던진 부분을 컨슘하는 곳에서 명확하게 처리해줄 상황이 아니라면,

Custom한 ErrorHandler를 구성하는게 바람직할 것 같다.

의문점

스크래핑 로직 실행 시간에 대해 의문을 가졌다.

평균 시간이 그럭저럭 다들 비슷한 수준에 머무는데, 이상하게 한 부분만 너무 느렸다. 증권사에 대한 스크래핑 내역이었는데,

특정 증권사만 중복 로그인이 감지되었을 경우 30초를 대기했다가 다시 수행할 수 있게 하는 로직이 들어있었고,

테스트 코드는 그에맞는 정말 30초를 기다리는지에 대한 여부를 테스트 하고 있었다.

🤔 그냥 30초를 기다리는것보단 특정한 법인이 아니라면 30초를 기다리지 않는다 를 테스트하면 되지 않을까?

이렇게 생각했던 이유는 30초를 테스트에서 기다리는 비용이 비싸다고 생각했고, 전체 테스트를 돌릴 때 이 부분때문에 30초를 더 기다려야 한다는 것이다. 그리고 milliseconds 차이값을 비교했기에 간헐적으로 실패하는 경우도 있었다.

그래서 더 고치고 싶었는데 일단 테스트를 돌려봤다.

디버깅

바로 디버깅 시작했는데 내가 stream 로직을 잘 구성하지 못하는 것인가? 라는 생각도 들었다.

stream 내부에 stream을 또 구성하고 있는 상황이었다.

이말은, stream 내부의 stream에선 어떤 값을 반환하던 간에 바깥에서 Stream으로 래핑을 하고 있으니 뭘해도 값이 존재한다는 가정이 먹힌다는 뜻이다.

그래서 위에서 나는 에러점이 보였던 것이다.

해당 부분을 사진을 좀 가져오려고 했는데 워낙에 적나라하게 코드를 다 보여주는 것 같아서 예시 코드를 가져ㅇ와봤다.

가장 짤막하게 예시를 들어볼 수 있는게 햄버거가 아닐까 싶었는데

안에 토핑을 생각하다가 간단하게 햄버거 이름, 빵종류, 소스가 무엇인지만 적어봤다.

물론 햄버거 재료중에 패티가 빠지면 섭한데 대충 보자. ㅋㅋㅋ

Java 14버전부터 해당 record 클래스가 등장했다.

나는 집에서는 java 17버전을 쓰고있기에 record로 적었고

이미지와 아래의 코드는 같은것이라고 생각하면 되겠다.

그러면서 getter가 기본으로 구현되어있고 getXXX() 방식이 아닌 바로 변수 이름과 같은 메소드를 호출할 수있다.

public final class Hamburger {
    private final String name;
    private final String bread;
    private final List<Sauce> sauces;
}

아무튼 이런 햄버거와 안에 들어간 소스를 봐야하는데 소스도 정말 대충 value라고 이름을 지었다.

무튼 이렇게 두개 클래스가 있는데

이런 코드를 예시로 만들었다. 로직만 다를뿐이지 이슈가 있던 로직과 같은 구현법이다.

햄버거들에서 소스만 발라내서 매운 소스가 들어간지 여부를 확인하여 있으면 true 없으면 false를 반환하려고 했었던 로직인 것 같다.

잘보면 뭐 햄버거들중에 소스를 발라내고 그 소스들을 뒤져가면서 null이 아니며 spicy 소스인 데이터를 찾으면 바로 true 반환하려고 했던 것 같다.

위에서 설명했던 30초 로직은 해당 메소드에서 true가 반환되면 타게 되어있었다.

public으로 구현했지만 회사의 로직은 private으로 구현되어 있는 메소드였다.

각설하고, 이부분에 대해 테스트를 해보니 "spicy" 라는 값을 넣지 않아도 무조건 테스트가 항상 true였다.

그래서 걸러주어야 할 로직들도 모두 참이 되어버리는 것이다.

소스에 살사와 머스타드를 넣었음에도 불구하고 spicy 소스가 없는데 참이 나온다.

자세히 보면

stream내부에 stream이 있지 않는가?

내부에서 뭘 수행하던간에 아직 stream 연산을 끝맺어주지 않았기 때문에 Stream<> 객체를 반환해줄 것이고 그렇기 때문에 항상 그 내부 연산이 존재하던 않던 상관없이 Stream 객체 자체가 null이 아니고 존재하기 때문에 findFirst() 를 수행해도 항상 참을 반환하는 것이었다.

개선하기

그래서 어떻게 이 부분을 개선했는가?

public boolean isExistSpicySauce(List<Hamburger> hamburgers) {

        final Set<Sauce> sauces = hamburgers.stream()
            .map(hamburger -> hamburger.sauces())
            .flatMap(List::stream)
            .collect(Collectors.toSet());

        return sauces
            .stream()
            .filter(sauce -> sauce.value() != null)
            .anyMatch(sauce -> "spicy".equals(sauce.value()));
}

이런식으로 개선했다.

결과는 당연히 spicy가 포함되지 않았기 때문에 false를 반환했고 테스트도 성공한것을 볼 수 있다.

stream으로 엮어주고 Set으로 변환한 이유는 어떤 햄버거든 소스들만 추려 spicy만 있으면 됐기 때문에 이런식으로 구현을 해주었다.

회사 코드에선 한 법인의 여러개의 공동인증서를 가지고 판별하는 문제였기 때문에 이것처럼 구현해보았다.

stream내부에서 stream을 또 사용할 때에 주의해서 써야겠다고 이 부분을 수정하면서 느꼈다.

어쩌면 너무 당연한 얘기일 수 있었지만, 이 부분 때문에 모든 스크래핑에서 30초 정도의 중복 로그인 대기 시간이 해소되었고 30초 * n개의 증권사 스크래핑 시간을 특정 기관에서만 중복로그인 30초만 대기할 수 있도록 수정되었다.

무의미한 스크래핑 대기 시간을 해소했다는 얘기이다!!

아무튼 한동안 너무 신규 기능개발건에 대해 초점이 맞춰져서 힘들었었는데, 계속해서 기존 코드 부수고 고치며 성능개선하며 이전 코드들을 돌아볼 수 있는 시간을 생각을 전환하니 어느새 갖고있게 되었다.

해당 생각의 전환은 https://jojoldu.tistory.com/710 여기서 자극을 많이 받게 되었다.

예제는 깃허브에 있다.

트랜잭션을 공부했다고 생각하고 업무에 임했던 나였는데, 도저히 풀리지않는 느낌으로 예외를 받았던게 있다.
이전에 한번 포스팅했던 Stream Closed 에러였는데 이게 또 한번 나를 붙잡았다.
처음 개발환경에서 OpenFeign을 사용하면서 또 logging level에서 IOException이 나는줄알고 이부분으로 삽질을 했다.

근데 그게 아니어서 이 포스팅을 작성했다.

• HelloController.java

@RestController
@RequiredArgsConstructor
public class TestController {

    private final TransactionParentService transactionParentService;

    @GetMapping("/transaction/test")
    public ResponseEntity<?> test() {
        transactionParentService.size();
        return ResponseEntity.ok().build();
    }
}

• TransactionParentService.java

@Service
@RequiredArgsConstructor
public class TransactionParentService {

    private final MemberService memberService;
    private final TransactionErrorService transactionErrorService;

    @Transactional
    public int size() {
        transactionErrorService.throwExceptionLog();
        return memberService.allSize();
    }
}

• TransactionErrorService.java

@Slf4j
@Service
@RequiredArgsConstructor
public class TransactionErrorService {

    private final MemberService memberService;

    @Transactional
    public void throwExceptionLog() {
        try {
            memberService.saveAndException();
        } catch (RuntimeException e) {
            log.error("error : {}", e.getMessage());
        }
        System.out.println("끝");
    }
}

• MemberService.java

@Service
@Transactional
@RequiredArgsConstructor
public class MemberService {

    private final MemberRepository memberRepository;

    public void save() {
        memberRepository.save(new Member(1L, "홍길동"));
    }

    @Transactional
    public void saveAndException() {
        memberRepository.save(new Member(null, "홍길동"));
        throw new RuntimeException();
    }

    public int allSize() {
        return memberRepository.findAll().size();
    }

}

TransactionParentService -> TransactionErrorService -> MemberService 순서로 서비스는 동작하게 되고,
기본적으로 Service에는 메소드마다 @Transactional 어노테이션이 붙어있다.

아래는 application.yaml에

logging:
  level:
    org.springframework.transaction: trace

이와 같은 로그설정을 해주고 찍어본 결과이다.

TRACE 66119 --- [nio-8080-exec-1] o.s.t.i.TransactionInterceptor           : Getting transaction for [com.github.lsj8367.service.TransactionParentService.size]
TRACE 66119 --- [nio-8080-exec-1] o.s.t.i.TransactionInterceptor           : Getting transaction for [com.github.lsj8367.service.TransactionErrorService.throwExceptionLog]
TRACE 66119 --- [nio-8080-exec-1] o.s.t.i.TransactionInterceptor           : Getting transaction for [com.github.lsj8367.service.MemberService.saveAndException]
TRACE 66119 --- [nio-8080-exec-1] o.s.t.i.TransactionInterceptor           : Getting transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]

TRACE 66119 --- [nio-8080-exec-1] o.s.t.i.TransactionInterceptor           : Completing transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.save]
TRACE 66119 --- [nio-8080-exec-1] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.github.lsj8367.service.MemberService.saveAndException] after exception: java.lang.RuntimeException

TRACE 66119 --- [nio-8080-exec-1] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.github.lsj8367.service.TransactionErrorService.throwExceptionLog]
TRACE 66119 --- [nio-8080-exec-1] o.s.t.i.TransactionInterceptor           : Getting transaction for [com.github.lsj8367.service.MemberService.allSize]
TRACE 66119 --- [nio-8080-exec-1] o.s.t.i.TransactionInterceptor           : Getting transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.findAll]

TRACE 66119 --- [nio-8080-exec-1] o.s.t.i.TransactionInterceptor           : Completing transaction for [org.springframework.data.jpa.repository.support.SimpleJpaRepository.findAll]
TRACE 66119 --- [nio-8080-exec-1] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.github.lsj8367.service.MemberService.allSize]
TRACE 66119 --- [nio-8080-exec-1] o.s.t.i.TransactionInterceptor           : Completing transaction for [com.github.lsj8367.service.TransactionParentService.size]

여기서 보면 정상적으로 트랜잭션을 처음엔 쭉 획득한다.
그리고나서 memberService.save()를 통해 SimpleJpaRepository까지 하기위한 transaction을 얻게된다.

에러가 나는 부분

우리의 코드에서는 save 이후 throw new RuntimeException()이 있다.
여기서 부터 after Exception : java.lang ...으로 보이는 rollback 마킹이 존재한다.

org.springframework.transaction.UnexpectedRollbackException: Transaction silently rolled back because it has been marked as rollback-only
    at org.springframework.transaction.support.AbstractPlatformTransactionManager.processCommit(AbstractPlatformTransactionManager.java:752) ~[spring-tx-5.3.22.jar:5.3.22]
    at org.springframework.transaction.support.AbstractPlatformTransactionManager.commit(AbstractPlatformTransactionManager.java:711) ~[spring-tx-5.3.22.jar:5.3.22]
    at org.springframework.transaction.interceptor.TransactionAspectSupport.commitTransactionAfterReturning(TransactionAspectSupport.java:654) ~[spring-tx-5.3.22.jar:5.3.22]
    at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:407) ~[spring-tx-5.3.22.jar:5.3.22]
    at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:119) ~[spring-tx-5.3.22.jar:5.3.22]
    ...생략
    at java.base/java.lang.Thread.run(Thread.java:833) ~[na:na]

이게 도대체 어떻게 동작하는것인가?
추상클래스인 TransactionAspectSupport.invokeWithinTransaction 메소드를 사용하는 TransactionInterceptor의 invoke를 호출해서 안쪽에 추상클래스의 메소드를 사용한다.

해당 @Transactional이 붙은 메소드에서 던져진 exception을 Throwable 객체로 catch절에서 받고있다.

• DefaultTransactionAttribute

• rollbackOn 메소드

이 스프링 기본 구성인 DefaultTransactionAttribute구현에서 rollbackOn 메소드를 사용하게 되는데
이 rollbackOn에서 그토록 얘기했던 말들이 나온다.

RuntimeException이 여기서 채택되어 instanceof로 체크하고 있다!!!

그래서 rollback 마크를 하나를 진행해두고 나머지 트랜잭션 어노테이션에 대해서도 commit을 할거냐 rollback을 할거냐에 대한 작업을 이 해당 aop를 통해 동작한다.

TransactionManager

각 트랜잭션 매니저들은 AbstractPlatformTransactionManager 추상클래스를 상속하여 사용한다.

그래서 트랜잭션 매니저가 전부 commit을 수행하는데,
rollback 전용 마크가 하나라도 붙게되면 UnexpectedRollbackException 이 에러가 나타나게 된다.
그래서 보이게되는 예외 메시지가 Transaction silently rolled back because it has been marked as rollback-only로 나오게 된다.

그럼 Stream Closed 는?

ㅋㅋㅋㅋ.. 그러게 말이다. 이거 왜뜬건지 도저히 이해가 되지 않는데, dev환경에선 이 로그만 보였지 위의 로그가 안보였었다.
근데 로컬에서 이 부분을 수정하니 말끔히 해결됐다.
아무튼 원초적인 문제는 이 무지성 붙이기 @Transactional이었다.
막판에 이게 떠올라서 지웠더니 해결되서 꿀잠각이다.

아무튼 @Transactional에 대한 디버깅 정리를 해본다.

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

그래서 우리는 주로 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 시켜주었다.

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

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

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

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

오랜만에 포스팅하는데 회사에서 그동안 앱 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를 적용하여 비동기로 여러개를 한방에 처리하는 방법을 고려중이다.

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

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

그래도 오늘도 해결했다!

Jenkins

위키백과에서 발췌한 내용에 따른다.
젠킨스(Jenkins)는 소프트웨어 개발 시 지속적 통합(continuous integration) 서비스를 제공하는 툴이다. 다수의 개발자들이 하나의 프로그램을 개발할 때 버전 충돌을 방지하기 위해 각자 작업한 내용을 공유 영역에 있는 Git등의 저장소에 빈번히 업로드함으로써 지속적 통합이 가능하도록 해 준다. MIT 라이선스를 따른다.

발생 시점

현재의 회사에서 배포를 젠킨스를 이용하여 배포를 진행한다.
신규 기능개발과 레거시를 청산하는 작업을 주로 해왔었어서 이쪽을 고치는게 우선은 아니었다.
그래서 모르고 있었던 것일수 있다.
에러 상황을 확인해보자

에러 상세

[Pipeline] End of Pipeline
java.lang.InterruptedException
    at java.base/java.lang.Object.wait(Native Method)
    at java.base/java.lang.Thread.join(Thread.java:1300)
    at java.base/java.lang.Thread.join(Thread.java:1375)
    at java.base/jdk.internal.reflect.GeneratedMethodAccessor774.invoke(Unknown Source)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.codehaus.groovy.reflection.CachedMethod.invoke(CachedMethod.java:93)
    at groovy.lang.MetaMethod.doMethodInvoke(MetaMethod.java:325)
    at groovy.lang.MetaClassImpl.invokeMethod(MetaClassImpl.java:1213)
    at groovy.lang.MetaClassImpl.invokeMethod(MetaClassImpl.java:1022)
    at org.codehaus.groovy.runtime.InvokerHelper.invokePojoMethod(InvokerHelper.java:913)
    at org.codehaus.groovy.runtime.InvokerHelper.invokeMethod(InvokerHelper.java:904)
    at org.codehaus.groovy.runtime.ScriptBytecodeAdapter.invokeMethodN(ScriptBytecodeAdapter.java:168)
    at org.codehaus.groovy.runtime.ScriptBytecodeAdapter.invokeMethodNSafe(ScriptBytecodeAdapter.java:176)
    at org.codehaus.groovy.runtime.ScriptBytecodeAdapter.invokeMethodNSpreadSafe(ScriptBytecodeAdapter.java:183)
    at org.codehaus.groovy.runtime.ScriptBytecodeAdapter.invokeMethod0SpreadSafe(ScriptBytecodeAdapter.java:198)
    at org.hidetake.groovy.ssh.interaction.Interactions.waitForEndOfStream(Interactions.groovy:97)
    at org.hidetake.groovy.ssh.interaction.Interactions$waitForEndOfStream$2.call(Unknown Source)
    at org.hidetake.groovy.ssh.operation.Command.execute(Command.groovy:83)
    at org.hidetake.groovy.ssh.operation.Operation$execute$0.call(Unknown Source)
    at org.hidetake.groovy.ssh.session.execution.Command$Helper.execute(Command.groovy:50)
    at jdk.internal.reflect.GeneratedMethodAccessor951.invoke(Unknown Source)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.codehaus.groovy.reflection.CachedMethod.invoke(CachedMethod.java:93)
    at org.codehaus.groovy.runtime.callsite.StaticMetaMethodSite$StaticMetaMethodSiteNoUnwrapNoCoerce.invoke(StaticMetaMethodSite.java:151)
    at org.codehaus.groovy.runtime.callsite.StaticMetaMethodSite.call(StaticMetaMethodSite.java:91)
    at org.codehaus.groovy.runtime.callsite.AbstractCallSite.call(AbstractCallSite.java:141)
    at org.hidetake.groovy.ssh.session.execution.Command$Trait$Helper.execute(Command.groovy:30)
    at org.hidetake.groovy.ssh.session.execution.Command$Trait$Helper$execute$0.call(Unknown Source)
    at org.hidetake.groovy.ssh.session.SessionHandler.execute(SessionHandler.groovy)
    at jdk.internal.reflect.GeneratedMethodAccessor949.invoke(Unknown Source)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.codehaus.groovy.runtime.callsite.PogoMetaMethodSite$PogoCachedMethodSite.invoke(PogoMetaMethodSite.java:169)
    at org.codehaus.groovy.runtime.callsite.PogoMetaMethodSite.call(PogoMetaMethodSite.java:71)
    at org.codehaus.groovy.runtime.callsite.AbstractCallSite.call(AbstractCallSite.java:133)
    at org.hidetake.groovy.ssh.session.execution.Command$Trait$Helper.execute(Command.groovy)
    at org.hidetake.groovy.ssh.session.execution.Command$Trait$Helper$execute.call(Unknown Source)
    at org.hidetake.groovy.ssh.session.SessionHandler.execute(SessionHandler.groovy)
    at jdk.internal.reflect.GeneratedMethodAccessor948.invoke(Unknown Source)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.codehaus.groovy.reflection.CachedMethod.invoke(CachedMethod.java:93)
    at groovy.lang.MetaMethod.doMethodInvoke(MetaMethod.java:325)
    at org.codehaus.groovy.runtime.metaclass.ClosureMetaClass.invokeMethod(ClosureMetaClass.java:384)
    at groovy.lang.MetaClassImpl.invokeMethod(MetaClassImpl.java:1022)
    at org.codehaus.groovy.runtime.callsite.PogoMetaClassSite.callCurrent(PogoMetaClassSite.java:69)
    at org.codehaus.groovy.runtime.callsite.AbstractCallSite.callCurrent(AbstractCallSite.java:166)
    at org.jenkinsci.plugins.sshsteps.SSHService$_executeCommand_closure3$_closure13.doCall(SSHService.groovy:182)
    at org.jenkinsci.plugins.sshsteps.SSHService$_executeCommand_closure3$_closure13.doCall(SSHService.groovy)
    at jdk.internal.reflect.GeneratedMethodAccessor947.invoke(Unknown Source)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.codehaus.groovy.reflection.CachedMethod.invoke(CachedMethod.java:93)
    at groovy.lang.MetaMethod.doMethodInvoke(MetaMethod.java:325)
    at org.codehaus.groovy.runtime.metaclass.ClosureMetaClass.invokeMethod(ClosureMetaClass.java:294)
    at groovy.lang.MetaClassImpl.invokeMethod(MetaClassImpl.java:1022)
    at org.codehaus.groovy.runtime.callsite.PogoMetaClassSite.call(PogoMetaClassSite.java:42)
    at org.codehaus.groovy.runtime.callsite.CallSiteArray.defaultCall(CallSiteArray.java:48)
    at org.codehaus.groovy.runtime.callsite.PogoMetaClassSite.call(PogoMetaClassSite.java:57)
    at org.hidetake.groovy.ssh.util.Utility.callWithDelegate(Utility.groovy:17)
    at jdk.internal.reflect.GeneratedMethodAccessor427.invoke(Unknown Source)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.codehaus.groovy.reflection.CachedMethod.invoke(CachedMethod.java:93)
    at groovy.lang.MetaMethod.doMethodInvoke(MetaMethod.java:325)
    at org.codehaus.groovy.runtime.callsite.StaticMetaMethodSite.invoke(StaticMetaMethodSite.java:46)
    at org.codehaus.groovy.runtime.callsite.StaticMetaMethodSite.callStatic(StaticMetaMethodSite.java:102)
    at org.codehaus.groovy.runtime.callsite.AbstractCallSite.callStatic(AbstractCallSite.java:214)
    at org.hidetake.groovy.ssh.session.SessionTask.wetRun(SessionTask.groovy:64)
    at jdk.internal.reflect.GeneratedMethodAccessor6430.invoke(Unknown Source)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.codehaus.groovy.runtime.callsite.PogoMetaMethodSite$PogoCachedMethodSiteNoUnwrapNoCoerce.invoke(PogoMetaMethodSite.java:210)
    at org.codehaus.groovy.runtime.callsite.PogoMetaMethodSite.callCurrent(PogoMetaMethodSite.java:59)
    at org.codehaus.groovy.runtime.callsite.AbstractCallSite.callCurrent(AbstractCallSite.java:158)
    at org.hidetake.groovy.ssh.session.SessionTask.call(SessionTask.groovy:48)
    at java_util_concurrent_Callable$call.call(Unknown Source)
    at org.hidetake.groovy.ssh.core.Service.run(Service.groovy:81)
    at org.hidetake.groovy.ssh.core.Service$run$1.call(Unknown Source)
    at org.jenkinsci.plugins.sshsteps.SSHService.executeCommand(SSHService.groovy:177)
    at org.jenkinsci.plugins.sshsteps.steps.CommandStep$Execution$CommandCallable.execute(CommandStep.java:84)
    at org.jenkinsci.plugins.sshsteps.util.SSHMasterToSlaveCallable.call(SSHMasterToSlaveCallable.java:32)
    at hudson.remoting.LocalChannel.call(LocalChannel.java:46)
    at org.jenkinsci.plugins.sshsteps.steps.CommandStep$Execution.run(CommandStep.java:72)
    at org.jenkinsci.plugins.sshsteps.util.SSHStepExecution.lambda$start$0(SSHStepExecution.java:84)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)
Finished: ABORTED

정확히 이 에러 윗부분까지는 클라이언트의 요청을 받아서 처리해주고 있다가
한방에 서버가 다운되어버렸다.
젠킨스에서의 배포 스크립트에도 문제가 있었다.
아래의 쉘 스크립트를 젠킨스가 도커 컨테이너 내부에서 실행해주게 만들었다.

#!/bin/bash
export JAVA_TOOL_OPTIONS="-Dfile.encoding='UTF8' -Duser.timezone=Asia/Seoul"
kill -9 $(lsof -t -i:8080)
cd /root/server/
git pull
git checkout $1
git pull origin $1
kill -9 $(lsof -t -i:8080)
mvn clean package -P dev

java -jar -Dspring.profiles.active=dev target/server-0.0.1-SNAPSHOT.jar

이러한 구성으로 된 쉘 스크립트를 통해 실행을 진행했기 때문에
jar를 즉각 실행하게 만들어서 로그가 그대로 젠킨스에 전부 찍히고,

왜? 🤔

그야 당연할것인데, 로그를 실시간으로 젠킨스가 배포 과정을 찍을텐데,

java -jar 명령어를 백그라운드로 실행시켜주지를 않았다.

젠킨스의 배포는 항상 finished상태가 나오질 않는 상태였다.

항상 이상태였다 ㅋㅋㅋ

gif 처음만들어봤는데 재밌네..
이부분에서 로그가 과다하게 많이 쌓이게 되어 에러를 내뱉고
was가 죽어버린 상태가 되어버렸다. (비정상적 셧다운)

해결

JENKINS-45150 large console logging can take Jenkins down or affecting performance - Jenkins Jira
검색을 진행해보니 위와같은 내용들도 얻을 수 있었다.
그래서 해결과정의 순서를 생각한 방식은 다음과 같다.

1. 젠킨스는 배포를 끝내서
   이러한 화면을 만들어주어야 한다.

2. 스프링 애플리케이션의 배포 스크립트를 바꿔주어야 한다. (java -jar를 백그라운드로)

이렇게 하면 되겠다!

그래서 전부 바꿔주게 된다.

#!/bin/bash
export JAVA_TOOL_OPTIONS="-Dfile.encoding='UTF8' -Duser.timezone=Asia/Seoul"
kill -15 $(lsof -t -i:8080)
cd /root/server/
git pull
git checkout $1
git pull origin $1
kill -15 $(lsof -t -i:8080)
mvn clean package -P dev

nohup java -jar -Dspring.profiles.active=dev target/server-0.0.1-SNAPSHOT.jar > ~/app.log 2>&1 &

echo "Deploy Success"

nohup을 이용한 중단없이 실행해주고 젠킨스는 밖으로 빠져나와야 했기 때문에
이 명령어를 선택하고 실행해주었다.

그와 동시에 간단하게 라이브로 볼수있게끔 기본적으로 생성되는 nohup.out을 혹시몰라 만들어둔채로 마무리를 해놓았다.

그리고 kill명령어를 15로 바꾸었는데,
9는 강제종료기 때문에 진행중이던 작업을 즉시 종료하고 데이터도 저장하지 않는다.
15는 자신이 하던 작업을 모두 안전하게 종료하는 절차를 밟는다.
메모리상에 있는 데이터와 각종 설정/환경 파일을 안전하게 저장한 후 프로세스를 종료한다.
15로 한다고 한들, 종료 명령어를 주게되면, 어떤 클라이언트가 요청을 보내서 작업중인 데이터도 끊어질 것이다.
그래서 spring에서 제공하는 graceful shutdown을 적용하고 kill -15를 같이 붙여주었다.
graceful shutdown은 지금 포스팅에서 다루지 않겠다.
이렇게 해서 젠킨스의 로그 과다 적재로 서버가 죽는 현상을 제거시키게 되었다.

정리

이렇게 되어 젠킨스에 기존에 (내가 건드리지 않은) 잘못되게 설정되어 있던 것을 고치게 되니
괜찮다.
그러면서 동시에 툴도 툴마다의 각자의 할일이 있는 것인데,
CI/CD를 위한 툴에서 로그 모니터링까지 하고 있었으니 과다적재로 에러를 뱉는다는 것은
어찌보면 당연한 것이었을 수 있다고 나는 생각한다.🔥