Spring kafka单元测试监听器未订阅主题

Question

我有一个用kafka探索spring的示例项目(在这里找到(。我有一个监听器订阅了主题我的测试主题上游，它只会找到消息和密钥，并将其发布到另一个主题我的测试话题下游。我试过这是本地kafka(docker compose文件在那里(，它很有效。

现在我正在尝试使用嵌入式kafka服务器为此编写一个测试。在测试中，我有一个嵌入式服务器正在启动(TestContext.java(，它应该在测试之前启动(覆盖junitbeforeAll。(。

private static EmbeddedKafkaBroker kafka() {
EmbeddedKafkaBroker kafkaEmbedded =
new EmbeddedKafkaBroker(
3,
false,
1,
"my-test-topic-upstream", "my-test-topic-downstream");
Map<String, String> brokerProperties = new HashMap<>();
brokerProperties.put("default.replication.factor", "1");
brokerProperties.put("offsets.topic.replication.factor", "1");
brokerProperties.put("group.initial.rebalance.delay.ms", "3000");
kafkaEmbedded.brokerProperties(brokerProperties);
try {
kafkaEmbedded.afterPropertiesSet();
} catch (Exception e) {
throw new RuntimeException(e);
}
return kafkaEmbedded;
}

然后我创建一个制作人(TickProducer(，并发布一条消息到我希望听众能够使用的主题。

public TickProducer(String brokers) {
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
producer = new KafkaProducer<>(props);
}
public RecordMetadata publishTick(String brand)
throws ExecutionException, InterruptedException {
return publish(TOPIC, brand, Instant.now().toString());
}
private RecordMetadata publish(String topic, String key, String value)
throws ExecutionException, InterruptedException {
final RecordMetadata recordMetadata;
recordMetadata = producer.send(new ProducerRecord<>(topic, key, value)).get();
producer.flush();
return recordMetadata;
}

我看到下面的日志消息继续记录。

11:32:35.745 [main] WARN  o.apache.kafka.clients.NetworkClient - [Consumer clientId=consumer-1, groupId=my-test-group] Connection to node -1 could not be established. Broker may not be available.

最终失败

11:36:52.774 [main] ERROR o.s.boot.SpringApplication - Application run failed
org.springframework.context.ApplicationContextException: Failed to start bean 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'; nested exception is org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata

这里有什么提示吗？

Answer 1

查看INFO日志ConsumerConfig，查看他试图连接的位置(将其与ProducerConfig进行比较(。我怀疑您还没有更新spring-bootbootstrap-servers属性以指向嵌入式代理。

参见

/**
* Set the system property with this name to the list of broker addresses.
* @param brokerListProperty the brokerListProperty to set
* @return this broker.
* @since 2.3
*/
public EmbeddedKafkaBroker brokerListProperty(String brokerListProperty) {
this.brokerListProperty = brokerListProperty;
return this;
}

将其设置为spring.kafka.bootstrap-servers，然后将使用它而不是SPRING_EMBEDDED_KAFKA_BROKERS。

顺便说一句，使用@EmbeddedKafka注释通常比自己实例化服务器更容易。