BE: Invalidate Admin client on kafka error (#1468)

Author: germanosin

api/src/main/java/io/kafbat/ui/service/AdminClientService.java

@@ -3,8 +3,10 @@
 import io.kafbat.ui.model.KafkaCluster;
 import reactor.core.publisher.Mono;
 
-public interface AdminClientService {
+public interface AdminClientService extends AutoCloseable {
 
   Mono<ReactiveAdminClient> get(KafkaCluster cluster);
 
+  void invalidate(KafkaCluster cluster, Throwable e);
+
 }

api/src/main/java/io/kafbat/ui/service/AdminClientServiceImpl.java

@@ -19,7 +19,7 @@
 
 @Service
 @Slf4j
-public class AdminClientServiceImpl implements AdminClientService, Closeable {
+public class AdminClientServiceImpl implements AdminClientService {
 
   private static final int DEFAULT_CLIENT_TIMEOUT_MS = 30_000;
 
@@ -62,6 +62,21 @@ private Mono<ReactiveAdminClient> createAdminClient(KafkaCluster cluster) {
             "Error while creating AdminClient for the cluster " + cluster.getName(), th));
   }
 
+  @Override
+  public void invalidate(KafkaCluster cluster, Throwable e) {
+    if (e.getClass().getCanonicalName().startsWith("org.apache.kafka.common.errors")) {
+      log.warn("AdminClient for the cluster {} is invalidated due to {}", cluster.getName(), e.getMessage());
+      ReactiveAdminClient client = adminClientCache.remove(cluster.getName());
+      if (client != null) {
+        try {
+          client.close();
+        } catch (Exception ce) {
+          log.info("Error while closing AdminClient for the cluster {}", cluster.getName(), ce);
+        }
+      }
+    }
+  }
+
   @Override
   public void close() {
     adminClientCache.values().forEach(ReactiveAdminClient::close);

api/src/main/java/io/kafbat/ui/service/BrokerService.java

@@ -73,7 +73,8 @@ public Flux<InternalBroker> getBrokers(KafkaCluster cluster) {
         .map(description -> description.getNodes().stream()
             .map(node -> new InternalBroker(node, partitionsDistribution, stats.getMetrics()))
             .collect(Collectors.toList()))
-        .flatMapMany(Flux::fromIterable);
+        .flatMapMany(Flux::fromIterable)
+        .doOnError(e -> adminClientService.invalidate(cluster, e));
   }
 
   public Mono<Void> updateBrokerLogDir(KafkaCluster cluster,

api/src/main/java/io/kafbat/ui/service/StatisticsService.java

@@ -49,7 +49,8 @@ private Mono<Statistics> getStatistics(KafkaCluster cluster) {
                         )
                     )
             ).doOnError(e ->
-                log.error("Failed to collect cluster {} info", cluster.getName(), e))
+                log.error("Failed to collect cluster {} info", cluster.getName(), e)
+            ).doOnError(e -> adminClientService.invalidate(cluster, e))
             .onErrorResume(t -> Mono.just(Statistics.statsUpdateError(t))));
   }
 

api/src/main/java/io/kafbat/ui/service/TopicsService.java

@@ -169,7 +169,8 @@ public Mono<List<ConfigEntry>> getTopicConfigs(KafkaCluster cluster, String topi
         .flatMap(ac -> ac.describeTopic(topicName)
             .switchIfEmpty(Mono.error(new TopicNotFoundException()))
             .then(ac.getTopicsConfig(List.of(topicName), true))
-            .map(m -> m.values().stream().findFirst().orElse(List.of())));
+            .map(m -> m.values().stream().findFirst().orElse(List.of())))
+            .doOnError(e -> adminClientService.invalidate(cluster, e));
   }
 
   private Mono<InternalTopic> createTopic(KafkaCluster c, ReactiveAdminClient adminClient, TopicCreationDTO topicData) {
@@ -178,6 +179,7 @@ private Mono<InternalTopic> createTopic(KafkaCluster c, ReactiveAdminClient admi
             topicData.getPartitions(),
             topicData.getReplicationFactor(),
             topicData.getConfigs())
+        .doOnError(e -> adminClientService.invalidate(c, e))
         .thenReturn(topicData)
         .onErrorMap(t -> new TopicMetadataException(t.getMessage(), t))
         .then(loadTopicAfterCreation(c, topicData.getName()));
@@ -492,7 +494,8 @@ private Mono<List<InternalTopic>> filterExisting(KafkaCluster cluster, Collectio
         .map(existing -> topics
             .stream()
             .filter(s -> existing.contains(s.getName()))
-            .collect(toList()));
+            .collect(toList())
+        ).doOnError(e -> adminClientService.invalidate(cluster, e));
   }
 
 }

api/src/test/java/io/kafbat/ui/service/AdminClientServiceTest.java

@@ -0,0 +1,25 @@
+package io.kafbat.ui.service;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.kafbat.ui.AbstractIntegrationTest;
+import io.kafbat.ui.model.KafkaCluster;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.junit.jupiter.api.Test;
+
+@Slf4j
+class AdminClientServiceTest extends AbstractIntegrationTest {
+  @Test
+  void testInvalidateOnFailure() {
+    AdminClientService adminClientService = applicationContext.getBean(AdminClientService.class);
+    ClustersStorage clustersStorage = applicationContext.getBean(ClustersStorage.class);
+    KafkaCluster cluster = clustersStorage.getClusterByName(LOCAL).get();
+    ReactiveAdminClient clientBefore = adminClientService.get(cluster).block();
+    ReactiveAdminClient clientBeforeRepeat = adminClientService.get(cluster).block();
+    assertThat(clientBeforeRepeat).isEqualTo(clientBefore);
+    adminClientService.invalidate(cluster, new TimeoutException());
+    ReactiveAdminClient clientAfter = adminClientService.get(cluster).block();
+    assertThat(clientAfter).isNotEqualTo(clientBefore);
+  }
+}