java-Spring Boot Embedded Kafka无法连接
作者:互联网
我正在尝试为我的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