java - Spring Boot Embedded Kafka 无法连接
问题描述
我正在尝试为我的 Kafka 消费者编写一个集成测试。我已经完成了官方参考文档,但是当我开始测试时,我只看到这个重复的广告:
-2019-04-03 15:47:34.002 WARN 13120 --- [main] org.apache.kafka.clients.NetworkClient : [Consumer clientId=consumer-1, groupId=my-group] 无法连接到节点 -1成立。经纪人可能不可用。
我究竟做错了什么?
我正在使用 JUnit5、Spring Bootspring-kafka
和spring-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
我也尝试设置 aMockSchemaRegistryClient
但我得到完全相同的重复消息。(这就是我尝试设置的方式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)
}
}
解决方案
我相信您缺少为您的测试设置代理网址。
文档中有关于如何获取此值的说明:
当 EmbeddedKafkaBroker 启动嵌入式 Kafka 和嵌入式 Zookeeper 服务器时,名为 spring.embedded.kafka.brokers 的系统属性设置为 Kafka 代理的地址,名为 spring.embedded.zookeeper.connect 的系统属性设置为动物园管理员的地址。为此属性提供了方便的常量(EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS 和 EmbeddedKafkaBroker.SPRING_EMBEDDED_ZOOKEEPER_CONNECT)。
(它位于此处junit部分的底部)
解决此问题的一种方法是kafka.consumers.bootstrap-servers
在测试中设置此值,例如
spring:
kafka:
consumer:
bootstrap-servers: ${spring.embedded.kafka.brokers}
推荐阅读
- django - 无法连接 Django 和我的 react 本机应用程序... [网络请求失败]
- python - 为什么 pandas 数据框在从 excel 中读取列名后会在列名中引入额外的数字?
- android - 是否可以更改某些 Room 数据库的主键值?
- angular - 将永远不会显示的逻辑放在哪里?
- odoo - Bus.bus 在工人大于 0 后不可用
- rpmbuild - 打包 AppImage 时如何跳过精灵检查?
- r - 从另一个数据帧中减去带有条件的数据帧中的值
- typescript - TypeScript 不使用 Pick 强制返回类型
- reactjs - 在 redux 操作中应用多个过滤器
- swift - 如何在 SwiftUI 中使用 Font Awesome 作为自定义字体?