Debugging

  동기식 또는 명령형 프로그래밍 방식은 Exception이 발생했을 때 스택트레이스(Stacktrace)를 확인하거나 예외 발생이 예상되는 코드에 브레이크포인트(Breakpoint)를 걸어서 디버깅합니다. 반면에 Reactor는 처리되는 작업들이 대부분 비동기적으로 실행되고, Reactor Sequence는 선언형 프로그래밍 방식으로 디버깅이 쉽지 않습니다.

 

Reactor에서의 Debugging 방법

• Debug 모드를 활성화 하는 방법(Globally)
• checkpoint() Operator를 사용하는 방법(Locally)
• log() Operator를 사용해서 Reactor Sequence에서 발생하는 Signal을 확인하는 방법

 

Debug Mode를 사용한 Debugging

  Reactor에서는 디버그 모드(Debug Mode)를 활성화해서 Reactor Sequence를 디버깅할 수 있습니다. Reactor에서의 디버그 모드 활성화는 Hooks.onOperatorDebug()를 통해서 이루어집니다. 다음은 Operator 체인이 시작되기 전에 Hooks.onOperatorDebug()를 통해 디버그 모드를 활성화 하는 예제 코드입니다. 실행 결과를 보면 에러가 발생한 지점을 좀 더 명확하게 찾을 수 있습니다.

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Hooks;
import reactor.core.scheduler.Schedulers;

import java.util.HashMap;
import java.util.Map;

/**
 * onOperatorDebug() Hook 메서드를 이용한 Debug mode 예
 * - 애플리케이션 전체에서 global 하게 동작한다.
 */
@Slf4j
public class Example12_1 {
    public static Map<String, String> fruits = new HashMap<>();

    static {
        fruits.put("banana", "바나나");
        fruits.put("apple", "사과");
        fruits.put("pear", "배");
        fruits.put("grape", "포도");
    }

    public static void main(String[] args) throws InterruptedException {
        Hooks.onOperatorDebug();

        Flux
                .fromArray(new String[]{"BANANAS", "APPLES", "PEARS", "MELONS"})
                .subscribeOn(Schedulers.boundedElastic())
                .publishOn(Schedulers.parallel())
                .map(String::toLowerCase)
                .map(fruit -> fruit.substring(0, fruit.length() - 1))
                .map(fruits::get)
                .map(translated -> "맛있는 " + translated)
                .subscribe(
                        log::info,
                        error -> log.error("# onError:", error));

        Thread.sleep(100L);
    }
}

Hooks.onOperatorDebug()

  디버그 모드의 활성화는 다음과 같이 애플리케이션 내에서 비용이 많이 드는 동작 과정을 거칩니다.

• 애플리케이션 내에 있는 모든 Operator의 스택트레이스(Stacktrace)를 캡처(Capture)합니다.
• 에러가 발생하면 캡처한 정보를 기반으로 에러가 발생한 Assembly의 스택트레이스를 원본 스택트레이스 중간에 끼어 넣습니다.

 

  따라서, 에러 원인을 추적하기 위해 처음부터 디버그 모드를 활성화하는 것은 권장하지 않습니다.

 

Note. 용어

• Stacktrace: 호출된 메서드에 대한 Stack Frame에 대한 리포트입니다. 
• Assembly: Operator에서 리턴하는 새로운 Mono or Flux가 선언된 지점입니다.
• Traceback: 에러가 발생한 Operator의 스택트레이스를 캡처한 Assembly 정보입니다. Suppressed Exception(https://www.baeldung.com/java-suppressed-exceptions) 형태로 원본 스택트레이스에 추가됩니다.

checkpoint() Operator를 사용한 Debuging

  디버그 모드를 활성화하는 방법이 애플리케이션 내에 있는 모든 Operator에서 스택트레이스를 캡처하는 반면에, checkpoint() Operator를 사용하면 특정 Operator 체인 내의 스택트레이스만 캡처합니다.

 

Traceback을 출력하는 방법

  checkpoint()를 사용하면 실제 에러가 발생한 assembly 지점 또는 에러가 전파된 assembly 지점의 Traceback이 추가됩니다. 다음 코드의 실행 결과를 보면 두 개의 checkpoint() 지점이 모두 에러와 관련이 있음을 알 수 있습니다(두 개 모두 Traceback이 출력됩니다).

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;

/**
 *  checkpoint()를 사용한 디버깅 예
 * - checkpoint()를 지정한 Operator 체인에서만 동작한다.
 */
@Slf4j
public class Example12_3 {
    public static void main(String[] args) {
        Flux
            .just(2, 4, 6, 8)
            .zipWith(Flux.just(1, 2, 3, 0), (x, y) -> x/y)
            .checkpoint()
            .map(num -> num + 2)
            .checkpoint()
            .subscribe(
                    data -> log.info("# onNext: {}", data),
                    error -> log.error("# onError:", error)
            );
    }
}

 

Traceback 출력 없이 식별자를 포함한 Description을 출력해서 에러 발생 지점을 예상하는 방법

  checkpoint(description)을 사용하면 에러 발생 시 Traceback을 생략하고 description을 통해 에러 발생 지점을 예상할 수 있습니다.

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;

/**
 * checkpoint(description)을 사용한 디버깅 예
 * - description 을 추가해서 에러가 발생한 지점을 구분할 수 있다.
 * - description 을 지정할 경우 traceback 을 추가하지 않는다.
 */
@Slf4j
public class Example12_4 {
    public static void main(String[] args) {
        Flux
            .just(2, 4, 6, 8)
            .zipWith(Flux.just(1, 2, 3, 0), (x, y) -> x/y)
            .checkpoint("Example12_4.zipWith.checkpoint")
            .map(num -> num + 2)
            .checkpoint("Example12_4.map.checkpoint")
            .subscribe(
                    data -> log.info("# onNext: {}", data),
                    error -> log.error("# onError:", error)
            );
    }
}

 

Traceback과 Description을 모두 출력하는 방법

  checkpoint(description, forceStackTrace)를 사용하면 description과 Traceback을 모두 출력할 수 있습니다.

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;

/**
 * checkpoint(description)을 사용한 디버깅 예
 * - description 을 추가해서 에러가 발생한 지점을 구분할 수 있다.
 * - forceStackTrace 을 true로 지정할 경우 traceback도 추가한다.
 */
@Slf4j
public class Example12_5 {
    public static void main(String[] args) {
        Flux
            .just(2, 4, 6, 8)
            .zipWith(Flux.just(1, 2, 3, 0), (x, y) -> x/y)
            .checkpoint("Example12_4.zipWith.checkpoint", true)
            .map(num -> num + 2)
            .checkpoint("Example12_4.map.checkpoint", true)
            .subscribe(
                    data -> log.info("# onNext: {}", data),
                    error -> log.error("# onError:", error)
            );
    }
}

 

  다음은 서로 다른 Operator 체인에서의 checkpoint()를 활용한 예제 코드입니다. 각각의 Operator 체인에 checkpoint()를 추가한 후, 범위를 좁혀 가면서 단계적으로 에러 발생 지점을 찾을 수 있습니다.

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;

/**
 * 복잡한 단계를 거치는 Operator 체인에서 checkpoint()를 사용하는 예제
 */
@Slf4j
public class Example12_6 {
    public static void main(String[] args) {
        Flux<Integer> source = Flux.just(2, 4, 6, 8);
        Flux<Integer> other = Flux.just(1, 2, 3, 0);

        Flux<Integer> multiplySource = divide(source, other).checkpoint();
        Flux<Integer> plusSource = plus(multiplySource).checkpoint();


        plusSource.subscribe(
                data -> log.info("# onNext: {}", data),
                error -> log.error("# onError:", error)
        );
    }

    private static Flux<Integer> divide(Flux<Integer> source, Flux<Integer> other) {
        return source.zipWith(other, (x, y) -> x/y);
    }

    private static Flux<Integer> plus(Flux<Integer> source) {
        return source.map(num -> num + 2);
    }
}

 

log() Operator를 사용한 Debugging

  log() Operator는 Reactor Sequence의 동작을 로그로 출력하는데, 이 로그를 통해 디버깅이 가능합니다. log() Operator는 사용 개수에 제한이 없기 때문에 1개 이상의 log() Operator로 Reactor Sequence의 내부 동작을 좀 더 상세하게 분석하면서 디버깅 할 수 있습니다.

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;

import java.util.HashMap;
import java.util.Map;

/**
 * log() operator를 사용한 예제
 */
@Slf4j
public class Example12_7 {
    public static Map<String, String> fruits = new HashMap<>();

    static {
        fruits.put("banana", "바나나");
        fruits.put("apple", "사과");
        fruits.put("pear", "배");
        fruits.put("grape", "포도");
    }

    public static void main(String[] args) {
        Flux.fromArray(new String[]{"BANANAS", "APPLES", "PEARS", "MELONS"})
                .map(String::toLowerCase)
                .map(fruit -> fruit.substring(0, fruit.length() - 1))
                .log()
//                .log("Fruit.Substring", Level.FINE)
                .map(fruits::get)
                .subscribe(
                        log::info,
                        error -> log.error("# onError:", error));
    }
}

 

 

 

[참고 정보]

• 스프링으로 시작하는 리액티브 프로그래밍
• 전체 소스코드