如何配置 Spring Boot Kafka 客户端使其不尝试连接

Question

这与Spring Boot Kafka 客户端是否有“断路器”有关？，但我仍然认为这是一个不同的问题:)

我们需要配置 Spring Boot Kafka 客户端，以便它根本不会尝试连接。

用例是，在测试环境中，我们没有运行 Kafka，但我们仍然需要构建完整的 Spring Boot 上下文，因此根据配置文件设置此 bean 是行不通的。我们不在乎它是否没有连接，但我们需要它存在。

问题是，不成功的连接尝试大约需要 30-40 秒，我们的测试速度明显减慢。

哪个配置参数或它们的哪个组合完全禁止连接尝试，或者至少强制客户端仅尝试一次？

多次重试连接的代码是这样的：

@Bean
public KafkaAdmin.NewTopics topics() {
    return new KafkaAdmin.NewTopics(
            TopicBuilder.name("MyTopic").build()
    );
}

它反复产生此警告：

WARN ... org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:29092) could not be established. Broker may not be available.

以下代码仅尝试连接一次：

@Bean
public ReactiveKafkaConsumerTemplate<String, MyEvent> myConsumer(KafkaProperties properties) {
    return createConsumer(properties, "MyTopic", "MyConsumerGroup");
}

public <E> ReactiveKafkaConsumerTemplate<String, E> createConsumer(KafkaProperties properties, String topic, String consumerGroup) {
    final Map<String, Object> map = configureKafkaProperties(properties, consumerGroup);

    return new ReactiveKafkaConsumerTemplate<>(
        ReceiverOptions.<String, E>create(map)
            .subscription(List.of(topic)));
}

产生

WARN 7268 ... org.apache.kafka.clients.NetworkClient   : Connection to node -1 (localhost/127.0.0.1:29092) could not be established. Broker may not be available.

我也尝试过设置属性，
spring.kafka.admin.fail-fast=true
但这似乎根本没有效果。

Answer 1

Spring Boot 自动配置一个KafkaAdmin，默认情况下，它将连接到代理以创建任何NewTopicbean。您可以将其autoCreate属性设置为 false。

    /**
     * Set to false to suppress auto creation of topics during context initialization.
     * @param autoCreate boolean flag to indicate creating topics or not during context initialization
     * @see #initialize()
     */
    public void setAutoCreate(boolean autoCreate) {

编辑

要获取对 的引用KafkaAdmin，只需将其作为参数添加到任何 bean 定义中即可。

例如

@Bean
public KafkaAdmin.NewTopics topics(KafkaAdmin admin) {
    admin.setAutoCreate(false);
    return new KafkaAdmin.NewTopics(
            TopicBuilder.name("MyTopic").build()
    );
}

另见KafkaAdmin.initialize()。

    /**
     * Call this method to check/add topics; this might be needed if the broker was not
     * available when the application context was initialized, and
     * {@link #setFatalIfBrokerNotAvailable(boolean) fatalIfBrokerNotAvailable} is false,
     * or {@link #setAutoCreate(boolean) autoCreate} was set to false.
     * @return true if successful.
     * @see #setFatalIfBrokerNotAvailable(boolean)
     * @see #setAutoCreate(boolean)
     */
    public final boolean initialize() {

当使用@KafkaListenersetautoStartup = "false"来防止消费者在初始化上下文时启动。

使用反应器，只需不要订阅方法Flux返回的内容receive*()（这就是触发消费者创建的原因）。