java

13.1 Kafka Streams 测试

Kafka Streams 测试策略：TopologyTestDriver 创建输入/输出 Topic 并模拟完整拓扑、自定义 Processor 的状态存储测试、eventd v3 清洗管道的确定性测试，以及单元/组件/集成测试的分工。

Leeting Yan 2026-11-01 4 分钟阅读 1645 字

为什么测试 Kafka Streams

Kafka Streams 与 Kafka 深度集成，但测试不必启动真实 Kafka：

单元测试：只需要 TopologyTestDriver
集成测试：可用 testcontainers 启动短暂 Kafka
确定性：输入固定 → 输出固定

eventd v3 的清洗管道使用 TopologyTestDriver 做核心逻辑测试。

TopologyTestDriver 基础

TopologyTestDriver 在内存中模拟 Kafka Streams，不需要 broker：

@Test
void testSimpleTopology() {
    StreamsBuilder builder = new StreamsBuilder();

    builder.<String, String>stream("input-topic")
        .mapValues(String::toUpperCase)
        .to("output-topic");

    Topology topology = builder.build();

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-app");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
    // .to("output-topic") 没写 Produced，取的就是默认 serde；
    // 不设这两项，TopologyTestDriver 会报 "Default value serde is not configured"
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);

    try (TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
        // 创建输入 Topic
        TestInputTopic<String, String> input = driver.createInputTopic(
            "input-topic",
            new StringSerializer(),
            new StringSerializer()
        );

        // 创建输出 Topic
        TestOutputTopic<String, String> output = driver.createOutputTopic(
            "output-topic",
            new StringDeserializer(),
            new StringDeserializer()
        );

        // 发送数据
        input.pipeInput("key1", "hello");
        input.pipeInput("key2", "world");

        // 读取并断言
        TestRecord<String, String> record1 = output.readRecord();
        assertEquals("key1", record1.key());
        assertEquals("HELLO", record1.value());

        TestRecord<String, String> record2 = output.readRecord();
        assertEquals("key2", record2.key());
        assertEquals("WORLD", record2.value());
    }
}

TestInputTopic / TestOutputTopic

TestInputTopic:
  pipeInput(key, value)        → 发送一条记录到 Topic
  pipeInputList(List)          → 发送多条
  pipeInput(record)            → 发送带时间戳的记录

TestOutputTopic:
  readRecord()                 → 读取下一条记录
  readKeyValue()               → 读取 key-value 对
  readValuesToList()           → 读取所有值到 List
  isEmpty()                    → 是否还有未读取的数据

时间戳控制：

// 指定记录的时间戳（用于聚合/窗口测试）
input.pipeInput("key1", "value1", 1700000000000L);  // ms 时间戳

eventd v3 清洗管道测试

@Test
void testCleansingPipeline() {
    CleansingTopology topology = new CleansingTopology();

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-cleansing");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");

    try (TopologyTestDriver driver = new TopologyTestDriver(topology.build(), props)) {
        TestInputTopic<String, String> input = driver.createInputTopic(
            "eventd-raw",
            new StringSerializer(),
            new StringSerializer()
        );
        TestOutputTopic<String, String> output = driver.createOutputTopic(
            "eventd-cleaned",
            new StringDeserializer(),
            new StringDeserializer()
        );
        TestOutputTopic<String, String> deadLetter = driver.createOutputTopic(
            "eventd-dead-letter",
            new StringDeserializer(),
            new StringDeserializer()
        );

        // 有效事件
        input.pipeInput("alice", validEventJson("e1", "ORDER", 100.0));
        input.pipeInput("bob",   validEventJson("e2", "order", 200.0));  // 小写 type

        // 重复事件
        input.pipeInput("alice", validEventJson("e1", "ORDER", 100.0));

        // 无效 JSON
        input.pipeInput("eve", "not-json");

        // 断言：只有 2 条有效输出（去重后）
        List<String> cleaned = output.readValuesToList();
        assertEquals(2, cleaned.size());

        // 断言：标准化（type 转大写）
        Event cleanedBob = JsonMapper.parse(cleaned.get(1), Event.class);
        assertEquals("ORDER", cleanedBob.type());

        // 断言：死信队列收到无效数据
        List<String> dead = deadLetter.readValuesToList();
        assertEquals(1, dead.size());
    }
}

private String validEventJson(String eventId, String type, double amount) {
    return String.format(
        "{\"eventId\":\"%s\",\"type\":\"%s\",\"userId\":\"alice\",\"amount\":%.0f,\"eventTime\":1700000000000}",
        eventId, type, amount
    );
}

自定义 Processor 的测试

FixedKeyProcessor 的上下文是 FixedKeyProcessorContext，而 kafka-streams-test-utils 提供的 MockProcessorContext 实现的是 ProcessorContext，两者不是同一个接口，不能拿来直接 init。最省事、也最接近真实拓扑的做法是用 TopologyTestDriver 从拓扑层测：

@Test
void testDeduplicateProcessor() {
    StreamsBuilder builder = new StreamsBuilder();
    builder.addStateStore(
        Stores.keyValueStoreBuilder(
                Stores.inMemoryKeyValueStore("dedup-store"),
                Serdes.String(),
                Serdes.Long())
            .withLoggingDisabled());

    builder.<String, String>stream("input")
        .mapValues(value -> Event.builder().eventId(value).type("ORDER").build())
        .processValues(
            () -> new DeduplicateProcessor("dedup-store", Duration.ofHours(168)),
            "dedup-store")
        .mapValues(Event::eventId)
        .to("output");

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-dedup");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
    // source 与 sink 都没写 Serde，TopologyTestDriver 会取默认值
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);

    try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
        TestInputTopic<String, String> input = driver.createInputTopic(
            "input", new StringSerializer(), new StringSerializer());
        TestOutputTopic<String, String> output = driver.createOutputTopic(
            "output", new StringDeserializer(), new StringDeserializer());

        input.pipeInput("k1", "e1");
        input.pipeInput("k1", "e1");   // 重复，应被丢弃
        input.pipeInput("k2", "e2");

        assertEquals(List.of("e1", "e2"), output.readValuesToList());
    }
}

三点注意：

状态 store 必须先用 builder.addStateStore(...) 注册，processValues(..., "dedup-store") 再把它挂到 Processor 上；漏掉任何一处，getStateStore 拿到 null，第一次 put 就 NPE。
inMemoryKeyValueStore 不写 changelog，测试里够用；生产用 persistentKeyValueStore，重启后由 changelog 恢复。
要单测会改 key 的 v2 Processor，可以用 processor.api.MockProcessorContext<K, V>，forward 出去的记录从 context.forwarded() 读；但 store 仍要先 init(context.getStateStoreContext(), store) 再 context.addStateStore(store)，addStateStore 只把 store 挂进 map，不做初始化。

// 直接单测 13.1 后面的 CountProcessor
MockProcessorContext<String, Long> context = new MockProcessorContext<>();
KeyValueStore<String, Long> store = Stores.keyValueStoreBuilder(
        Stores.inMemoryKeyValueStore("counts"), Serdes.String(), Serdes.Long())
    .withLoggingDisabled()
    .build();
store.init(context.getStateStoreContext(), store);   // 必须先初始化
context.addStateStore(store);                         // 再挂到 context 上

CountProcessor processor = new CountProcessor("counts");
processor.init(context);
processor.process(new Record<>("alice", "a", 0L));

assertEquals(1L, context.forwarded().get(0).record().value());

含状态操作测试

用 process + Processor 统计每个 key 的出现次数。与 FixedKeyProcessor 不同，Processor 可以改 key，因此计数结果要自己 forward 出去：

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

    public CountProcessor(String storeName) {
        this.storeName = storeName;
    }

    @Override
    public void init(ProcessorContext<String, Long> context) {
        this.context = context;
        this.store = context.getStateStore(storeName);
    }

    @Override
    public void process(Record<String, String> record) {
        Long previous = store.get(record.key());
        long next = previous == null ? 1L : previous + 1;
        store.put(record.key(), next);
        context.forward(record.withValue(next));
    }
}

@Test
void testStateStoreTopology() {
    StreamsBuilder builder = new StreamsBuilder();

    // 创建状态存储
    builder.addStateStore(
        Stores.keyValueStoreBuilder(
            Stores.inMemoryKeyValueStore("counts"),
            Serdes.String(),
            Serdes.Long())
        .withLoggingDisabled());

    // process 返回新的 KStream，可以继续接 sink
    builder.<String, String>stream("input")
        .process(() -> new CountProcessor("counts"), "counts")
        .to("output", Produced.with(Serdes.String(), Serdes.Long()));

    // props 是每个测试自己的，别指望上一个测试的变量还在作用域里
    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-store");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
    // 输入 topic 的值是 String，默认 value serde 必须与输入一致；
    // 输出 topic 的 Long 由 to(..., Produced) 显式指定
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);

    try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
        TestInputTopic<String, String> input = driver.createInputTopic(
            "input", new StringSerializer(), new StringSerializer());
        TestOutputTopic<String, Long> output = driver.createOutputTopic(
            "output", new StringDeserializer(), new LongDeserializer());

        input.pipeInput("alice", "a");
        input.pipeInput("alice", "b");
        input.pipeInput("alice", "c");

        assertEquals(1L, output.readValue());  // alice: 第 1 条
        assertEquals(2L, output.readValue());  // alice: 第 2 条
        assertEquals(3L, output.readValue());  // alice: 第 3 条
    }
}

测试策略总结

测试类型	工具	启动时间	覆盖
单元测试	`TopologyTestDriver`	< 1s	拓扑逻辑、状态操作
组件测试	`MockProcessorContext`	< 1s	自定义 `Processor`（v2 API）
集成测试	Testcontainers Kafka	10-30s	真实 Kafka 交互

eventd v3 测试策略：

清洗管道：TopologyTestDriver + 确定性断言
状态操作：MockProcessorContext + 状态验证
端到端：集成测试（可选，非必须）

小结

工具	用途
`TopologyTestDriver`	端到端拓扑测试
`TestInputTopic`	发送测试数据
`TestOutputTopic`	读取输出断言
`MockProcessorContext`	Processor 单元测试

下一节 13.2 Flink DataStream 测试：MiniCluster、TestHarness、Watermark 控制。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页