Streams 애플리케이션 테스트하기

Streams 애플리케이션 테스트하기

스트림즈 앱을 테스트할 때 진짜 카프카 클러스터를 띄우는 건 번거롭죠. 카프카가 제공하는 test-utils 모듈의 TopologyTestDriver를 쓰면 실제 카프카 없이도 토폴로지에 데이터를 흘려보내면서 결과를 검증할 수 있어요. 이 페이지에서 테스트 유틸리티 가져오기부터 TopologyTestDriver, TestInputTopic, TestOutputTopic, 그리고 Processor 유닛 테스트용 MockProcessorContext까지 정리해드릴게요.

출처: 문서

본문

Kafka Streams 테스팅

테스트 유틸리티 가져오기

Kafka Streams 애플리케이션을 테스트하기 위해 카프카는 테스트 코드 베이스에 일반 의존성으로 추가할 수 있는 test-utils 아티팩트를 제공해요. Maven을 사용할 때의 예시 pom.xml 스니펫:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams-test-utils</artifactId>
    <version>4.3.1</version>
    <scope>test</scope>
</dependency>

Streams 애플리케이션 테스트

test-utils 패키지는 Processor API를 사용해 수동으로 조립하거나 StreamsBuilder를 사용해 DSL로 조립한 Topology를 통해 데이터를 흘려보내는 데 사용할 수 있는 TopologyTestDriver를 제공해요. 테스트 드라이버는 입력 토픽에서 레코드를 지속적으로 가져와 토폴로지를 가로질러 처리하는 라이브러리 런타임을 시뮬레이션해요. 테스트 드라이버를 사용해 지정된 프로세서 토폴로지가 수동으로 흘려보낸 데이터 레코드로 올바른 결과를 계산하는지 검증할 수 있어요. 테스트 드라이버는 결과 레코드를 캡처하고 임베드된 상태 저장소를 쿼리할 수 있게 해줘요.

// Processor API
Topology topology = new Topology();
topology.addSource("sourceProcessor", "input-topic");
topology.addProcessor("processor", ..., "sourceProcessor");
topology.addSink("sinkProcessor", "output-topic", "processor");
// or
// using DSL
StreamsBuilder builder = new StreamsBuilder();
builder.stream("input-topic").filter(...).to("output-topic");
Topology topology = builder.build();

// create test driver
TopologyTestDriver testDriver = new TopologyTestDriver(topology);

테스트 드라이버로 토픽 이름과 해당 시리얼라이저를 주는 TestInputTopic을 만들 수 있어요. TestInputTopic은 새 메시지 값, 키와 값, 또는 KeyValue 객체 목록을 파이프하는 다양한 메서드를 제공해요.

TestInputTopic<String, Long> inputTopic = testDriver.createInputTopic("input-topic", stringSerde.serializer(), longSerde.serializer());
inputTopic.pipeInput("key", 42L);

출력을 검증하려면 초기화 중에 토픽과 해당 디시리얼라이저를 구성하는 TestOutputTopic을 사용할 수 있어요. 이것은 결과 레코드의 특정 부분만 또는 레코드의 컬렉션만 읽는 헬퍼 메서드를 제공해요. 예를 들어 결과 레코드의 타임스탬프가 아니라 키와 값만 신경 쓴다면 반환된 KeyValue를 표준 어서션으로 검증할 수 있어요.

TestOutputTopic<String, Long> outputTopic = testDriver.createOutputTopic("output-topic", stringSerde.deserializer(), longSerde.deserializer());
assertThat(outputTopic.readKeyValue(), equalTo(new KeyValue<>("key", 42L)));

TopologyTestDriver는 punctuation도 지원해요. 이벤트-시간(event-time) punctuation은 처리된 레코드의 타임스탬프에 기반해 자동으로 촉발돼요. 벽시계-시간(wall-clock-time) punctuation은 테스트 드라이버의 벽시계 시간을 진행시켜 촉발할 수도 있어요 (드라이버는 내부적으로 벽시계 시간을 모킹해 사용자에게 제어권을 줍니다).

testDriver.advanceWallClockTime(Duration.ofSeconds(20));

추가로, 테스트 전이나 후에 테스트 드라이버를 통해 상태 저장소에 접근할 수 있어요. 테스트 전에 저장소에 접근하는 것은 일부 초기 값으로 저장소를 미리 채우는 데 유용해요. 데이터가 처리된 후에는 저장소에 대한 예상 업데이트를 검증할 수 있어요.

KeyValueStore store = testDriver.getKeyValueStore("store-name");

모든 리소스가 제대로 해제되도록 항상 마지막에 테스트 드라이버를 닫아야 한다는 점을 유의해요.

testDriver.close();

예시

다음 예시는 테스트 드라이버와 헬퍼 클래스를 사용하는 방법을 보여줘요. 예시는 키-값 저장소를 사용해 키당 최대값을 계산하는 토폴로지를 만들어요. 처리 중에는 출력이 생성되지 않고 저장소만 업데이트돼요. 출력은 이벤트-시간과 벽시계 punctuation을 기반으로만 다운스트림으로 보내져요.

private TopologyTestDriver testDriver;
private TestInputTopic<String, Long> inputTopic;
private TestOutputTopic<String, Long> outputTopic;
private KeyValueStore<String, Long> store;

private Serde<String> stringSerde = new Serdes.StringSerde();
private Serde<Long> longSerde = new Serdes.LongSerde();

@Before
public void setup() {
    Topology topology = new Topology();
    topology.addSource("sourceProcessor", "input-topic");
    topology.addProcessor("aggregator", new CustomMaxAggregatorSupplier(), "sourceProcessor");
    topology.addStateStore(
        Stores.keyValueStoreBuilder(
            Stores.inMemoryKeyValueStore("aggStore"),
            Serdes.String(),
            Serdes.Long()).withLoggingDisabled(), // need to disable logging to allow store pre-populating
        "aggregator");
    topology.addSink("sinkProcessor", "result-topic", "aggregator");

    // setup test driver
    Properties props = new Properties();
    props.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Long().getClass().getName());
    testDriver = new TopologyTestDriver(topology, props);

    // setup test topics
    inputTopic = testDriver.createInputTopic("input-topic", stringSerde.serializer(), longSerde.serializer());
    outputTopic = testDriver.createOutputTopic("result-topic", stringSerde.deserializer(), longSerde.deserializer());

    // pre-populate store
    store = testDriver.getKeyValueStore("aggStore");
    store.put("a", 21L);
}

@After
public void tearDown() {
    testDriver.close();
}

@Test
public void shouldFlushStoreForFirstInput() {
    inputTopic.pipeInput("a", 1L);
    assertThat(outputTopic.readKeyValue(), equalTo(new KeyValue<>("a", 21L)));
    assertThat(outputTopic.isEmpty(), is(true));
}

@Test
public void shouldNotUpdateStoreForSmallerValue() {
    inputTopic.pipeInput("a", 1L);
    assertThat(store.get("a"), equalTo(21L));
    assertThat(outputTopic.readKeyValue(), equalTo(new KeyValue<>("a", 21L)));
    assertThat(outputTopic.isEmpty(), is(true));
}

@Test
public void shouldNotUpdateStoreForLargerValue() {
    inputTopic.pipeInput("a", 42L);
    assertThat(store.get("a"), equalTo(42L));
    assertThat(outputTopic.readKeyValue(), equalTo(new KeyValue<>("a", 42L)));
    assertThat(outputTopic.isEmpty(), is(true));
}

@Test
public void shouldUpdateStoreForNewKey() {
    inputTopic.pipeInput("b", 21L);
    assertThat(store.get("b"), equalTo(21L));
    assertThat(outputTopic.readKeyValue(), equalTo(new KeyValue<>("a", 21L)));
    assertThat(outputTopic.readKeyValue(), equalTo(new KeyValue<>("b", 21L)));
    assertThat(outputTopic.isEmpty(), is(true));
}

@Test
public void shouldPunctuateIfEvenTimeAdvances() {
    final Instant recordTime = Instant.now();
    inputTopic.pipeInput("a", 1L,  recordTime);
    assertThat(outputTopic.readKeyValue(), equalTo(new KeyValue<>("a", 21L)));

    inputTopic.pipeInput("a", 1L,  recordTime);
    assertThat(outputTopic.isEmpty(), is(true));

    inputTopic.pipeInput("a", 1L, recordTime.plusSeconds(10L));
    assertThat(outputTopic.readKeyValue(), equalTo(new KeyValue<>("a", 21L)));
    assertThat(outputTopic.isEmpty(), is(true));
}

@Test
public void shouldPunctuateIfWallClockTimeAdvances() {
    testDriver.advanceWallClockTime(Duration.ofSeconds(60));
    assertThat(outputTopic.readKeyValue(), equalTo(new KeyValue<>("a", 21L)));
    assertThat(outputTopic.isEmpty(), is(true));
}

public class CustomMaxAggregatorSupplier implements ProcessorSupplier<String, Long> {
    @Override
    public Processor<String, Long> get() {
        return new CustomMaxAggregator();
    }
}

public class CustomMaxAggregator implements Processor<String, Long> {
    ProcessorContext context;
    private KeyValueStore<String, Long> store;

    @SuppressWarnings("unchecked")
    @Override
    public void init(ProcessorContext context) {
        this.context = context;
        context.schedule(Duration.ofSeconds(60), PunctuationType.WALL_CLOCK_TIME, time -> flushStore());
        context.schedule(Duration.ofSeconds(10), PunctuationType.STREAM_TIME, time -> flushStore());
        store = (KeyValueStore<String, Long>) context.getStateStore("aggStore");
    }

    @Override
    public void process(String key, Long value) {
        Long oldValue = store.get(key);
        if (oldValue == null || value > oldValue) {
            store.put(key, value);
        }
    }

    private void flushStore() {
        KeyValueIterator<String, Long> it = store.all();
        while (it.hasNext()) {
            KeyValue<String, Long> next = it.next();
            context.forward(next.key, next.value);
        }
    }

    @Override
    public void close() {}
}

프로세서 유닛 테스트

Processor를 작성한다면 그것을 테스트하고 싶을 거예요. Processor는 결과를 반환하는 대신 컨텍스트로 전달(forward)하므로, 유닛 테스트는 검사를 위해 전달된 데이터를 캡처할 수 있는 모킹된 컨텍스트가 필요해요. 이러한 이유로 test-utils에 MockProcessorContext를 제공해요.

구성

먼저 프로세서를 인스턴스화하고 모크 컨텍스트로 초기화해요:

final Processor processorUnderTest = ...;
final MockProcessorContext<String, Long> context = new MockProcessorContext<>();
processorUnderTest.init(context);

프로세서에 구성(config)을 전달하거나 기본 serde를 설정해야 한다면, config로 모크를 만들 수 있어요:

final Properties props = new Properties();
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Long().getClass());
props.put("some.other.config", "some config value");
final MockProcessorContext<String, Long> context = new MockProcessorContext<>(props);

캡처된 데이터

모크는 프로세서가 전달하는 모든 값을 캡처해요. 그것들에 대해 어서션을 만들 수 있어요:

processorUnderTest.process("key", "value");

final Iterator<CapturedForward<? extends String, ? extends Long>> forwarded = context.forwarded().iterator();
assertEquals(forwarded.next().record(), new Record<>(..., ...));
assertFalse(forwarded.hasNext());

// you can reset forwards to clear the captured data. This may be helpful in constructing longer scenarios.
context.resetForwards();

assertEquals(context.forwarded().size(), 0);

프로세서가 특정 자식 프로세서로 전달한다면, 자식 이름으로 캡처된 데이터를 컨텍스트에 쿼리할 수 있어요:

final List<CapturedForward<? extends String, ? extends Long>> captures = context.forwarded("childProcessorName");

모크는 프로세서가 컨텍스트에서 commit()을 호출했는지도 캡처해요:

assertTrue(context.committed());

// commit captures can also be reset.
context.resetCommit();

assertFalse(context.committed());

레코드 메타데이터 설정

프로세서 로직이 레코드 메타데이터(토픽, 파티션, 오프셋)에 의존한다면 컨텍스트에 설정할 수 있어요:

context.setRecordMetadata("topicName", /*partition*/ 0, /*offset*/ 0L);

이것들이 설정되면 새 값을 설정할 때까지 컨텍스트가 같은 값을 계속 반환해요.

상태 저장소

punctuator가 상태 저장(stateful)이라면 모크 컨텍스트에서 상태 저장소를 등록할 수 있어요. 적절한 유형(KeyValue, Windowed, 또는 Session)의 간단한 인메모리 저장소를 사용하는 것을 권장해요. 모크 컨텍스트는 체인지로그, 상태 디렉터리 등을 관리하지 않기 때문이에요.

final KeyValueStore<String, Integer> store =
    Stores.keyValueStoreBuilder(
            Stores.inMemoryKeyValueStore("myStore"),
            Serdes.String(),
            Serdes.Integer()
        )
        .withLoggingDisabled() // Changelog is not supported by MockProcessorContext.
        .build();
store.init(context, store);
context.register(store, /*deprecated parameter*/ false, /*parameter unused in mock*/ null);

Punctuator 검증

프로세서는 주기적 작업을 처리하기 위해 punctuator를 스케줄할 수 있어요. 모크 컨텍스트는 punctuator를 자동으로 실행하지 않지만, 그것들도 유닛 테스트할 수 있도록 캡처해요:

final MockProcessorContext.CapturedPunctuator capturedPunctuator = context.scheduledPunctuators().get(0);
final long interval = capturedPunctuator.getIntervalMs();
final PunctuationType type = capturedPunctuator.getType();
final boolean cancelled = capturedPunctuator.cancelled();
final Punctuator punctuator = capturedPunctuator.getPunctuator();
punctuator.punctuate(/*timestamp*/ 0L);

스케줄된 punctuator의 자동 촉발을 포함하는 테스트를 작성해야 한다면, 프로세서로 간단한 토폴로지를 만들고 TopologyTestDriver를 사용하는 것을 권장해요.

더 알아보기

• Processor API — Processor/Punctuator를 작성하는 방법을 봐요.
• 데이터 타입과 직렬화 — 테스트에서 시리얼라이저·디시리얼라이저를 설정하는 법을 봐요.
• 보간 쿼리 — 테스트에서 상태 저장소를 쿼리하는 맥락을 봐요.