测试应用程序

spring-kafka-testjar 包含一些有用的实用程序,可帮助测试您的应用程序。spring-doc.cadn.net.cn

嵌入式 Kafka 代理

提供了两种实现:spring-doc.cadn.net.cn

  • EmbeddedKafkaZKBroker- legacy 实现,用于启动嵌入式Zookeeper实例(在使用EmbeddedKafka).spring-doc.cadn.net.cn

  • EmbeddedKafkaKraftBroker-使用Kraft而不是Zookeeper在 controller 和 broker 组合模式下(自 3.1 起)。spring-doc.cadn.net.cn

有几种技术可用于配置代理,如以下各节所述。spring-doc.cadn.net.cn

KafkaTestUtils

org.springframework.kafka.test.utils.KafkaTestUtils提供了许多 static 帮助程序方法来使用记录、检索各种记录偏移量等。 有关完整详细信息,请参阅其 Javadocsspring-doc.cadn.net.cn

JUnit

org.springframework.kafka.test.utils.KafkaTestUtils还提供了一些静态方法来设置 producer 和 consumer 属性。 下面的清单显示了这些方法签名:spring-doc.cadn.net.cn

/**
 * 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) { ... }

从版本 2.5 开始,consumerPropsmethod 设置ConsumerConfig.AUTO_OFFSET_RESET_CONFIGearliest. 这是因为,在大多数情况下,您希望使用者使用测试用例中发送的任何消息。 这ConsumerConfig默认值为latest这意味着测试在 Consumer 启动之前已经发送的消息将不会收到这些记录。 要恢复到之前的行为,请将该属性设置为latest调用该方法后。spring-doc.cadn.net.cn

使用嵌入式代理时,通常最佳做法是为每个测试使用不同的主题,以防止串扰。 如果由于某种原因无法做到这一点,请注意,consumeFromEmbeddedTopics方法的默认行为是在分配后将分配的分区查找到开头。 由于它无权访问使用者属性,因此您必须使用采用seekToEndboolean 参数查找到末尾而不是开头。spring-doc.cadn.net.cn

一个 JUnit 4@Rulewrapper 的EmbeddedKafkaZKBroker用于创建嵌入式 Kafka 和嵌入式 Zookeeper 服务器。 (@EmbeddedKafka有关使用@EmbeddedKafka使用 JUnit 5)。 下面的清单显示了这些方法的签名:spring-doc.cadn.net.cn

/**
 * 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) { ... }
EmbeddedKafkaKraftBroker在 JUnit4 中不受支持。

EmbeddedKafkaBrokerclass 有一个 Utility 方法,允许您使用它创建的所有主题。 以下示例演示如何使用它:spring-doc.cadn.net.cn

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);

KafkaTestUtils具有一些实用程序方法,用于从 Consumer 那里获取结果。 下面的清单显示了这些方法签名:spring-doc.cadn.net.cn

/**
 * 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) { ... }

以下示例演示如何使用KafkaTestUtils:spring-doc.cadn.net.cn

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

当嵌入式 Kafka 和嵌入式 Zookeeper 服务器由EmbeddedKafkaBroker中,名为spring.embedded.kafka.brokers设置为 Kafka 代理的地址,并且名为spring.embedded.zookeeper.connect设置为 Zookeeper 的地址。 方便的常数 (EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERSEmbeddedKafkaBroker.SPRING_EMBEDDED_ZOOKEEPER_CONNECT) 为此属性提供。spring-doc.cadn.net.cn

而不是 defaultspring.embedded.kafka.brokerssystem 属性,Kafka broker 的地址可以暴露给任何任意且方便的属性。 为此,一个spring.embedded.kafka.brokers.property (EmbeddedKafkaBroker.BROKER_LIST_PROPERTY) 系统属性。 例如,使用 Spring Boot 时,使用spring.kafka.bootstrap-serversconfiguration 属性应分别为自动配置 Kafka 客户端设置。 因此,在随机端口上使用嵌入式 Kafka 运行测试之前,我们可以设置spring.embedded.kafka.brokers.property=spring.kafka.bootstrap-servers作为系统属性,并将EmbeddedKafkaBroker将使用它来公开其代理地址。 现在,这是此属性的默认值(从版本 3.0.10 开始)。spring-doc.cadn.net.cn

使用EmbeddedKafkaBroker.brokerProperties(Map<String, String>)中,您可以为 Kafka 服务器提供其他属性。 有关可能的代理属性的更多信息,请参阅 Kafka Configspring-doc.cadn.net.cn

配置主题

以下示例配置创建名为cathat有 5 个分区时,一个名为thing1具有 10 个分区,并且有一个名为thing2有 15 个分区:spring-doc.cadn.net.cn

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));
        ...
    }

}

默认情况下,addTopics将在出现问题时抛出异常(例如添加已存在的 Topic)。 版本 2.6 添加了该方法的新版本,该方法返回Map<String, Exception>;key 是 topic name,值为null以获得成功,或者Exception失败。spring-doc.cadn.net.cn

对多个测试类使用相同的代理

您可以将同一个 broker 用于多个测试类,类似于以下内容:spring-doc.cadn.net.cn

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();
    }

}

这假定 Spring Boot 环境,并且嵌入式代理替换了 bootstrap servers 属性。spring-doc.cadn.net.cn

然后,在每个测试类中,您可以使用类似于以下内容的内容:spring-doc.cadn.net.cn

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

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

如果您不使用 Spring Boot,则可以使用broker.getBrokersAsString().spring-doc.cadn.net.cn

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

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

  • spring.kafka.embedded.count- 要管理的 Kafka 代理数量;spring-doc.cadn.net.cn

  • spring.kafka.embedded.ports- 每个 Kafka 代理启动的端口(逗号分隔值),0如果首选 random 端口;值的数量必须等于count上述;spring-doc.cadn.net.cn

  • spring.kafka.embedded.topics- 要在启动的 Kafka 集群中创建的主题(逗号分隔值);spring-doc.cadn.net.cn

  • spring.kafka.embedded.partitions- 要为创建的主题预置的分区数;spring-doc.cadn.net.cn

  • spring.kafka.embedded.broker.properties.location- 其他 Kafka 代理配置属性的文件位置;此属性的值必须遵循 Spring 资源抽象模式;spring-doc.cadn.net.cn

  • spring.kafka.embedded.kraft- default false,当 true 时,使用EmbeddedKafkaKraftBroker而不是EmbeddedKafkaZKBroker.spring-doc.cadn.net.cn

从本质上讲,这些属性模拟了一些@EmbeddedKafka属性。spring-doc.cadn.net.cn

有关配置属性以及如何提供这些属性的更多信息,请参阅 JUnit 5 用户指南。 例如,spring.embedded.kafka.brokers.property=my.bootstrap-servers条目可以添加到junit-platform.properties文件。 从版本 3.0.10 开始,代理会自动将其设置为spring.kafka.bootstrap-servers,用于使用 Spring Boot 应用程序进行测试。spring-doc.cadn.net.cn

建议不要将全局嵌入式 Kafka 和 per-test 类合并到单个测试套件中。 它们共享相同的系统属性,因此很可能会导致意外行为。
spring-kafka-test具有传递依赖项junit-jupiter-apijunit-platform-launcher(后者支持 Global Embedded Broker)。 如果您希望使用嵌入式代理但不使用 JUnit,则可能希望排除这些依赖项。

@EmbeddedKafka注解

我们通常建议您将规则用作@ClassRule以避免在测试之间启动和停止代理(并为每个测试使用不同的主题)。 从版本 2.0 开始,如果使用 Spring 的测试应用程序上下文缓存,还可以声明EmbeddedKafkaBrokerbean,因此单个代理可以跨多个测试类使用。 为方便起见,我们提供了一个名为@EmbeddedKafka要注册EmbeddedKafkaBroker豆。 以下示例演示如何使用它:spring-doc.cadn.net.cn

@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 KafkaStreamsConfiguration {

        @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);
        }

    }

}

从版本 2.2.4 开始,您还可以使用@EmbeddedKafka注解指定 Kafka ports 属性。spring-doc.cadn.net.cn

从版本 3.2 开始,将kraftproperty 设置为true要使用EmbeddedKafkaKraftBroker而不是EmbeddedKafkaZKBroker.spring-doc.cadn.net.cn

以下示例将topics,brokerPropertiesbrokerPropertiesLocation的属性@EmbeddedKafka支持属性占位符解析:spring-doc.cadn.net.cn

@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")

在前面的示例中,属性 placeholders${kafka.topics.another-topic},${kafka.broker.logs-dir}${kafka.broker.port}从 Spring 解析Environment. 此外,代理属性是从broker.propertiesclasspath 资源由brokerPropertiesLocation. 属性占位符为brokerPropertiesLocationURL 以及在资源中找到的任何属性占位符。 属性定义brokerProperties覆盖 中找到的属性brokerPropertiesLocation.spring-doc.cadn.net.cn

您可以使用@EmbeddedKafka使用 JUnit 4 或 JUnit 5 进行注释。spring-doc.cadn.net.cn

@EmbeddedKafka使用 JUnit5 进行注释

从版本 2.3 开始,有两种方法可以使用@EmbeddedKafka注释。 当与@SpringJunitConfig注解中,嵌入式代理将添加到 Test Application 上下文中。 您可以在类或方法级别将代理自动连接到测试中,以获取代理地址列表。spring-doc.cadn.net.cn

不使用 spring 测试上下文时,EmbdeddedKafkaCondition创建代理;该条件包括一个参数解析器,因此您可以在测试方法中访问代理。spring-doc.cadn.net.cn

@EmbeddedKafka
public class EmbeddedKafkaConditionTests {

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

}

将创建一个独立的代理(在 Spring 的 TestContext 之外),除非有一个带 Comments 的类@EmbeddedKafka也被注释(或元注释)为ExtendWith(SpringExtension.class).@SpringJunitConfig@SpringBootTest是元注释的,并且当这些注释中的任何一个也存在时,将使用基于上下文的代理。spring-doc.cadn.net.cn

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

嵌入式代理@SpringBootTest附注

Spring Initializr 现在会自动添加spring-kafka-test对项目配置的 test 作用域中的依赖项。spring-doc.cadn.net.cn

如果您的应用程序在spring-cloud-stream如果要使用嵌入式代理进行测试,则必须删除spring-cloud-stream-test-support依赖项,因为它将真正的 Binder 替换为测试用例的测试 Binder。 如果您希望某些测试使用测试 Binder,而一些测试使用嵌入式代理,则使用真实 Binder 的测试需要通过在测试类中排除 Binder auto 配置来禁用测试 Binder。 以下示例显示了如何执行此作:spring-doc.cadn.net.cn

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

在 Spring Boot 应用程序测试中,有几种方法可以使用嵌入式代理。spring-doc.cadn.net.cn

他们包括:spring-doc.cadn.net.cn

JUnit4 类规则

以下示例说明如何使用 JUnit4 类规则创建嵌入式代理:spring-doc.cadn.net.cn

@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() {
        ...
    }

}

请注意,由于这是一个 Spring Boot 应用程序,因此我们覆盖 broker list 属性以设置 Spring Boot 的属性。spring-doc.cadn.net.cn

@EmbeddedKafka@SpringJunitConfig

使用@EmbeddedKafka@SpringJUnitConfig,建议使用@DirtiesContext在 test 类上。 这是为了防止在测试套件中运行多个测试后,在 JVM 关闭期间发生潜在的争用条件。 例如,如果不使用@DirtiesContextEmbeddedKafkaBroker可能会提前关闭,而应用程序上下文仍然需要来自它的资源。 由于每个EmbeddedKafkatest-runs 创建自己的临时目录,当这种争用条件发生时,它将产生错误日志消息,指示它尝试删除或清理的文件不再可用。 添加@DirtiesContext将确保 application context 在每次测试后被清理而不是缓存,使其不易受到此类潜在资源争用条件的影响。spring-doc.cadn.net.cn

@EmbeddedKafkaAnnotation 或EmbeddedKafkaBroker

以下示例演示如何使用@EmbeddedKafkaAnnotation 创建嵌入式代理:spring-doc.cadn.net.cn

@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() {
        ...
    }

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

Hamcrest 匹配器

org.springframework.kafka.test.hamcrest.KafkaMatchers提供以下匹配器:spring-doc.cadn.net.cn

/**
 * @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);
}

AssertJ 条件

您可以使用以下 AssertJ 条件:spring-doc.cadn.net.cn

/**
 * @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);
}

以下示例汇集了本章中介绍的大部分主题:spring-doc.cadn.net.cn

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"));
    }

}

前面的示例使用 Hamcrest 匹配程序。 跟AssertJ,最后一部分类似于以下代码:spring-doc.cadn.net.cn

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)));

模拟 Consumer 和 Producer

kafka-clientslibrary 提供MockConsumerMockProducer类进行测试。spring-doc.cadn.net.cn

如果您希望在某些测试中使用这些类,并且带有侦听器容器或KafkaTemplate分别从 3.0.7 版本开始,框架现在提供MockConsumerFactoryMockProducerFactory实现。spring-doc.cadn.net.cn

这些工厂可以在侦听器容器和模板中使用,而不是默认工厂,后者需要运行(或嵌入)代理。spring-doc.cadn.net.cn

下面是一个返回单个使用者的简单实现示例:spring-doc.cadn.net.cn

@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);
}

如果您希望使用并发进行测试,则Supplierlambda 的构造函数中,每次都需要创建一个新实例。spring-doc.cadn.net.cn

使用MockProducerFactory,有两个构造函数;一个用于创建简单工厂,另一个用于创建支持事务的工厂。spring-doc.cadn.net.cn

以下是示例:spring-doc.cadn.net.cn

@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");
}

请注意,在第二种情况下,lambda 是BiFunction<Boolean, String>其中,如果调用方需要事务性生产者,则第一个参数为 true;可选的第二个参数包含事务 ID。 这可以是默认值(如构造函数中提供),也可以被KafkaTransactionManager(或KafkaTemplate对于本地事务),如果已配置。 如果您希望使用不同的 ID,则会提供事务 IDMockProducer基于此值。spring-doc.cadn.net.cn

如果您在多线程环境中使用创建器,则BiFunction应返回多个生产者(可能使用ThreadLocal).spring-doc.cadn.net.cn

事务MockProducer必须通过调用initTransaction().

使用MockProducer,如果您不想在每次发送后关闭 Producer,则可以提供自定义MockProducer实现,该实现会覆盖close方法,该方法不调用close方法。 这在验证同一生产者上的多个发布而不关闭它时,便于进行测试。spring-doc.cadn.net.cn

下面是一个示例:spring-doc.cadn.net.cn

@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);
}

APP信息