0ct0pu5/kafka-ui
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

AdminClient creation error handling fix

Browse source 
1. AdminClient error logging added (#1136)
2. MetricsService error handling fixed
This commit is contained in:
Ilya Kuramshin 2021-11-29 12:44:49 +03:00 • committed by GitHub
parent d674808f3c
commit 4d3db7a7de
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 10 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/AdminClientServiceImpl.java
View file

@ -38,7 +38,10 @@ public class AdminClientServiceImpl implements AdminClientService, Closeable {
.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.getBootstrapServers());
properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, clientTimeout);
return AdminClient.create(properties);
}).flatMap(ReactiveAdminClient::create);
})
.flatMap(ReactiveAdminClient::create)
.onErrorMap(th -> new IllegalStateException(
"Error while creating AdminClient for Cluster " + cluster.getName(), th));
}
@Override

12
kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/MetricsService.java
View file

@ -31,7 +31,7 @@ public class MetricsService {
private Mono<MetricsCache.Metrics> getMetrics(KafkaCluster cluster) {
return adminClientService.get(cluster).flatMap(ac ->
ac.describeCluster().flatMap(description ->
ac.describeCluster().flatMap(description ->
Mono.zip(
List.of(
jmxClusterUtil.getBrokerMetrics(cluster, description.getNodes()),
@ -52,11 +52,11 @@ public class MetricsService {
.topicConfigs((Map<String, List<ConfigEntry>>) results[4])
.topicDescriptions((Map<String, TopicDescription>) results[5])
.build()
))
.doOnError(e ->
log.error("Failed to collect cluster {} info", cluster.getName(), e))
.onErrorResume(
e -> Mono.just(MetricsCache.empty().toBuilder().lastKafkaException(e).build())));
)))
.doOnError(e ->
log.error("Failed to collect cluster {} info", cluster.getName(), e))
.onErrorResume(
e -> Mono.just(MetricsCache.empty().toBuilder().lastKafkaException(e).build()));
}
private Mono<InternalLogDirStats> getLogDirInfo(KafkaCluster cluster, ReactiveAdminClient c) {