Kafka 应用测试策略：Testcontainers 与集成测试

① 单元测试（Unit）：Mock 掉 Kafka，只测业务逻辑  → 毫秒级
② 集成测试（Integration）：真实/嵌入式 Kafka      → 秒级
③ 端到端测试（E2E）：跨服务真实链路               → 分钟级

业务逻辑（序列化、转换、校验）→ 单元测试
生产者/消费者行为、分区、事务 → 集成测试
跨服务链路、Schema 兼容       → 端到端 / 契约测试

@Test
void shouldSendOrderCreated() {
    MockProducer<String, String> mock = new MockProducer<>(
        true,                                    // autoComplete：立即视为成功
        new StringSerializer(), new StringSerializer());

    OrderService service = new OrderService(mock);
    service.createOrder("order-1", 100L);

    List<ProducerRecord<String, String>> history = mock.history();
    assertEquals(1, history.size());
    assertEquals("order-1", history.get(0).key());
    assertEquals("orders", history.get(0).topic());
}

@Test
void shouldProcessRecords() {
    MockConsumer<String, String> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
    TopicPartition tp = new TopicPartition("orders", 0);
    consumer.assign(List.of(tp));
    consumer.updateBeginningOffsets(Map.of(tp, 0L));

    consumer.addRecord(new ConsumerRecord<>("orders", 0, 0L, "k", "v"));
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

    assertEquals(1, records.count());
}

能测：序列化、业务转换、发送/提交逻辑、错误分支
不能测：真实分区分配、再平衡、事务、真实 broker 行为

@SpringBootTest
@EmbeddedKafka(partitions = 3, topics = {"orders"},
    brokerProperties = {"listeners=PLAINTEXT://localhost:9092"})
class EmbeddedKafkaTest {
    @Autowired KafkaTemplate<String, String> template;
    // ...
}

① 镜像版本 == 生产版本（可精确锁定 tag）
② 每个测试类可独立容器，隔离彻底
③ 不污染 JVM（Embedded 的静态状态常致测试互相干扰）
④ 与 CI 天然契合（Docker 是标准环境）

已有大量 @EmbeddedKafka 测试、不想引入 Docker 依赖
单元-集成边界测试、对启动速度极敏感

<dependency>
  <groupId>org.testcontainers</groupId>
  <artifactId>kafka</artifactId>
  <version>1.20.1</version>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.testcontainers</groupId>
  <artifactId>junit-jupiter</artifactId>
  <version>1.20.1</version>
  <scope>test</scope>
</dependency>

class KafkaContainerTest {
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.6.0"));

    @BeforeAll
    static void start() { kafka.start(); }

    @AfterAll
    static void stop() { kafka.stop(); }

    @Test
    void shouldProduceAndConsume() {
        String bootstrap = kafka.getBootstrapServers();
        // 用 bootstrap 建 Producer/Consumer，跑真实收发
    }
}

@Testcontainers
class KafkaIT {
    @Container
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.6.0"));

    @Test
    void shouldProduceAndConsume() {
        try (Producer<String, String> p = producer(kafka.getBootstrapServers())) {
            p.send(new ProducerRecord<>("orders", "k", "v"));
        }
        // 消费并断言
    }
}

@Testcontainers
class SchemaRegistryIT {
    static Network net = Network.newNetwork();

    @Container
    static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.6.0"))
        .withNetwork(net).withNetworkAliases("kafka");

    @Container
    static GenericContainer<?> registry = new GenericContainer<>("confluentinc/cp-schema-registry:7.6.0")
        .withNetwork(net)
        .withExposedPorts(8081)
        .withEnv("SCHEMA_REGISTRY_HOST_NAME", "schema-registry")
        .withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "PLAINTEXT://kafka:9092")
        .dependsOn(kafka);
}

public abstract class KafkaTestBase {
    static final KafkaContainer KAFKA = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.6.0"));

    static {
        KAFKA.start();   // 全测试套件只启一次
    }
}

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"orders"})
class OrderListenerTest {
    @Autowired KafkaTemplate<String, String> template;
    @Autowired OrderListener listener;

    @Test
    void shouldReceive() {
        template.send("orders", "k", "v");
        // 用 Awaitility 等待异步消费
        await().atMost(5, SECONDS).until(() -> listener.received().size() == 1);
    }
}

@SpringBootTest
@Testcontainers
class OrderIntegrationTest {
    @Container
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.6.0"));

    @DynamicPropertySource
    static void props(DynamicPropertyRegistry r) {
        r.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
    }
}

// 错误：直接断言，消费还没发生
template.send("orders", "k", "v");
assertEquals(1, listener.received().size());   // 大概率失败

// 正确：Awaitility 轮询等待
await().atMost(Duration.ofSeconds(10))
       .pollInterval(Duration.ofMillis(100))
       .untilAsserted(() -> assertEquals(1, listener.received().size()));

Producer（服务 A）→ Kafka → Consumer（服务 B）→ 结果断言
覆盖：序列化兼容、分区、消费组、跨服务契约
代价：慢（分钟级）、脆（依赖多服务）

@Test
void schemaShouldBeBackwardCompatible() {
    SchemaRegistryClient client = new MockSchemaRegistryClient();
    // 注册 v1，再注册 v2，断言兼容
    assertTrue(client.testCompatibility("orders-value", v2Schema).isCompatible());
}

# GitHub Actions 片段
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-java@v4
        with: { java-version: '17' }
      - run: ./mvnw verify        # Testcontainers 需 Docker