如果Kafka和Zookeeper没有运行，为什么Maven Clean安装不会完成？
为了学习Apache Kafka，我开发了一个Spring Boot应用程序，如果我向调用Kafka发送方法的控制器发送POST请求，它会向KafkaTemplate主题发送消息。我正在运行Ubuntu19.04，并在本地成功设置和安装了Kafka和Zookeeper。一切都很好。
当我关闭Zookeeper或Kafka时，问题就出现了。如果我这样做了，那么在启动时，我的应用程序的Kafka AdminClient会定期尝试查找broer，但会将此消息发送到控制台
Connection to node -1 could not be established. Broker may not be available.我实现了这里的Kafka + Zookeeper: Connection to node -1 could not be established. Broker may not be available和这里的Spring-Boot and Kafka : How to handle broker not available?建议的修复。但是如果我运行一个maven clean install，如果Zookeeper和Kafka没有运行，那么构建永远不会完成。为什么会这样?有没有一种方法可以配置应用程序，使其在启动时检查Kafka可用性，并在服务不可用时正常处理？
下面是我调用KafkaTemplate的服务类
@Autowired
public PingMessageServiceImpl(KafkaTemplate kafkaTemplate, KafkaTopicConfiguration kafkaTopicConfiguration) {
this.kafkaTemplate = kafkaTemplate;
this.kafkaTopicConfiguration = kafkaTopicConfiguration;
}
@Override
public void sendMessage(String message) {
log.info(String.format("Received following ping message %s", message));
if (!isValidPingRequest(message)) {
log.warn("Received invalid ping request");
throw new InvalidPingRequestException();
}
log.info(String.format("Sending message=[%s]", message));
ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(kafkaTopicConfiguration.getPingTopic(), message);
future.addCallback(buildListenableFutureCallback(message));
}
private boolean isValidPingRequest(String message) {
return "ping".equalsIgnoreCase(message);
}
private ListenableFutureCallback<SendResult<String, String>> buildListenableFutureCallback(String message) {
return new ListenableFutureCallback<SendResult<String, String>>() {
@Override
public void onSuccess(SendResult<String, String> result) {
log.info(String.format("Sent message=[%s] with offset=[%d]", message, result.getRecordMetadata().offset()));
}
@Override
public void onFailure(Throwable ex) {
log.info(String.format("Unable to send message=[%s] due to %s", message, ex.getMessage()));
}
};
}下面是我用来从属性文件中提取Kafka配置属性的配置类
@NotNull(message = "bootstrapAddress cannot be null")
@NotBlank(message = "bootstrapAddress cannot be blank")
private String bootstrapAddress;
@NotNull(message = "pingTopic cannot be null")
@NotBlank(message = "pingTopic cannot be blank")
private String pingTopic;
@NotNull(message = "reconnectBackoffMs cannot be null")
@NotBlank(message = "reconnectBackoffMs cannot be blank")
@Value("${kafka.reconnect.backoff.ms}")
private String reconnectBackoffMs;
@Bean
public KafkaAdmin kafkaAdmin() {
Map<String, Object> configurations = new HashMap<>();
configurations.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
configurations.put(AdminClientConfig.RECONNECT_BACKOFF_MS_CONFIG, reconnectBackoffMs);
return new KafkaAdmin(configurations);
}
@Bean
public NewTopic pingTopic() {
return new NewTopic(pingTopic, 1, (short) 1);
}
@PostConstruct
private void displayOnStartup() {
log.info(String.format("bootstrapAddress is %s", bootstrapAddress));
log.info(String.format("reconnectBackoffMs is %s", reconnectBackoffMs));
}回答 1
Stack Overflow用户
发布于 2019-07-22 00:41:04
如果在加载ApplicationContext spring kafka bean(如KafakTemplate )时有任何Spring-boot集成测试，KafkaAdmin将尝试使用yml或properties文件中指定的属性连接kafka服务器
因此，为了避免这种情况，您可以使用spring-embedded-kafka-server，这样kafka beans将在测试执行期间连接到嵌入式服务器。
或者很简单，您可以在集成测试用例中使用@MockBean注释来模拟KafakTemplate和KafkaAdmin bean
https://stackoverflow.com/questions/57134920
复制相似问题
腾讯云开发者
