Spring Boot Embedded Kafka 无法连接

Posted 2023-02-27

【中文标题】Spring Boot Embedded Kafka 无法连接

【英文标题】：Spring Boot Embedded Kafka can't connect

【发布时间】：2019-08-25 02:36:57

【问题描述】：

我正在尝试为我的 Kafka 消费者编写集成测试。我已经完成了official reference documentation，但是当我开始测试时，我只看到这个重复的广告：

-2019-04-03 15:47:34.002 WARN 13120 --- [主] org.apache.kafka.clients.NetworkClient：[消费者 clientId=consumer-1, groupId=my-group] 无法连接到节点 -1 成立。经纪人可能不可用。

我做错了什么？

我正在使用 JUnit5、Spring Boot 以及 spring-kafka 和 spring-kafka-test。

我的@Configuration 类上有@EnableKafka 注释。

这是我的测试类的样子：

@ExtendWith(SpringExtension::class)
@SpringBootTest(classes = [TestKafkaConfig::class])
@DirtiesContext
@EmbeddedKafka(
        partitions = 1,
        topics = [KafkaIntegrationTest.MY_TOPIC])
class KafkaIntegrationTest 

    @Autowired
    private lateinit var embeddedKafka: EmbeddedKafkaBroker

    @Test
    fun test() 
        val senderProps = KafkaTestUtils.senderProps(embeddedKafka.brokersAsString)
        val template = KafkaTemplate(DefaultKafkaProducerFactory<Int, String>(senderProps))
        template.defaultTopic = KafkaIntegrationTest.MY_TOPIC
        template.sendDefault("foo")
    

我的application.yml 看起来像这样：

kafka:
  consumer:
    group-id: my-group
    bootstrap-servers: $BOOTSTRAP_SERVERS:localhost:9092
    value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
    key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    properties:
      schema.registry.url: $SCHEMA_REGISTRY_URL:http://localhost:8081
      specific.avro.reader: true

我也尝试设置MockSchemaRegistryClient，但我收到完全相同的重复消息。 （这就是我尝试设置MockSchemaRegistryClient的方式）：

@TestConfiguration
@Import(TestConfig::class)
class TestKafkaConfig 

    @Autowired
    private lateinit var props: KafkaProperties

    @Bean
    fun schemaRegistryClient() = MockSchemaRegistryClient()

    @Bean
    fun kafkaAvroSerializer() = KafkaAvroSerializer(schemaRegistryClient())

    @Bean
    fun kafkaAvroDeserializer() = KafkaAvroDeserializer(schemaRegistryClient(), props.buildConsumerProperties())

    @Bean
    fun producerFactory(): ProducerFactory<*, *> = DefaultKafkaProducerFactory(
            props.buildProducerProperties(),
            StringSerializer(),
            kafkaAvroSerializer())

    @Bean
    fun consumerFactory(): ConsumerFactory<*, *> = DefaultKafkaConsumerFactory(
            props.buildConsumerProperties(),
            StringDeserializer(),
            kafkaAvroDeserializer()
    )

    @Bean
    fun kafkaListenerContainerFactory() = ConcurrentKafkaListenerContainerFactory<Any, Any>().apply 
        setConsumerFactory(consumerFactory() as ConsumerFactory<in Any, in Any>?)
    

我做错了什么？ 请注意我正在使用 Confluent Schema Registry 并尝试从 Avro 反序列化。

我要测试的是我的消费者是否工作，看起来像这样：

open class SomeConsumer(private val someUseCase) 

    @KafkaListener(topics = ["\$kafka.some-topic"])
    open fun processMessage(record: ConsumerRecord<String, SomeObject>) 
        someUseCase.call(record)
    

【参考方案1】：

我相信您没有为您的测试设置代理网址。

文档中有关于如何获取这个值的说明：

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

（位于junit部分底部here）

解决此问题的一种方法是在测试中将 kafka.consumers.bootstrap-servers 设置为此值，例如

spring:
    kafka:
        consumer:
            bootstrap-servers: $spring.embedded.kafka.brokers

【讨论】：

我的@KafkaListener 仍然无法正常工作。它被创建，但我没有收到我用KafkaTemplate 发送的消息。我的错误处理程序也没有被调用。

您不再收到Connection to node -1 could not be established. Broker may not be available. 消息吗？如果是这样，那么您至少已成功连接到嵌入式代理。

是的。我的意思是我不再得到它了。感谢您的帮助。

我注意到的另一件事是您正在嵌入式代理上创建主题KafkaIntegrationTest.MY_TOPIC，但将消息发送到template.defaultTopic = INSERTED_VARIANT_TOPIC，您是否有可能存在主题不匹配？跨度>

不，我只是忘记在示例中将其替换为MY_TOPIC。感谢您指出这一点。