Spring Boot嵌入式Kafka无法连接

Question

我正试图为我的Kafka消费者编写一个集成测试。我已经看完了官方的参考文件，但当我开始测试时，我只看到这个无限重复：

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

我做错了什么？

我正在使用JUnit5，Spring Boot和sping-kafka和sping-kafka-test。

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

这就是我的测试类的样子：

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

}

我做错了什么？请注意，我正在使用融合模式注册表，并尝试从Avro反序列化。

我想测试的是我的消费者是否工作，如下所示：

open class SomeConsumer(private val someUseCase) {

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

Answer 1

我相信你错过了设置经纪人网址为您的测试。

文档中有一个关于如何获取该值的注释：

当embedded Kafka和embedded Zookeeper服务器由embedded Kafka Broker启动时，会出现一个名为spring的系统属性。嵌入的Kafka。brokers设置为Kafka brokers的地址和名为spring的系统属性。嵌入的动物园管理员。connect设置为Zookeeper的地址。为该属性提供了方便的常量（EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS和EmbeddedKafkaBroker.SPRING_EMBEDDED_ZOOKEEPER_CONNECT）。

（它位于junit部分的底部）

解决这个问题的一种方法是设置kafka。消费者。引导服务器在测试中设置为该值，例如。

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