如何配置 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() {

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

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