Merge branch 'main' into issues/1332-be

Author: germanosin

api/src/main/java/io/kafbat/ui/client/RetryingKafkaConnectClient.java

@@ -7,11 +7,13 @@
 import io.kafbat.ui.connect.ApiClient;
 import io.kafbat.ui.connect.api.KafkaConnectClientApi;
 import io.kafbat.ui.connect.model.Connector;
+import io.kafbat.ui.connect.model.ConnectorExpand;
 import io.kafbat.ui.connect.model.ConnectorPlugin;
 import io.kafbat.ui.connect.model.ConnectorPluginConfigValidationResponse;
 import io.kafbat.ui.connect.model.ConnectorStatus;
 import io.kafbat.ui.connect.model.ConnectorTask;
 import io.kafbat.ui.connect.model.ConnectorTopics;
+import io.kafbat.ui.connect.model.ExpandedConnector;
 import io.kafbat.ui.connect.model.NewConnector;
 import io.kafbat.ui.connect.model.TaskStatus;
 import io.kafbat.ui.exception.KafkaConnectConflictResponseException;
@@ -221,13 +223,17 @@ public Mono<ResponseEntity<Map<String, ConnectorTopics>>> getConnectorTopicsWith
   }
 
   @Override
-  public Mono<List<String>> getConnectors(String search) throws WebClientResponseException {
-    return withRetryOnConflictOrRebalance(super.getConnectors(search));
+  public Mono<Map<String, ExpandedConnector>> getConnectors(
+      String search, List<ConnectorExpand> expand
+  ) throws WebClientResponseException {
+    return withRetryOnConflictOrRebalance(super.getConnectors(search, expand));
   }
 
   @Override
-  public Mono<ResponseEntity<List<String>>> getConnectorsWithHttpInfo(String search) throws WebClientResponseException {
-    return withRetryOnConflictOrRebalance(super.getConnectorsWithHttpInfo(search));
+  public Mono<ResponseEntity<Map<String, ExpandedConnector>>> getConnectorsWithHttpInfo(
+      String search, List<ConnectorExpand> expand
+  ) throws WebClientResponseException {
+    return withRetryOnConflictOrRebalance(super.getConnectorsWithHttpInfo(search, expand));
   }
 
   @Override

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

@@ -227,7 +227,6 @@ public enum LogLevel {
   @AllArgsConstructor
   public static class CacheProperties {
     boolean enabled = true;
-    Duration connectCacheExpiry = Duration.ofMinutes(1);
     Duration connectClusterCacheExpiry = Duration.ofHours(24);
   }
 

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

@@ -67,8 +67,12 @@ public Mono<ResponseEntity<Flux<String>>> getConnectors(String clusterName, Stri
         .build();
 
     return validateAccess(context)
-        .thenReturn(ResponseEntity.ok(kafkaConnectService.getConnectorNames(getCluster(clusterName), connectName)))
-        .doOnEach(sig -> audit(context, sig));
+        .thenReturn(
+            ResponseEntity.ok(
+              kafkaConnectService.getConnectors(getCluster(clusterName), connectName)
+                  .flatMapMany(m -> Flux.fromIterable(m.keySet()))
+            )
+        ).doOnEach(sig -> audit(context, sig));
   }
 
   @Override

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

@@ -4,6 +4,7 @@
 import io.kafbat.ui.connect.model.ClusterInfo;
 import io.kafbat.ui.connect.model.ConnectorStatusConnector;
 import io.kafbat.ui.connect.model.ConnectorTask;
+import io.kafbat.ui.connect.model.ExpandedConnector;
 import io.kafbat.ui.connect.model.NewConnector;
 import io.kafbat.ui.model.ConnectDTO;
 import io.kafbat.ui.model.ConnectorDTO;
@@ -14,10 +15,15 @@
 import io.kafbat.ui.model.ConnectorTaskStatusDTO;
 import io.kafbat.ui.model.FullConnectorInfoDTO;
 import io.kafbat.ui.model.TaskDTO;
+import io.kafbat.ui.model.TaskIdDTO;
 import io.kafbat.ui.model.TaskStatusDTO;
 import io.kafbat.ui.model.connect.InternalConnectorInfo;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
 import org.mapstruct.Mapper;
 import org.mapstruct.Mapping;
 
@@ -43,6 +49,39 @@ ConnectorPluginConfigValidationResponseDTO fromClient(
       io.kafbat.ui.connect.model.ConnectorPluginConfigValidationResponse
           connectorPluginConfigValidationResponse);
 
+  default InternalConnectorInfo fromClient(String connect, ExpandedConnector connector, @Nullable List<String> topics) {
+    Objects.requireNonNull(connector.getInfo());
+    Objects.requireNonNull(connector.getStatus());
+    List<TaskDTO> tasks = List.of();
+
+    if (connector.getInfo().getTasks() != null
+        && connector.getStatus().getTasks() != null
+    ) {
+      Map<Integer, TaskIdDTO> taskIds = connector.getInfo().getTasks()
+          .stream().map(t -> new TaskIdDTO().task(t.getTask()).connector(t.getConnector()))
+          .collect(Collectors.toMap(
+              TaskIdDTO::getTask,
+              t -> t
+          ));
+
+      tasks = connector.getStatus().getTasks().stream()
+          .map(s ->
+              new TaskDTO().status(fromClient(s)).id(taskIds.get(s.getId()))
+          ).toList();
+    }
+
+    ConnectorDTO connectorDto = fromClient(connector.getInfo())
+        .connect(connect)
+        .status(fromClient(connector.getStatus().getConnector()));
+
+    return InternalConnectorInfo.builder()
+        .connector(connectorDto)
+        .config(connector.getInfo().getConfig())
+        .tasks(tasks)
+        .topics(topics)
+        .build();
+  }
+
   default ConnectDTO toKafkaConnect(
       ClustersProperties.ConnectCluster connect,
       List<InternalConnectorInfo> connectors,

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

@@ -3,6 +3,7 @@
 import static io.kafbat.ui.api.model.AuthType.DISABLED;
 import static io.kafbat.ui.api.model.AuthType.OAUTH2;
 import static io.kafbat.ui.model.ApplicationInfoDTO.EnabledFeaturesEnum;
+import static io.kafbat.ui.util.GithubReleaseInfo.GITHUB_RELEASE_INFO_ENABLED;
 import static io.kafbat.ui.util.GithubReleaseInfo.GITHUB_RELEASE_INFO_TIMEOUT;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -15,12 +16,14 @@
 import io.kafbat.ui.model.OAuthProviderDTO;
 import io.kafbat.ui.util.DynamicConfigOperations;
 import io.kafbat.ui.util.GithubReleaseInfo;
+import jakarta.annotation.Nullable;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.Properties;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.info.BuildProperties;
@@ -33,7 +36,9 @@
 import org.springframework.stereotype.Service;
 
 @Service
+@Slf4j
 public class ApplicationInfoService {
+  @Nullable
   private final GithubReleaseInfo githubReleaseInfo;
   private final ApplicationContext applicationContext;
   private final DynamicConfigOperations dynamicConfigOperations;
@@ -44,36 +49,52 @@ public ApplicationInfoService(DynamicConfigOperations dynamicConfigOperations,
                                 ApplicationContext applicationContext,
                                 @Autowired(required = false) BuildProperties buildProperties,
                                 @Autowired(required = false) GitProperties gitProperties,
+                                @Value("${" + GITHUB_RELEASE_INFO_ENABLED + ":true}") boolean githubInfoEnabled,
                                 @Value("${" + GITHUB_RELEASE_INFO_TIMEOUT + ":10}") int githubApiMaxWaitTime) {
     this.applicationContext = applicationContext;
     this.dynamicConfigOperations = dynamicConfigOperations;
     this.buildProperties = Optional.ofNullable(buildProperties).orElse(new BuildProperties(new Properties()));
     this.gitProperties = Optional.ofNullable(gitProperties).orElse(new GitProperties(new Properties()));
-    githubReleaseInfo = new GithubReleaseInfo(githubApiMaxWaitTime);
+    if (githubInfoEnabled) {
+      this.githubReleaseInfo = new GithubReleaseInfo(githubApiMaxWaitTime);
+    } else {
+      this.githubReleaseInfo = null;
+      log.warn("Check for latest release is disabled."
+          + " Note that old versions are not supported, please make sure that your system is up to date.");
+    }
   }
 
   public ApplicationInfoDTO getApplicationInfo() {
-    var releaseInfo = githubReleaseInfo.get();
+    var releaseInfo = githubReleaseInfo != null ? githubReleaseInfo.get() : null;
     return new ApplicationInfoDTO()
         .build(getBuildInfo(releaseInfo))
         .enabledFeatures(getEnabledFeatures())
         .latestRelease(convert(releaseInfo));
   }
 
+  @Nullable
   private ApplicationInfoLatestReleaseDTO convert(GithubReleaseInfo.GithubReleaseDto releaseInfo) {
+    if (releaseInfo == null) {
+      return null;
+    }
     return new ApplicationInfoLatestReleaseDTO()
         .htmlUrl(releaseInfo.html_url())
         .publishedAt(releaseInfo.published_at())
         .versionTag(releaseInfo.tag_name());
   }
 
   private ApplicationInfoBuildDTO getBuildInfo(GithubReleaseInfo.GithubReleaseDto release) {
-    return new ApplicationInfoBuildDTO()
-        .isLatestRelease(release.tag_name() != null && release.tag_name().equals(buildProperties.getVersion()))
+    var buildInfo = new ApplicationInfoBuildDTO()
         .commitId(gitProperties.getShortCommitId())
         .version(buildProperties.getVersion())
         .buildTime(buildProperties.getTime() != null
             ? DateTimeFormatter.ISO_INSTANT.format(buildProperties.getTime()) : null);
+    if (release != null) {
+      buildInfo = buildInfo.isLatestRelease(
+          release.tag_name() != null && release.tag_name().equals(buildProperties.getVersion())
+      );
+    }
+    return buildInfo;
   }
 
   private List<EnabledFeaturesEnum> getEnabledFeatures() {
@@ -119,10 +140,13 @@ private List<OAuthProviderDTO> getOAuthProviders() {
   // updating on startup and every hour
   @Scheduled(fixedRateString = "${github-release-info-update-rate:3600000}")
   public void updateGithubReleaseInfo() {
-    githubReleaseInfo.refresh().subscribe();
+    if (githubReleaseInfo != null) {
+      githubReleaseInfo.refresh().subscribe();
+    }
   }
 
   @VisibleForTesting
+  @Nullable
   GithubReleaseInfo githubReleaseInfo() {
     return githubReleaseInfo;
   }

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

@@ -19,7 +19,7 @@
 import org.springframework.stereotype.Component;
 
 @Component
-class KafkaConfigSanitizer {
+public class KafkaConfigSanitizer {
 
   private static final String SANITIZED_VALUE = "******";