Added configs, stats and cache

Author: germanosin

api/build.gradle

@@ -25,6 +25,7 @@ dependencies {
     implementation libs.spring.starter.oauth2.client
     implementation libs.spring.security.oauth2.resource.server
     implementation libs.spring.boot.actuator
+
     compileOnly libs.spring.boot.devtools
 
     implementation libs.spring.security.ldap
@@ -48,6 +49,7 @@ dependencies {
 
     implementation libs.jackson.databind.nullable
     implementation libs.cel
+    implementation libs.caffeine
     antlr libs.antlr
     implementation libs.antlr.runtime
 

api/src/main/java/io/kafbat/ui/config/ClustersProperties.java

@@ -5,6 +5,7 @@
 import jakarta.validation.Valid;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -36,6 +37,8 @@ public class ClustersProperties {
 
   PollingProperties polling = new PollingProperties();
 
+  CacheProperties cache = new CacheProperties();
+
   @Data
   public static class Cluster {
     @NotBlank(message = "field name for for cluster could not be blank")
@@ -183,6 +186,14 @@ public enum LogLevel {
     }
   }
 
+  @Data
+  @NoArgsConstructor
+  @AllArgsConstructor
+  public static class CacheProperties {
+    boolean enabled = true;
+    Duration connectCacheExpiry = Duration.ofMinutes(1);
+  }
+
   @PostConstruct
   public void validateAndSetDefaults() {
     if (clusters != null) {

api/src/main/java/io/kafbat/ui/controller/KafkaConnectController.java

@@ -45,10 +45,12 @@ public class KafkaConnectController extends AbstractController implements KafkaC
 
   @Override
   public Mono<ResponseEntity<Flux<ConnectDTO>>> getConnects(String clusterName,
+                                                            Boolean withStats,
                                                             ServerWebExchange exchange) {
 
-    Flux<ConnectDTO> availableConnects = kafkaConnectService.getConnects(getCluster(clusterName))
-        .filterWhen(dto -> accessControlService.isConnectAccessible(dto, clusterName));
+    Flux<ConnectDTO> availableConnects = kafkaConnectService.getConnects(
+        getCluster(clusterName), withStats != null ? withStats : false
+        ).filterWhen(dto -> accessControlService.isConnectAccessible(dto, clusterName));
 
     return Mono.just(ResponseEntity.ok(availableConnects));
   }

api/src/main/java/io/kafbat/ui/mapper/ClusterMapper.java

@@ -1,6 +1,5 @@
 package io.kafbat.ui.mapper;
 
-import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.model.BrokerConfigDTO;
 import io.kafbat.ui.model.BrokerDTO;
 import io.kafbat.ui.model.BrokerDiskUsageDTO;
@@ -11,11 +10,6 @@
 import io.kafbat.ui.model.ClusterStatsDTO;
 import io.kafbat.ui.model.ConfigSourceDTO;
 import io.kafbat.ui.model.ConfigSynonymDTO;
-import io.kafbat.ui.model.ConnectDTO;
-import io.kafbat.ui.model.ConnectorDTO;
-import io.kafbat.ui.model.ConnectorStateDTO;
-import io.kafbat.ui.model.ConnectorStatusDTO;
-import io.kafbat.ui.model.ConnectorTaskStatusDTO;
 import io.kafbat.ui.model.InternalBroker;
 import io.kafbat.ui.model.InternalBrokerConfig;
 import io.kafbat.ui.model.InternalBrokerDiskUsage;
@@ -31,16 +25,13 @@
 import io.kafbat.ui.model.Metrics;
 import io.kafbat.ui.model.PartitionDTO;
 import io.kafbat.ui.model.ReplicaDTO;
-import io.kafbat.ui.model.TaskDTO;
 import io.kafbat.ui.model.TopicConfigDTO;
 import io.kafbat.ui.model.TopicDTO;
 import io.kafbat.ui.model.TopicDetailsDTO;
 import io.kafbat.ui.model.TopicProducerStateDTO;
-import io.kafbat.ui.model.connect.InternalConnectorInfo;
 import io.kafbat.ui.service.metrics.RawMetric;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.ProducerState;
 import org.apache.kafka.common.acl.AccessControlEntry;
@@ -114,39 +105,6 @@ default ConfigSynonymDTO toConfigSynonym(ConfigEntry.ConfigSynonym config) {
 
   ReplicaDTO toReplica(InternalReplica replica);
 
-  default ConnectDTO toKafkaConnect(ClustersProperties.ConnectCluster connect, List<InternalConnectorInfo> connectors) {
-    int connectorCount = connectors.size();
-    int failedConnectors = 0;
-    int tasksCount = 0;
-    int failedTasksCount = 0;
-
-    for (InternalConnectorInfo connector : connectors) {
-      Optional<ConnectorDTO> internalConnector = Optional.ofNullable(connector.getConnector());
-
-      failedConnectors += internalConnector
-          .map(ConnectorDTO::getStatus)
-          .map(ConnectorStatusDTO::getState)
-          .filter(ConnectorStateDTO.FAILED::equals)
-          .map(s -> 1).orElse(0);
-
-      tasksCount += internalConnector.map(c -> c.getTasks().size()).orElse(0);
-
-      for (TaskDTO task : connector.getTasks()) {
-        if (task.getStatus() != null && ConnectorTaskStatusDTO.FAILED.equals(task.getStatus().getState())) {
-          failedTasksCount += tasksCount;
-        }
-      }
-    }
-
-    return new ConnectDTO()
-        .address(connect.getAddress())
-        .name(connect.getName())
-        .connectorsCount(connectorCount)
-        .failedConnectorsCount(failedConnectors)
-        .tasksCount(tasksCount)
-        .failedTasksCount(failedTasksCount);
-  }
-
   List<ClusterDTO.FeaturesEnum> toFeaturesEnum(List<ClusterFeature> features);
 
   default List<PartitionDTO> map(Map<Integer, InternalPartition> map) {

api/src/main/java/io/kafbat/ui/mapper/KafkaConnectMapper.java

@@ -1,18 +1,22 @@
 package io.kafbat.ui.mapper;
 
+import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.connect.model.ConnectorStatusConnector;
 import io.kafbat.ui.connect.model.ConnectorTask;
 import io.kafbat.ui.connect.model.NewConnector;
+import io.kafbat.ui.model.ConnectDTO;
 import io.kafbat.ui.model.ConnectorDTO;
 import io.kafbat.ui.model.ConnectorPluginConfigValidationResponseDTO;
 import io.kafbat.ui.model.ConnectorPluginDTO;
+import io.kafbat.ui.model.ConnectorStateDTO;
 import io.kafbat.ui.model.ConnectorStatusDTO;
 import io.kafbat.ui.model.ConnectorTaskStatusDTO;
 import io.kafbat.ui.model.FullConnectorInfoDTO;
 import io.kafbat.ui.model.TaskDTO;
 import io.kafbat.ui.model.TaskStatusDTO;
 import io.kafbat.ui.model.connect.InternalConnectorInfo;
 import java.util.List;
+import java.util.Optional;
 import org.mapstruct.Mapper;
 import org.mapstruct.Mapping;
 
@@ -38,6 +42,50 @@ ConnectorPluginConfigValidationResponseDTO fromClient(
       io.kafbat.ui.connect.model.ConnectorPluginConfigValidationResponse
           connectorPluginConfigValidationResponse);
 
+  default ConnectDTO toKafkaConnect(
+      ClustersProperties.ConnectCluster connect,
+      List<InternalConnectorInfo> connectors,
+      boolean withStats) {
+    Integer connectorCount = null;
+    Integer failedConnectors = null;
+    Integer tasksCount = null;
+    Integer failedTasksCount = null;
+
+    if (withStats) {
+      connectorCount = connectors.size();
+      failedConnectors = 0;
+      tasksCount = 0;
+      failedTasksCount = 0;
+
+      for (InternalConnectorInfo connector : connectors) {
+        Optional<ConnectorDTO> internalConnector = Optional.ofNullable(connector.getConnector());
+
+        failedConnectors += internalConnector
+            .map(ConnectorDTO::getStatus)
+            .map(ConnectorStatusDTO::getState)
+            .filter(ConnectorStateDTO.FAILED::equals)
+            .map(s -> 1).orElse(0);
+
+        tasksCount += internalConnector.map(c -> c.getTasks().size()).orElse(0);
+
+        for (TaskDTO task : connector.getTasks()) {
+          if (task.getStatus() != null && ConnectorTaskStatusDTO.FAILED.equals(task.getStatus().getState())) {
+            failedTasksCount += tasksCount;
+          }
+        }
+      }
+
+    }
+
+    return new ConnectDTO()
+        .address(connect.getAddress())
+        .name(connect.getName())
+        .connectorsCount(connectorCount)
+        .failedConnectorsCount(failedConnectors)
+        .tasksCount(tasksCount)
+        .failedTasksCount(failedTasksCount);
+  }
+
   default FullConnectorInfoDTO fullConnectorInfo(InternalConnectorInfo connectInfo) {
     ConnectorDTO connector = connectInfo.getConnector();
     List<TaskDTO> tasks = connectInfo.getTasks();

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

@@ -1,5 +1,9 @@
 package io.kafbat.ui.service;
 
+
+import com.github.benmanes.caffeine.cache.AsyncCache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import io.kafbat.ui.config.ClustersProperties;
 import io.kafbat.ui.connect.api.KafkaConnectClientApi;
 import io.kafbat.ui.connect.model.ConnectorStatus;
 import io.kafbat.ui.connect.model.ConnectorStatusConnector;
@@ -23,13 +27,13 @@
 import io.kafbat.ui.model.TaskDTO;
 import io.kafbat.ui.model.connect.InternalConnectorInfo;
 import io.kafbat.ui.util.ReactiveFailover;
+import jakarta.validation.Valid;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
 import javax.annotation.Nullable;
-import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Service;
@@ -39,36 +43,67 @@
 
 @Service
 @Slf4j
-@RequiredArgsConstructor
 public class KafkaConnectService {
   private final ClusterMapper clusterMapper;
   private final KafkaConnectMapper kafkaConnectMapper;
   private final KafkaConfigSanitizer kafkaConfigSanitizer;
+  private final ClustersProperties clustersProperties;
 
-  public Flux<ConnectDTO> getConnects(KafkaCluster cluster) {
-    return Optional.ofNullable(cluster.getOriginalProperties().getKafkaConnect())
-        .map(connects -> Flux.fromIterable(connects).flatMap(connect ->
-            getConnectorNamesWithErrorsSuppress(cluster, connect.getName()).flatMap(connectorName ->
-                Mono.zip(
-                    getConnector(cluster, connect.getName(), connectorName),
-                    getConnectorTasks(cluster, connect.getName(), connectorName).collectList()
-                ).map(tuple ->
-                    InternalConnectorInfo.builder()
-                        .connector(tuple.getT1())
-                        .config(null)
-                        .tasks(tuple.getT2())
-                        .topics(null)
-                        .build()
-                )
-            ).collectList().map(connectors ->
-                clusterMapper.toKafkaConnect(connect, connectors)
-            )
-        )).orElse(Flux.fromIterable(List.of()));
+  private final AsyncCache<ConnectCacheKey, List<InternalConnectorInfo>> cachedConnectors;
+
+  public KafkaConnectService(ClusterMapper clusterMapper, KafkaConnectMapper kafkaConnectMapper,
+                             KafkaConfigSanitizer kafkaConfigSanitizer,
+                             ClustersProperties clustersProperties) {
+    this.clusterMapper = clusterMapper;
+    this.kafkaConnectMapper = kafkaConnectMapper;
+    this.kafkaConfigSanitizer = kafkaConfigSanitizer;
+    this.clustersProperties = clustersProperties;
+    this.cachedConnectors = Caffeine.newBuilder()
+        .expireAfterWrite(clustersProperties.getCache().getConnectCacheExpiry())
+        .buildAsync();
+  }
+
+  public Flux<ConnectDTO> getConnects(KafkaCluster cluster, boolean withStats) {
+    Optional<List<ClustersProperties.@Valid ConnectCluster>> connectClusters =
+        Optional.ofNullable(cluster.getOriginalProperties().getKafkaConnect());
+    if (withStats) {
+      return connectClusters.map(connects -> Flux.fromIterable(connects).flatMap(connect -> (
+                  clustersProperties.getCache().isEnabled() ? Mono.fromFuture(
+                      cachedConnectors.get(new ConnectCacheKey(cluster, connect), (t, e) ->
+                          getConnectConnectors(t.cluster(), t.connect()).collectList().toFuture())
+                  ) : getConnectConnectors(cluster, connect).collectList()).map(connectors ->
+                  kafkaConnectMapper.toKafkaConnect(connect, connectors, withStats)
+              )
+          )
+      ).orElse(Flux.fromIterable(List.of()));
+    } else {
+      return Flux.fromIterable(connectClusters.map(connects ->
+          connects.stream().map(c -> kafkaConnectMapper.toKafkaConnect(c, List.of(), withStats)).toList()
+      ).orElse(List.of()));
+    }
+  }
+
+  private Flux<InternalConnectorInfo> getConnectConnectors(
+      KafkaCluster cluster,
+      ClustersProperties.ConnectCluster connect) {
+    return getConnectorNamesWithErrorsSuppress(cluster, connect.getName()).flatMap(connectorName ->
+        Mono.zip(
+            getConnector(cluster, connect.getName(), connectorName),
+            getConnectorTasks(cluster, connect.getName(), connectorName).collectList()
+        ).map(tuple ->
+            InternalConnectorInfo.builder()
+                .connector(tuple.getT1())
+                .config(null)
+                .tasks(tuple.getT2())
+                .topics(null)
+                .build()
+        )
+    );
   }
 
   public Flux<FullConnectorInfoDTO> getAllConnectors(final KafkaCluster cluster,
                                                      @Nullable final String search) {
-    return getConnects(cluster)
+    return getConnects(cluster, false)
         .flatMap(connect ->
             getConnectorNamesWithErrorsSuppress(cluster, connect.getName())
                 .flatMap(connectorName ->
@@ -302,4 +337,6 @@ public Mono<Void> resetConnectorOffsets(KafkaCluster cluster, String connectName
                       .formatted(connectorName, connectName));
             });
   }
+
+  record ConnectCacheKey(KafkaCluster cluster, ClustersProperties.ConnectCluster connect) {}
 }

api/src/main/java/io/kafbat/ui/service/integration/odd/ConnectorsExporter.java

@@ -24,7 +24,7 @@ class ConnectorsExporter {
   private final KafkaConnectService kafkaConnectService;
 
   Flux<DataEntityList> export(KafkaCluster cluster) {
-    return kafkaConnectService.getConnects(cluster)
+    return kafkaConnectService.getConnects(cluster, false)
         .flatMap(connect -> kafkaConnectService.getConnectorNamesWithErrorsSuppress(cluster, connect.getName())
             .flatMap(connectorName -> kafkaConnectService.getConnector(cluster, connect.getName(), connectorName))
             .flatMap(connectorDTO ->
@@ -41,7 +41,7 @@ Flux<DataEntityList> export(KafkaCluster cluster) {
   }
 
   Flux<DataSource> getConnectDataSources(KafkaCluster cluster) {
-    return kafkaConnectService.getConnects(cluster)
+    return kafkaConnectService.getConnects(cluster, false)
         .map(ConnectorsExporter::toDataSource);
   }
 

api/src/main/java/io/kafbat/ui/util/DynamicConfigOperations.java

@@ -187,20 +187,7 @@ private void writeYamlToFile(String yaml, Path path) {
   }
 
   private String serializeToYaml(PropertiesStructure props) {
-    //representer, that skips fields with null values
-    Representer representer = new Representer(new DumperOptions()) {
-      @Override
-      protected NodeTuple representJavaBeanProperty(Object javaBean,
-                                                    Property property,
-                                                    Object propertyValue,
-                                                    Tag customTag) {
-        if (propertyValue == null) {
-          return null; // if value of property is null, ignore it.
-        } else {
-          return super.representJavaBeanProperty(javaBean, property, propertyValue, customTag);
-        }
-      }
-    };
+    Representer representer = new YamlNullSkipRepresenter(new DumperOptions());
     var propertyUtils = new PropertyUtils();
     propertyUtils.setBeanAccess(BeanAccess.FIELD);
     representer.setPropertyUtils(propertyUtils);