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

Question

这与Is there a "Circuit Breaker" for Spring Boot Kafka client?有关，但我仍然认为这是一个不同的问题:)

我们需要配置 Spring Boot Kafka 客户端，使其根本不尝试连接。

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

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

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

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

@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，默认情况下，它将连接到代理以创建任何 NewTopic bean。您可以将其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() {

当使用@KafkaListener 设置autoStartup = "false" 以防止消费者在上下文初始化时启动。

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