java-Spring Boot Embedded Kafka无法连接

2019-11-08 06:04:05 阅读：415 来源： 互联网

标签：confluent spring-boot apache-kafka kotlin java

我正在尝试为我的Kafka使用者编写集成测试.我已经遵循了official reference documentation,但是当我开始测试时,我只会看到这个重复的广告无限：

-2019-04-03 15:47:34.002 WARN 13120 — [ main] org.apache.kafka.clients.NetworkClient : [Consumer
clientId=consumer-1, groupId=my-group] Connection to node -1 could not
be established. Broker may not be available.

我究竟做错了什么？

我正在使用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)
    }
}

解决方法:

我相信您缺少为测试设置代理URL.

在文档中有关于如何获取此值的注释：

When the embedded Kafka and embedded Zookeeper server are started by the EmbeddedKafkaBroker, a system property named spring.embedded.kafka.brokers is set to the address of the Kafka brokers and a system property named spring.embedded.zookeeper.connect is set to the address of Zookeeper. Convenient constants (EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS and EmbeddedKafkaBroker.SPRING_EMBEDDED_ZOOKEEPER_CONNECT) are provided for this property.

(位于junit部分here的底部)

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

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

标签：confluent,spring-boot,apache-kafka,kotlin,java
来源： https://codeday.me/bug/20191108/2005901.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

java-Spring Boot Embedded Kafka无法连接