Skip to content

Commit

Permalink
Create Kafka Admin Client outside of the Parallel scheduler thread po…
Browse files Browse the repository at this point in the history
…ol (#7)

* Create Kafka Admin Client outside of the Parallel scheduler thread pool
  • Loading branch information
tnewman-at-gm authored Jan 15, 2025
1 parent ed49499 commit 5ebdd2d
Showing 1 changed file with 3 additions and 2 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import lombok.extern.slf4j.Slf4j;
Expand Down Expand Up @@ -40,7 +41,7 @@ public Mono<ReactiveAdminClient> get(KafkaCluster cluster) {
}

private Mono<ReactiveAdminClient> createAdminClient(KafkaCluster cluster) {
return Mono.fromSupplier(() -> {
return Mono.fromFuture(CompletableFuture.supplyAsync(() -> {
Properties properties = new Properties();
KafkaClientSslPropertiesUtil.addKafkaSslProperties(cluster.getOriginalProperties().getSsl(), properties);
properties.putAll(cluster.getProperties());
Expand All @@ -51,7 +52,7 @@ private Mono<ReactiveAdminClient> createAdminClient(KafkaCluster cluster) {
"kafbat-ui-admin-" + Instant.now().getEpochSecond() + "-" + CLIENT_ID_SEQ.incrementAndGet()
);
return AdminClient.create(properties);
}).flatMap(ac -> ReactiveAdminClient.create(ac).doOnError(th -> ac.close()))
})).flatMap(ac -> ReactiveAdminClient.create(ac).doOnError(th -> ac.close()))
.onErrorMap(th -> new IllegalStateException(
"Error while creating AdminClient for the cluster " + cluster.getName(), th));
}
Expand Down

0 comments on commit 5ebdd2d

Please sign in to comment.