跳到主要内容

测试应用程序

ChatGPT-4o-mini 中英对照 Testing Applications

spring-kafka-test jar 包含一些有用的工具，帮助测试你的应用程序。

嵌入式 Kafka Broker​

提供了两种实现：

  • EmbeddedKafkaZKBroker - 传统实现，启动一个嵌入式 Zookeeper 实例（在使用 EmbeddedKafka 时仍然是默认选项）。

  • EmbeddedKafkaKraftBroker - 在组合控制器和代理模式下使用 Kraft 代替 Zookeeper（自 3.1 版本起）。

有几种技术可以配置代理，如以下各节所讨论的。

KafkaTestUtils​

org.springframework.kafka.test.utils.KafkaTestUtils 提供了一些静态辅助方法，用于消费记录、检索各种记录偏移量等。有关完整细节，请参阅其 Javadocs。

JUnit​

org.springframework.kafka.test.utils.KafkaTestUtils 还提供了一些静态方法来设置生产者和消费者属性。以下列表显示了这些方法的签名：

/**
 * Set up test properties for an {@code <Integer, String>} consumer.
 * @param group the group id.
 * @param autoCommit the auto commit.
 * @param embeddedKafka a {@link EmbeddedKafkaBroker} instance.
 * @return the properties.
 */
public static Map<String, Object> consumerProps(String group, String autoCommit,
                                       EmbeddedKafkaBroker embeddedKafka) { ... }

/**
 * Set up test properties for an {@code <Integer, String>} producer.
 * @param embeddedKafka a {@link EmbeddedKafkaBroker} instance.
 * @return the properties.
 */
public static Map<String, Object> producerProps(EmbeddedKafkaBroker embeddedKafka) { ... }
java
备注

从版本 2.5 开始，consumerProps 方法将 ConsumerConfig.AUTO_OFFSET_RESET_CONFIG 设置为 earliest。这是因为在大多数情况下，您希望消费者能够消费测试用例中发送的任何消息。ConsumerConfig 的默认值是 latest，这意味着在消费者启动之前，测试已经发送的消息将无法接收这些记录。要恢复到以前的行为，请在调用该方法后将属性设置为 latest。

在使用嵌入式代理时，通常最佳实践是为每个测试使用不同的主题，以防止交叉干扰。如果由于某种原因无法做到这一点，请注意，consumeFromEmbeddedTopics 方法的默认行为是在分配后将分配的分区寻址到开始位置。由于它无法访问消费者属性，因此您必须使用重载的方法，该方法接受一个 seekToEnd 布尔参数，以便寻址到结束位置而不是开始位置。

提供了一个 JUnit 4 @Rule 包装器，用于 EmbeddedKafkaZKBroker，以创建一个嵌入式 Kafka 和一个嵌入式 Zookeeper 服务器。（有关在 JUnit 5 中使用 @EmbeddedKafka 的信息，请参见 @EmbeddedKafka Annotation）。以下列表显示了这些方法的签名：

/**
 * Create embedded Kafka brokers.
 * @param count the number of brokers.
 * @param controlledShutdown passed into TestUtils.createBrokerConfig.
 * @param topics the topics to create (2 partitions per).
 */
public EmbeddedKafkaRule(int count, boolean controlledShutdown, String... topics) { ... }

/**
 *
 * Create embedded Kafka brokers.
 * @param count the number of brokers.
 * @param controlledShutdown passed into TestUtils.createBrokerConfig.
 * @param partitions partitions per topic.
 * @param topics the topics to create.
 */
public EmbeddedKafkaRule(int count, boolean controlledShutdown, int partitions, String... topics) { ... }
java
备注

EmbeddedKafkaKraftBroker 不支持 JUnit4。

EmbeddedKafkaBroker 类有一个实用方法，可以让你消费它创建的所有主题。以下示例展示了如何使用它：

Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testT", "false", embeddedKafka);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
Consumer<Integer, String> consumer = cf.createConsumer();
embeddedKafka.consumeFromAllEmbeddedTopics(consumer);
java

KafkaTestUtils 提供了一些实用方法来从消费者获取结果。以下列表显示了这些方法的签名：

/**
 * Poll the consumer, expecting a single record for the specified topic.
 * @param consumer the consumer.
 * @param topic the topic.
 * @return the record.
 * @throws org.junit.ComparisonFailure if exactly one record is not received.
 */
public static <K, V> ConsumerRecord<K, V> getSingleRecord(Consumer<K, V> consumer, String topic) { ... }

/**
 * Poll the consumer for records.
 * @param consumer the consumer.
 * @return the records.
 */
public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer) { ... }
java

以下示例展示了如何使用 KafkaTestUtils：

...
template.sendDefault(0, 2, "bar");
ConsumerRecord<Integer, String> received = KafkaTestUtils.getSingleRecord(consumer, "topic");
...
java

当 EmbeddedKafkaBroker 启动嵌入式 Kafka 和嵌入式 Zookeeper 服务器时，一个名为 spring.embedded.kafka.brokers 的系统属性被设置为 Kafka 代理的地址，一个名为 spring.embedded.zookeeper.connect 的系统属性被设置为 Zookeeper 的地址。为此属性提供了方便的常量 (EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS 和 EmbeddedKafkaBroker.SPRING_EMBEDDED_ZOOKEEPER_CONNECT)。

除了默认的 spring.embedded.kafka.brokers 系统属性外，Kafka 代理的地址可以暴露为任何任意和方便的属性。为此，可以在启动嵌入式 Kafka 之前设置 spring.embedded.kafka.brokers.property (EmbeddedKafkaBroker.BROKER_LIST_PROPERTY) 系统属性。例如，在 Spring Boot 中，期望设置 spring.kafka.bootstrap-servers 配置属性以自动配置 Kafka 客户端。因此，在使用随机端口运行嵌入式 Kafka 测试之前，我们可以将 spring.embedded.kafka.brokers.property=spring.kafka.bootstrap-servers 设置为系统属性 - EmbeddedKafkaBroker 将使用它来暴露其代理地址。现在这是该属性的默认值（从版本 3.0.10 开始）。

使用 EmbeddedKafkaBroker.brokerProperties(Map<String, String>)，您可以为 Kafka 服务器提供额外的属性。有关可能的代理属性的更多信息，请参见 Kafka Config。

配置主题​

以下示例配置创建了名为 cat 和 hat 的主题，每个主题有五个分区，名为 thing1 的主题有 10 个分区，以及名为 thing2 的主题有 15 个分区：

public class MyTests {

    @ClassRule
    private static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, false, 5, "cat", "hat");

    @Test
    public void test() {
        embeddedKafkaRule.getEmbeddedKafka()
              .addTopics(new NewTopic("thing1", 10, (short) 1), new NewTopic("thing2", 15, (short) 1));
        ...
    }

}
java

默认情况下，addTopics 在出现问题时会抛出异常（例如，添加一个已经存在的主题）。版本 2.6 添加了该方法的新版本，它返回一个 Map<String, Exception>；键是主题名称，值在成功时为 null，在失败时为 Exception。

Request error occurred:

您可以使用相同的 broker 进行多个测试类，类似于以下内容：

public final class EmbeddedKafkaHolder {

    private static EmbeddedKafkaBroker embeddedKafka = new EmbeddedKafkaZKBroker(1, false)
            .brokerListProperty("spring.kafka.bootstrap-servers");

    private static boolean started;

    public static EmbeddedKafkaBroker getEmbeddedKafka() {
        if (!started) {
            try {
                embeddedKafka.afterPropertiesSet();
            }
            catch (Exception e) {
                throw new KafkaException("Embedded broker failed to start", e);
            }
            started = true;
        }
        return embeddedKafka;
    }

    private EmbeddedKafkaHolder() {
        super();
    }

}
java

这假设了一个 Spring Boot 环境，并且嵌入式代理替代了 bootstrap servers 属性。

然后，在每个测试类中，您可以使用类似以下的内容：

static {
    EmbeddedKafkaHolder.getEmbeddedKafka().addTopics("topic1", "topic2");
}

private static final EmbeddedKafkaBroker broker = EmbeddedKafkaHolder.getEmbeddedKafka();
java

如果您没有使用 Spring Boot，可以通过 broker.getBrokersAsString() 获取引导服务器。

important

前面的示例没有提供在所有测试完成时关闭 broker 的机制。如果你在 Gradle 守护进程中运行测试，这可能会成为一个问题。在这种情况下，你不应该使用这种技术，或者你应该使用某种方法在测试完成时调用 destroy() 来关闭 EmbeddedKafkaBroker。

从版本 3.0 开始，该框架为 JUnit 平台公开了一个 GlobalEmbeddedKafkaTestExecutionListener；默认情况下是禁用的。这需要 JUnit 平台 1.8 或更高版本。此监听器的目的是为整个测试计划启动一个全局的 EmbeddedKafkaBroker，并在计划结束时停止它。要启用此监听器，从而为项目中的所有测试提供一个单一的全局嵌入式 Kafka 集群，必须通过系统属性或 JUnit 平台配置将 spring.kafka.global.embedded.enabled 属性设置为 true。此外，可以提供以下属性：

  • spring.kafka.embedded.count - 要管理的 Kafka 代理数量；

  • spring.kafka.embedded.ports - 每个 Kafka 代理启动的端口（以逗号分隔的值），如果首选随机端口则为 0；值的数量必须等于上述提到的 count；

  • spring.kafka.embedded.topics - 在启动的 Kafka 集群中创建的主题（以逗号分隔的值）；

  • spring.kafka.embedded.partitions - 为创建的主题预留的分区数量；

  • spring.kafka.embedded.broker.properties.location - 额外 Kafka 代理配置属性文件的位置；该属性的值必须遵循 Spring 资源抽象模式；

  • spring.kafka.embedded.kraft - 默认值为 false，当为 true 时，使用 EmbeddedKafkaKraftBroker 而不是 EmbeddedKafkaZKBroker。

本质上，这些属性模拟了一些 @EmbeddedKafka 属性。

有关配置属性的更多信息以及如何提供它们，请参见 JUnit 5 用户指南。例如，可以将 spring.embedded.kafka.brokers.property=my.bootstrap-servers 条目添加到测试类路径中的 junit-platform.properties 文件中。从版本 3.0.10 开始，代理默认会将其自动设置为 spring.kafka.bootstrap-servers，以便与 Spring Boot 应用程序进行测试。

备注

建议不要在单个测试套件中将全局嵌入式 Kafka 和每个测试类结合在一起。它们共享相同的系统属性，因此很可能会导致意外行为。

备注

spring-kafka-test 依赖于 junit-jupiter-api 和 junit-platform-launcher（后者用于支持全局嵌入式代理）。如果您希望使用嵌入式代理并且不使用 JUnit，您可能希望排除这些依赖。

@EmbeddedKafka 注解​

我们通常建议您将规则作为 @ClassRule 使用，以避免在测试之间启动和停止代理（并为每个测试使用不同的主题）。从版本 2.0 开始，如果您使用 Spring 的测试应用程序上下文缓存，您还可以声明一个 EmbeddedKafkaBroker bean，这样可以在多个测试类之间使用单个代理。为了方便，我们提供了一个测试类级别的注解 @EmbeddedKafka 来注册 EmbeddedKafkaBroker bean。以下示例展示了如何使用它：

@RunWith(SpringRunner.class)
@DirtiesContext
@EmbeddedKafka(partitions = 1,
         topics = {
                 KafkaStreamsTests.STREAMING_TOPIC1,
                 KafkaStreamsTests.STREAMING_TOPIC2 })
public class KafkaStreamsTests {

    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;

    @Test
    public void someTest() {
        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testGroup", "true", this.embeddedKafka);
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        ConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
        Consumer<Integer, String> consumer = cf.createConsumer();
        this.embeddedKafka.consumeFromAnEmbeddedTopic(consumer, KafkaStreamsTests.STREAMING_TOPIC2);
        ConsumerRecords<Integer, String> replies = KafkaTestUtils.getRecords(consumer);
        assertThat(replies.count()).isGreaterThanOrEqualTo(1);
    }

    @Configuration
    @EnableKafkaStreams
    public static class TestKafkaStreamsConfiguration {

        @Value("${" + EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
        private String brokerAddresses;

        @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
        public KafkaStreamsConfiguration kStreamsConfigs() {
            Map<String, Object> props = new HashMap<>();
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, "testStreams");
            props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddresses);
            return new KafkaStreamsConfiguration(props);
        }

    }

}
java

从版本 2.2.4 开始，您还可以使用 @EmbeddedKafka 注解来指定 Kafka 端口属性。

从版本 3.2 开始，将 kraft 属性设置为 true 以使用 EmbeddedKafkaKraftBroker 而不是 EmbeddedKafkaZKBroker。

以下示例设置了 @EmbeddedKafka 的 topics、brokerProperties 和 brokerPropertiesLocation 属性，以支持属性占位符解析：

@TestPropertySource(locations = "classpath:/test.properties")
@EmbeddedKafka(topics = { "any-topic", "${kafka.topics.another-topic}" },
        brokerProperties = { "log.dir=${kafka.broker.logs-dir}",
                            "listeners=PLAINTEXT://localhost:${kafka.broker.port}",
                            "auto.create.topics.enable=${kafka.broker.topics-enable:true}" },
        brokerPropertiesLocation = "classpath:/broker.properties")
java

在前面的示例中，属性占位符 ${kafka.topics.another-topic}、${kafka.broker.logs-dir} 和 ${kafka.broker.port} 是从 Spring Environment 中解析的。此外，代理属性是从 broker.properties 类路径资源中加载的，该资源由 brokerPropertiesLocation 指定。属性占位符会解析 brokerPropertiesLocation URL 以及资源中找到的任何属性占位符。由 brokerProperties 定义的属性会覆盖在 brokerPropertiesLocation 中找到的属性。

您可以在 JUnit 4 或 JUnit 5 中使用 @EmbeddedKafka 注解。

@EmbeddedKafka 注解与 JUnit5​

从版本 2.3 开始，有两种方法可以在 JUnit5 中使用 @EmbeddedKafka 注解。当与 @SpringJunitConfig 注解一起使用时，嵌入式代理会被添加到测试应用程序上下文中。您可以在类级别或方法级别将代理自动注入到测试中，以获取代理地址列表。

当 不 使用 Spring 测试上下文时，EmbdeddedKafkaCondition 会创建一个代理；该条件包含一个参数解析器，以便您可以在测试方法中访问代理。

@EmbeddedKafka
public class EmbeddedKafkaConditionTests {

    @Test
    public void test(EmbeddedKafkaBroker broker) {
        String brokerList = broker.getBrokersAsString();
        ...
    }

}
java

除非一个被注解为 @EmbeddedKafka 的类也被注解（或元注解）为 ExtendWith(SpringExtension.class)，否则将创建一个独立的代理（不在 Spring 的 TestContext 内）。@SpringJunitConfig 和 @SpringBootTest 是如此元注解的，当这两个注解中的任何一个也存在时，将使用基于上下文的代理。

important

当有 Spring 测试应用程序上下文可用时，主题和代理属性可以包含属性占位符，只要属性在某处被定义，这些占位符就会被解析。如果没有可用的 Spring 上下文，这些占位符将不会被解析。

@SpringBootTest 注解中的嵌入式代理​

Spring Initializr 现在自动将 spring-kafka-test 依赖项以测试范围添加到项目配置中。

important

如果您的应用程序在 spring-cloud-stream 中使用 Kafka 绑定器，并且您希望在测试中使用嵌入式代理，则必须移除 spring-cloud-stream-test-support 依赖，因为它会将真实的绑定器替换为测试绑定器以用于测试用例。如果您希望某些测试使用测试绑定器而某些测试使用嵌入式代理，则使用真实绑定器的测试需要通过在测试类中排除绑定器自动配置来禁用测试绑定器。以下示例展示了如何做到这一点：

@RunWith(SpringRunner.class)
@SpringBootTest(properties = "spring.autoconfigure.exclude="
    + "org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration")
public class MyApplicationTests {
    ...
}
java

在 Spring Boot 应用程序测试中，有几种方法可以使用嵌入式代理。

它们包括：

JUnit4 类规则​

以下示例展示了如何使用 JUnit4 类规则来创建一个嵌入式代理：

@RunWith(SpringRunner.class)
@SpringBootTest
public class MyApplicationTests {

    @ClassRule
    public static EmbeddedKafkaRule broker = new EmbeddedKafkaRule(1, false, "someTopic")
            .brokerListProperty("spring.kafka.bootstrap-servers");

    @Autowired
    private KafkaTemplate<String, String> template;

    @Test
    public void test() {
        ...
    }

}
java

注意，由于这是一个 Spring Boot 应用程序，我们重写了 broker list 属性以设置 Spring Boot 的属性。

@EmbeddedKafka 与 @SpringJunitConfig​

在使用 @EmbeddedKafka 和 @SpringJUnitConfig 时，建议在测试类上使用 @DirtiesContext。这是为了防止在测试套件中运行多个测试后，JVM 关闭时可能发生的竞争条件。例如，如果不使用 @DirtiesContext，EmbeddedKafkaBroker 可能会在应用程序上下文仍然需要它的资源时提前关闭。由于每个 EmbeddedKafka 测试运行都会创建自己的临时目录，当发生这种竞争条件时，会产生错误日志消息，指示它试图删除或清理的文件不再可用。添加 @DirtiesContext 将确保在每个测试后清理应用程序上下文，而不是缓存，从而使其不那么容易受到这些潜在资源竞争条件的影响。

@EmbeddedKafka 注解或 EmbeddedKafkaBroker Bean​

以下示例演示了如何使用 @EmbeddedKafka 注解来创建一个嵌入式代理：

@RunWith(SpringRunner.class)
@EmbeddedKafka(topics = "someTopic",
        bootstrapServersProperty = "spring.kafka.bootstrap-servers") // this is now the default
public class MyApplicationTests {

    @Autowired
    private KafkaTemplate<String, String> template;

    @Test
    public void test() {
        ...
    }

}
java
备注

bootstrapServersProperty 默认情况下从版本 3.0.10 开始自动设置为 spring.kafka.bootstrap-servers。

Hamcrest 匹配器​

org.springframework.kafka.test.hamcrest.KafkaMatchers 提供了以下匹配器：

/**
 * @param key the key
 * @param <K> the type.
 * @return a Matcher that matches the key in a consumer record.
 */
public static <K> Matcher<ConsumerRecord<K, ?>> hasKey(K key) { ... }

/**
 * @param value the value.
 * @param <V> the type.
 * @return a Matcher that matches the value in a consumer record.
 */
public static <V> Matcher<ConsumerRecord<?, V>> hasValue(V value) { ... }

/**
 * @param partition the partition.
 * @return a Matcher that matches the partition in a consumer record.
 */
public static Matcher<ConsumerRecord<?, ?>> hasPartition(int partition) { ... }

/**
 * Matcher testing the timestamp of a {@link ConsumerRecord} assuming the topic has been set with
 * {@link org.apache.kafka.common.record.TimestampType#CREATE_TIME CreateTime}.
 *
 * @param ts timestamp of the consumer record.
 * @return a Matcher that matches the timestamp in a consumer record.
 */
public static Matcher<ConsumerRecord<?, ?>> hasTimestamp(long ts) {
  return hasTimestamp(TimestampType.CREATE_TIME, ts);
}

/**
 * Matcher testing the timestamp of a {@link ConsumerRecord}
 * @param type timestamp type of the record
 * @param ts timestamp of the consumer record.
 * @return a Matcher that matches the timestamp in a consumer record.
 */
public static Matcher<ConsumerRecord<?, ?>> hasTimestamp(TimestampType type, long ts) {
  return new ConsumerRecordTimestampMatcher(type, ts);
}
java

AssertJ 条件​

您可以使用以下 AssertJ 条件：

/**
 * @param key the key
 * @param <K> the type.
 * @return a Condition that matches the key in a consumer record.
 */
public static <K> Condition<ConsumerRecord<K, ?>> key(K key) { ... }

/**
 * @param value the value.
 * @param <V> the type.
 * @return a Condition that matches the value in a consumer record.
 */
public static <V> Condition<ConsumerRecord<?, V>> value(V value) { ... }

/**
 * @param key the key.
 * @param value the value.
 * @param <K> the key type.
 * @param <V> the value type.
 * @return a Condition that matches the key in a consumer record.
 * @since 2.2.12
 */
public static <K, V> Condition<ConsumerRecord<K, V>> keyValue(K key, V value) { ... }

/**
 * @param partition the partition.
 * @return a Condition that matches the partition in a consumer record.
 */
public static Condition<ConsumerRecord<?, ?>> partition(int partition) { ... }

/**
 * @param value the timestamp.
 * @return a Condition that matches the timestamp value in a consumer record.
 */
public static Condition<ConsumerRecord<?, ?>> timestamp(long value) {
  return new ConsumerRecordTimestampCondition(TimestampType.CREATE_TIME, value);
}

/**
 * @param type the type of timestamp
 * @param value the timestamp.
 * @return a Condition that matches the timestamp value in a consumer record.
 */
public static Condition<ConsumerRecord<?, ?>> timestamp(TimestampType type, long value) {
  return new ConsumerRecordTimestampCondition(type, value);
}
java

示例​

以下示例汇集了本章中涵盖的大多数主题：

public class KafkaTemplateTests {

    private static final String TEMPLATE_TOPIC = "templateTopic";

    @ClassRule
    public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, TEMPLATE_TOPIC);

    @Test
    public void testTemplate() throws Exception {
        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testT", "false",
            embeddedKafka.getEmbeddedKafka());
        DefaultKafkaConsumerFactory<Integer, String> cf =
                            new DefaultKafkaConsumerFactory<>(consumerProps);
        ContainerProperties containerProperties = new ContainerProperties(TEMPLATE_TOPIC);
        KafkaMessageListenerContainer<Integer, String> container =
                            new KafkaMessageListenerContainer<>(cf, containerProperties);
        final BlockingQueue<ConsumerRecord<Integer, String>> records = new LinkedBlockingQueue<>();
        container.setupMessageListener(new MessageListener<Integer, String>() {

            @Override
            public void onMessage(ConsumerRecord<Integer, String> record) {
                System.out.println(record);
                records.add(record);
            }

        });
        container.setBeanName("templateTests");
        container.start();
        ContainerTestUtils.waitForAssignment(container,
                            embeddedKafka.getEmbeddedKafka().getPartitionsPerTopic());
        Map<String, Object> producerProps =
                            KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka());
        ProducerFactory<Integer, String> pf =
                            new DefaultKafkaProducerFactory<>(producerProps);
        KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
        template.setDefaultTopic(TEMPLATE_TOPIC);
        template.sendDefault("foo");
        assertThat(records.poll(10, TimeUnit.SECONDS), hasValue("foo"));
        template.sendDefault(0, 2, "bar");
        ConsumerRecord<Integer, String> received = records.poll(10, TimeUnit.SECONDS);
        assertThat(received, hasKey(2));
        assertThat(received, hasPartition(0));
        assertThat(received, hasValue("bar"));
        template.send(TEMPLATE_TOPIC, 0, 2, "baz");
        received = records.poll(10, TimeUnit.SECONDS);
        assertThat(received, hasKey(2));
        assertThat(received, hasPartition(0));
        assertThat(received, hasValue("baz"));
    }

}
java

前面的例子使用了 Hamcrest 匹配器。使用 AssertJ 时，最后一部分看起来如下代码：

assertThat(records.poll(10, TimeUnit.SECONDS)).has(value("foo"));
template.sendDefault(0, 2, "bar");
ConsumerRecord<Integer, String> received = records.poll(10, TimeUnit.SECONDS);
// using individual assertions
assertThat(received).has(key(2));
assertThat(received).has(value("bar"));
assertThat(received).has(partition(0));
template.send(TEMPLATE_TOPIC, 0, 2, "baz");
received = records.poll(10, TimeUnit.SECONDS);
// using allOf()
assertThat(received).has(allOf(keyValue(2, "baz"), partition(0)));
java

Mock Consumer and Producer​

kafka-clients 库提供了 MockConsumer 和 MockProducer 类用于测试目的。

如果您希望在某些测试中使用这些类与监听器容器或 KafkaTemplate，从版本 3.0.7 开始，框架现在提供了 MockConsumerFactory 和 MockProducerFactory 的实现。

这些工厂可以在监听器容器和模板中使用，替代默认工厂，这些默认工厂需要一个运行中的（或嵌入式的）代理。

这是一个简单实现返回单个消费者的示例：

@Bean
ConsumerFactory<String, String> consumerFactory() {
    MockConsumer<String, String> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
    TopicPartition topicPartition0 = new TopicPartition("topic", 0);
    List<TopicPartition> topicPartitions = Collections.singletonList(topicPartition0);
    Map<TopicPartition, Long> beginningOffsets = topicPartitions.stream().collect(Collectors
            .toMap(Function.identity(), tp -> 0L));
    consumer.updateBeginningOffsets(beginningOffsets);
    consumer.schedulePollTask(() -> {
        consumer.addRecord(
                new ConsumerRecord<>("topic", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "test1",
                        new RecordHeaders(), Optional.empty()));
        consumer.addRecord(
                new ConsumerRecord<>("topic", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "test2",
                        new RecordHeaders(), Optional.empty()));
    });
    return new MockConsumerFactory(() -> consumer);
}
java

如果您希望进行并发测试，工厂构造函数中的 Supplier lambda 需要每次创建一个新实例。

使用 MockProducerFactory 有两个构造函数；一个用于创建简单工厂，另一个用于创建支持事务的工厂。

这里是一些示例：

@Bean
ProducerFactory<String, String> nonTransFactory() {
    return new MockProducerFactory<>(() ->
            new MockProducer<>(true, new StringSerializer(), new StringSerializer()));
}

@Bean
ProducerFactory<String, String> transFactory() {
    MockProducer<String, String> mockProducer =
            new MockProducer<>(true, new StringSerializer(), new StringSerializer());
    mockProducer.initTransactions();
    return new MockProducerFactory<String, String>((tx, id) -> mockProducer, "defaultTxId");
}
java

注意在第二种情况下，lambda 是一个 BiFunction<Boolean, String>，其中第一个参数为 true 表示调用者希望使用事务性生产者；可选的第二个参数包含事务性 ID。这可以是默认值（如构造函数中提供的），或者可以由 KafkaTransactionManager（或用于本地事务的 KafkaTemplate）覆盖，如果进行了相应配置。提供事务性 ID 是为了在您希望根据该值使用不同的 MockProducer 时使用。

如果您在多线程环境中使用生产者，BiFunction 应该返回多个生产者（可能使用 ThreadLocal 进行线程绑定）。

important

事务性的 MockProducer 必须通过调用 initTransaction() 来初始化以支持事务。

当使用 MockProducer 时，如果您不想在每次发送后关闭生产者，则可以提供一个自定义的 MockProducer 实现，重写 close 方法，使其不调用超类的 close 方法。这在测试时非常方便，可以在不关闭生产者的情况下验证多次发布。

这是一个例子：

@Bean
MockProducer<String, String> mockProducer() {
    return new MockProducer<>(false, new StringSerializer(), new StringSerializer()) {
        @Override
        public void close() {

        }
    };
}

@Bean
ProducerFactory<String, String> mockProducerFactory(MockProducer<String, String> mockProducer) {
    return new MockProducerFactory<>(() -> mockProducer);
}
java