Application Testing

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

// 1. 토폴로지를 만든다 (프로덕션 코드와 같은 빌더를 쓴다)
StreamsBuilder builder = new StreamsBuilder();
builder.stream("input-topic")
       .filter((k, v) -> v != null)
       .to("output-topic");
Topology topology = builder.build();

// 2. 테스트 드라이버 생성
TopologyTestDriver testDriver = new TopologyTestDriver(topology);

// 3. 입력 토픽 — 직렬화기(serializer)를 준다
TestInputTopic<String, Long> inputTopic =
    testDriver.createInputTopic("input-topic",
        Serdes.String().serializer(), Serdes.Long().serializer());

// 4. 출력 토픽 — 역직렬화기(deserializer)를 준다
TestOutputTopic<String, Long> outputTopic =
    testDriver.createOutputTopic("output-topic",
        Serdes.String().deserializer(), Serdes.Long().deserializer());

// 5. 데이터를 넣고 결과를 검증
inputTopic.pipeInput("key", 42L);
assertThat(outputTopic.readKeyValue(), equalTo(new KeyValue<>("key", 42L)));

// 6. 상태 저장소를 직접 열어 확인할 수 있다
KeyValueStore<String, Long> store = testDriver.getKeyValueStore("counts-store");
assertThat(store.get("key"), equalTo(1L));

// 7. 벽시계 시각을 전진시켜 punctuation 을 발동시킨다
testDriver.advanceWallClockTime(Duration.ofSeconds(20));

// 8. 반드시 닫는다 (리소스 해제)
testDriver.close();

// autoComplete=true 면 send() 가 즉시 성공 처리된다
MockProducer<String, String> producer =
    new MockProducer<>(true, new StringSerializer(), new StringSerializer());

OrderPublisher publisher = new OrderPublisher(producer);   // 인터페이스 주입
publisher.publish(new Order("order-42", 19900));

List<ProducerRecord<String, String>> sent = producer.history();
assertThat(sent, hasSize(1));
assertThat(sent.get(0).topic(), equalTo("orders"));
assertThat(sent.get(0).key(),   equalTo("order-42"));   // 키를 지정했는지 검증

// 실패 경로도 테스트할 수 있다
MockProducer<String, String> failing =
    new MockProducer<>(false, new StringSerializer(), new StringSerializer());
// … publish 호출 후
failing.errorNext(new org.apache.kafka.common.errors.TimeoutException("boom"));

MockConsumer<String, String> consumer =
    new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name());

TopicPartition tp = new TopicPartition("orders", 0);
consumer.assign(List.of(tp));                       // assign 으로 파티션 고정
consumer.updateBeginningOffsets(Map.of(tp, 0L));    // 시작 오프셋을 알려 줘야 한다

consumer.addRecord(new ConsumerRecord<>("orders", 0, 0L, "order-42", "CREATED"));
consumer.addRecord(new ConsumerRecord<>("orders", 0, 1L, "order-42", "PAID"));

new OrderProcessor(consumer, sink).pollOnce();
assertThat(sink.processed(), hasSize(2));

bin/kafka-producer-perf-test.sh \
  --topic perf-test \
  --num-records 1000000 \
  --record-size 1024 \
  --throughput -1 \
  --bootstrap-server localhost:9092 \
  --command-property acks=all \
  --command-property linger.ms=5

bin/kafka-consumer-perf-test.sh \
  --topic perf-test \
  --num-records 1000000 \
  --bootstrap-server localhost:9092 \
  --group perf-consumer

MockProducer<String, Order> producer =
    new MockProducer<>(true, new StringSerializer(), new OrderAvroSerializer());

new OrderPublisher(producer).publish(new Order("order-42", 19900));
assertThat(producer.history(), hasSize(1));

// 5분 tumbling 윈도우로 집계하는 토폴로지
TopologyTestDriver driver = new TopologyTestDriver(topology);
TestInputTopic<String, String> in = driver.createInputTopic("events",
    Serdes.String().serializer(), Serdes.String().serializer());

in.pipeInput("user-1", "click");
in.pipeInput("user-1", "click");
in.pipeInput("user-1", "click");

// 윈도우 집계 결과를 기대하지만 아무것도 읽히지 않는다
assertThat(outputTopic.readKeyValue().value, equalTo(3L));

$ bin/kafka-producer-perf-test.sh --topic t --num-records 5000000 \
    --record-size 1024 --throughput -1 \
    --bootstrap-server localhost:9092 --command-property acks=all

5000000 records sent, 412300 records/sec (402.6 MB/sec),
  1842.11 ms avg latency, 5231.00 ms max latency,
  1701 ms 50th, 4102 ms 95th, 4980 ms 99th, 5180 ms 99.9th