java

13.3 端到端测试

eventd v3 端到端可运行测试：基于 Testcontainers 的 Kafka 与 Redis 集成测试、全链路数据从 v2 Producer → Kafka Streams 清洗 → Flink 聚合 → Redis 的链路验证，以及测试数据的确定性生成、Watermark 推进与断言策略。

Leeting Yan 5 分钟阅读 2282 字

为什么选择 Testcontainers

端到端测试需要真实的外部系统，但又不能依赖环境中已有的 Kafka/Redis。Testcontainers 在测试启动时自动创建 Docker 容器，测试结束后销毁：

// 注意包名：支持 apache/kafka 镜像的是 org.testcontainers.kafka.KafkaContainer，
// 老的同名类 org.testcontainers.containers.KafkaContainer 只认 Confluent 镜像
import org.testcontainers.kafka.KafkaContainer;

@Container
static KafkaContainer kafka = new KafkaContainer(
    DockerImageName.parse("apache/kafka-native:3.8.1"));

@Container
static GenericContainer<?> redis = new GenericContainer<>(
    DockerImageName.parse("redis:7-alpine"))
    .withExposedPorts(6379);

KafkaContainer 来自 org.testcontainers:kafka 模块，默认用 KRaft 模式跑单进程 broker。旧教程里的 confluentinc/cp-kafka + withEmbeddedZookeeper() 属于已淘汰的写法；镜像 tag 也会随版本更新，以官方镜像页为准。

E2E 测试框架

@Testcontainers
class EventdEndToEndTest {

    @Container
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("apache/kafka-native:3.8.1"));

    @Container
    static GenericContainer<?> redis = new GenericContainer<>(
        DockerImageName.parse("redis:7-alpine"))
        .withExposedPorts(6379);

    @BeforeAll
    static void init() {
        // 容器端口是随机映射的，被测组件必须从这里取地址，不能写死 localhost:9092
        System.setProperty("kafka.bootstrap-servers", kafka.getBootstrapServers());
        System.setProperty("redis.host", redis.getHost());
        System.setProperty("redis.port", String.valueOf(redis.getMappedPort(6379)));
    }

    @Test
    void testFullPipeline() throws Exception {
        // 1. 建 Topic：测试里用 4 分区，生产是 12 分区
        createTopics();

        // 2. 启动清洗管道（Kafka Streams 拓扑，见 9.2）
        startCleansing();

        // 3. 启动 Flink 聚合（MiniCluster 内提交作业，见 8.3）
        startAnalytics();

        // 4. 生产落在同一个分钟窗口内的测试数据
        long windowStart = System.currentTimeMillis() / 60_000 * 60_000;
        produceTestEvents(windowStart);

        // 5. 等窗口关闭后断言具体窗口 key，而不是 :latest 指针
        awaitOrderCount(windowStart, 10, Duration.ofSeconds(60));

        // 6. 有效数据不应进入死信队列
        assertDeadLetterEmpty();
    }
}

startCleansing() / startAnalytics() 在这里省略实现，它们就是前面章节的拓扑和作业，只是把 bootstrap servers 换成 kafka.getBootstrapServers()。真实项目里要把 KafkaStreams 实例和 Flink JobClient 存成静态字段，在 @AfterAll 里 close() / cancel()——容器销毁后还在跑的客户端会不断刷重连日志。

建 Topic

KafkaContainer 不会按旧教程的 KAFKA_CREATE_TOPICS 环境变量建 Topic（那是另一套镜像的特性）。明确用 AdminClient 建，分区数、副本数才可控：

private static void createTopics() throws Exception {
    Properties props = new Properties();
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());

    try (Admin admin = Admin.create(props)) {
        admin.createTopics(List.of(
            new NewTopic("eventd-raw", 4, (short) 1),
            new NewTopic("eventd-cleaned", 4, (short) 1),
            new NewTopic("eventd-dead-letter", 4, (short) 1)
        )).all().get(30, TimeUnit.SECONDS);
    }
}

发送测试数据

事件时间必须落在已知的窗口里，断言才能算准：

private static void produceTestEvents(long windowStart) throws Exception {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    ObjectMapper mapper = new ObjectMapper();

    try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
        // 10 个用户 × 每个 5 条，全部落在 [windowStart, windowStart + 60000)
        for (int u = 0; u < 10; u++) {
            String userId = "user" + u;
            for (int i = 0; i < 5; i++) {
                Event event = Event.builder()
                    .eventId(UUID.randomUUID().toString())
                    .type("ORDER")
                    .userId(userId)
                    .amount(100.0 + i)
                    .symbol("BTC")
                    .eventTime(windowStart + i * 1000L)
                    .build();
                producer.send(new ProducerRecord<>("eventd-raw", userId,
                    mapper.writeValueAsString(event)));
            }
        }

        producer.flush();

        // 先确认清洗管道已经把 50 条事件写进 eventd-cleaned，再推进 Watermark。
        // 如果 pumper 先到，Flink 会先推进 Watermark，随后到达的窗口内事件
        // 会被当成迟到数据丢掉，窗口结果就永远少数据。
        awaitCleanedEvents(50, Duration.ofSeconds(30));

        // 水印推进事件：事件时间放在窗口结束后 71 秒。
        // 容忍度 10 秒时 Watermark = 71 - 10 - 1 = 60 秒，刚好越过窗口末尾。
        // 流量很小的 E2E 测试没有这条事件，窗口永远不会关闭，断言必然超时。
        Event pump = Event.builder()
            .eventId(UUID.randomUUID().toString())
            .type("ORDER")
            .userId("__pump__")
            .amount(1.0)
            .symbol("BTC")
            .eventTime(windowStart + 71_000L)
            .build();
        // analytics 读的是 eventd-cleaned，不是 eventd-raw。
        // Kafka Streams 会按 key 重新分区：往 raw 的 4 个分区各发一条，
        // 并不能保证 cleaned 的 4 个分区都收到；只要有一个 cleaned 分区没有数据，
        // Flink 的 Watermark 就仍被它压住，下游窗口不会关闭。
        // 所以 pumper 直接写 cleaned 的每个分区，绕开重分区。
        for (int partition = 0; partition < 4; partition++) {
            producer.send(new ProducerRecord<>("eventd-cleaned", partition,
                "__pump__" + partition, mapper.writeValueAsString(pump)));
        }

        producer.flush();
    }
}

private static void awaitCleanedEvents(int expected, Duration timeout) throws Exception {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "e2e-wait-" + UUID.randomUUID());
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

    long deadline = System.nanoTime() + timeout.toNanos();
    int seen = 0;
    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
        consumer.subscribe(List.of("eventd-cleaned"));
        while (seen < expected && System.nanoTime() < deadline) {
            for (ConsumerRecord<String, String> ignored : consumer.poll(Duration.ofMillis(200))) {
                seen++;
            }
        }
    }
    if (seen < expected) {
        throw new AssertionError("eventd-cleaned 只看到 " + seen + " 条，清洗管道可能没启动");
    }
}

这里的 71_000 不是随手写的数：forBoundedOutOfOrderness(Duration.ofSeconds(10)) 产生的 Watermark 是 maxTimestamp - 10s - 1ms，窗口在 Watermark >= windowEnd 时关闭，所以推进事件至少要比窗口末尾晚 10 秒以上。

验证点

Redis 指标验证

private static void awaitOrderCount(long windowStart, int users, Duration timeout)
        throws Exception {
    ObjectMapper mapper = new ObjectMapper();

    try (StatefulRedisConnection<String, String> connection = RedisClient
            .create("redis://" + redis.getHost() + ":" + redis.getMappedPort(6379))
            .connect()) {
        RedisCommands<String, String> commands = connection.sync();

        for (int u = 0; u < users; u++) {
            String key = "eventd:metrics:order-count:" + windowStart + ":user" + u;
            // 每个 key 单独计时：共用一个 deadline 时，
            // 前几个 key 慢一点就会把后面 key 的等待时间吃光
            long deadline = System.nanoTime() + timeout.toNanos();
            String payload = null;
            while (payload == null && System.nanoTime() < deadline) {
                payload = commands.get(key);
                if (payload == null) Thread.sleep(200);
            }

            assertNotNull(payload, "窗口指标缺失: " + key);
            assertEquals(5, mapper.readTree(payload).get("count").asInt(),
                "计数不对: " + key);
        }
    }
}

三个要点：

  • 断言具体窗口 key（metric:windowStart:dimensionValue），不要断言 :latest。后者由哪条记录最后写入不确定，重放或乱序都会改变它，验收会变成碰运气（11.1 有完整讨论）。
  • 轮询而不是固定 sleep：固定 25 秒在慢机器上会偶发失败，在快机器上又浪费时间。
  • 按 JSON 解析数值，不要 contains("\"count\":5")——序列化格式一改，断言就假失败。

死信队列验证

private static void assertDeadLetterEmpty() {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-dlq-consumer");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
        consumer.subscribe(List.of("eventd-dead-letter"));
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(3));
        assertEquals(0, records.count(), "有效数据不应进入死信队列");
    }
}

assertTrue(records.count() >= 0) 这种断言永远为真，等于没测。想验证死信链路，应该故意插一条坏数据（例如 type 为空的 JSON），再断言恰好收到 1 条。

测试速度优化

Testcontainers 启动慢，优化策略：

  1. 同测试类复用容器：@Container + static，整个类只启动一次
  2. 减少 Kafka 分区数：测试用 4 分区而非生产 12 分区
  3. 减少测试数据量：100 条以内足够覆盖聚合逻辑
  4. 并行测试：JUnit 5 的 @Execution(ExecutionMode.CONCURRENT)
@Testcontainers(parallel = true)  // 容器并行创建
@Execution(ExecutionMode.CONCURRENT)
class EventdEndToEndTest {
    // 静态共享容器 + 并发测试时，每个测试方法必须用独立的 userId / 窗口，
    // 否则会读到彼此写入的 Redis 指标
}

E2E 测试 vs 单元测试

维度E2E 测试单元测试
启动时间30-60s< 1s
外部依赖Kafka + Redis + Flink无
覆盖范围全链路单个模块
调试难度高低
运行频率CI/CD 阶段开发阶段持续运行

eventd v3 的测试策略：

  • 开发时：单元测试（TopologyTestDriver + MiniCluster）
  • 提交前：集成测试（Testcontainers E2E）
  • CI/CD：全量 E2E（每构建一次跑一次）

小结

工具用途
Testcontainers Kafka真实 Kafka 测试（KRaft，无需 Zookeeper）
Testcontainers Redis真实 Redis 测试
AdminClient在测试里显式建 Topic
Watermark 推进事件让窗口在小流量测试中关闭
@Container static复用容器，减少启动时间

第 13 章结束。下一节进入性能与故障排查：14.1 性能基线与调优 。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南