12.3 v3 验收

# Compose v2 会等待 healthcheck 通过，比 sleep 30 可靠
docker compose -f eventd-v3/docker-compose.yml up -d --wait

# 组件健康
docker compose -f eventd-v3/docker-compose.yml ps

# Flink 作业状态
curl -s localhost:8081/jobs/overview | jq '.jobs[] | {name, state}'

# 清空上一轮遗留的 topic 数据与 Redis 指标，避免旧窗口干扰

@Test
void testEndToEndPipeline() throws Exception {
    List<Event> events = generateEvents(100, "acceptance-e2e");
    events.forEach(e -> sendToKafka(e, "eventd-raw"));

    // 轮询到 100 条或超时，不要用固定 sleep 猜处理时间
    List<Event> cleaned = awaitEvents("eventd-cleaned", 100, Duration.ofSeconds(30));
    assertEquals(100, cleaned.size(), "清洗输出数量不符");

    // 窗口 key 只有在 Watermark 越过窗口结束、窗口触发后才存在。
    // 这里补一个事件时间更大的哨兵，否则用例 1 会一直等不到 key
    // （哨兵同样要覆盖所有活跃分区，原因见用例 2 的第 3 条）
    long lastEventTime = events.stream()
        .mapToLong(Event::eventTime)
        .max()
        .orElseThrow();
    sendWatermarkAdvance(lastEventTime + Duration.ofMinutes(2).toMillis());

    long windowStart = windowStartOf(events.get(0).eventTime());
    awaitWindow("eventd:metrics:order-count", windowStart, Duration.ofSeconds(30));

    String countKey = "eventd:metrics:order-count:" + windowStart + ":acceptance-e2e";
    assertTrue(redis().exists(countKey) > 0, "Redis 缺少验收用户的窗口计数: " + countKey);

    // SSE 是无限流，只能用 ofLines 读一帧，绝不能用 ofString
    HttpRequest request = HttpRequest.newBuilder()
        .uri(URI.create("http://localhost:8083/api/stream/metrics"))
        .header("Accept", "text/event-stream")
        .timeout(Duration.ofSeconds(5))
        .GET()
        .build();

    HttpResponse<Stream<String>> response = HttpClient.newHttpClient()
        .send(request, HttpResponse.BodyHandlers.ofLines());
    assertEquals(200, response.statusCode());

    try (Stream<String> lines = response.body()) {
        String firstData = CompletableFuture
            .supplyAsync(() -> lines.filter(l -> l.startsWith("data:")).findFirst().orElse(null))
            .get(5, TimeUnit.SECONDS);
        assertNotNull(firstData, "5 秒内未收到 SSE data 帧");
    }
}

@Test
void testMetricsAccuracy() throws Exception {
    long base = 1_700_000_000_000L;
    Map<String, Integer> expected = new LinkedHashMap<>();

    for (int u = 0; u < 10; u++) {
        String userId = "u" + u;
        expected.put(userId, 5);
        for (int i = 0; i < 5; i++) {
            sendToKafka(Event.builder()
                .eventId("acc-" + userId + "-" + i)
                .type("ORDER")
                .userId(userId)
                .amount(100.0 + i)
                .symbol("BTC")
                .eventTime(base + i * 1000L)
                .build(), "eventd-raw");
        }
    }

    // Kafka 是无界源，不会自动发 MAX_WATERMARK；
    // 用一条 eventTime 更大的事件把 Watermark 推过窗口结束。
    sendWatermarkAdvance(base + Duration.ofMinutes(2).toMillis());

    long windowStart = windowStartOf(base);
    awaitWindow("eventd:metrics:order-count", windowStart, Duration.ofSeconds(30));

    for (Map.Entry<String, Integer> e : expected.entrySet()) {
        String key = "eventd:metrics:order-count:" + windowStart + ":" + e.getKey();
        String json = redis().get(key);
        assertNotNull(json, "缺少用户窗口计数: " + key);
        assertEquals(e.getValue().intValue(),
            JsonMapper.parse(json).get("count").asInt(),
            "用户计数不符: " + key);
    }
}

@Test
void testLatencyBudget() throws Exception {
    long sentAt = System.currentTimeMillis();
    String userId = "latency-user";
    sendToKafka(Event.builder()
        .eventId("latency-marker-" + sentAt)
        .type("ORDER")
        .userId(userId)
        .amount(1.0)
        .symbol("TEST")
        .eventTime(sentAt)
        .build(), "eventd-raw");

    // 指标是事件时间窗口聚合，单条事件不会自己关闭窗口：
    // 必须补一条事件时间更晚的 pumper 把 Watermark 推过窗口结束，否则永远等不到
    sendWatermarkAdvance(sentAt + Duration.ofMinutes(2).toMillis());

    String key = "eventd:metrics:order-count:" + windowStartOf(sentAt) + ":" + userId;
    Long observedAt = null;
    for (int i = 0; i < 50 && observedAt == null; i++) {
        Thread.sleep(200);
        if (redis().exists(key) > 0) {
            observedAt = System.currentTimeMillis();
        }
    }

    assertNotNull(observedAt, "10 秒内未观测到标记事件，延迟无法判定");
    long latencyMs = observedAt - sentAt;
    assertTrue(latencyMs < 10_000, "端到端延迟超预算: " + latencyMs + "ms");
}

@Test
void testFailureRecovery() throws Exception {
    List<Event> before = generateEvents(20, "recovery-before");
    before.forEach(e -> sendToKafka(e, "eventd-raw"));
    awaitEvents("eventd-cleaned", 20, Duration.ofSeconds(30));

    // 1. 用 Savepoint 建立显式恢复点。Checkpoint 是否保留取决于
    //    externalized 配置和清理策略，不能假设 cancel 后一定还有
    String savepoint = triggerSavepointAndWait(jobId(), Duration.ofMinutes(2));
    cancelJob(jobId());

    // 2. 作业停机期间继续生产：这些事件只可能由恢复后的作业处理
    List<Event> after = generateEvents(20, "recovery-after");
    after.forEach(e -> sendToKafka(e, "eventd-raw"));

    // 3. 从 Savepoint 重新提交
    submitFromSavepoint(savepoint);
    awaitCondition(() -> "RUNNING".equals(jobState(jobId())), Duration.ofMinutes(2));

    // 4. 两批共 40 个 eventId，每个都必须恰好处理一次
    Set<String> expected = Stream.concat(before.stream(), after.stream())
        .map(Event::eventId)
        .collect(Collectors.toSet());
    Map<String, Long> processed = awaitProcessedCounts(
        "eventd:metrics:processed", 40, Duration.ofSeconds(30));
    assertEquals(expected, processed.keySet(), "恢复后事件集合不一致（有丢失）");
    assertTrue(processed.values().stream().allMatch(c -> c == 1L),
        "存在重复处理: " + processed.entrySet().stream()
            .filter(e -> e.getValue() > 1L).toList());
}

@Test
void testBackpressureSmoke() throws Exception {
    int total = 10_000;
    ExecutorService sender = Executors.newSingleThreadExecutor();
    try {
        Future<?> producing = sender.submit(() -> {
            try (RateLimitedProducer producer = new RateLimitedProducer(1_000)) { // 1000 条/秒
                for (int i = 0; i < total; i++) {
                    producer.send(generateSingleEvent(i), "eventd-raw");
                }
            }
            return null;
        });

        // 发送进行中就检查：等发完再看「峰值期间」等于没看
        awaitCondition(() -> "RUNNING".equals(jobState(jobId())), Duration.ofMinutes(1));
        awaitCondition(() -> consecutiveCheckpointFailures(jobId()) == 0, Duration.ofMinutes(1));

        producing.get(Duration.ofMinutes(2).toMillis(), TimeUnit.MILLISECONDS);
    } finally {
        sender.shutdownNow();
    }

    // 停止生产后，lag 必须回落到阈值以下
    awaitCondition(() -> totalLag("eventd-analytics") < 1_000, Duration.ofMinutes(5));
}

eventd v3 验收记录
==================
日期:
执行人:
代码 commit:
环境: <机器 / JDK / Kafka 分区 / Flink Slot / Redis>

用例 1 端到端可运行: <PASS / FAIL>  证据: <日志或截图路径>
用例 2 指标正确性:   <PASS / FAIL>  证据: <输入清单 + 窗口键值>
用例 3 延迟预算:     <PASS / FAIL>  证据: <单样本 __ ms 或 p99 = __ ms（样本数 __），预算 __ ms>
用例 4 故障恢复:     <PASS / FAIL>  证据: <eventId 集合差异 = 0>
用例 5 反压冒烟:     <PASS / FAIL>  证据: <峰值 lag / 恢复时间>

结论: <PASS / FAIL>
未执行或证据缺失的用例，一律记 FAIL，不得记 PASS。