Merge branch 'main' into feature/avro-message-details

Author: mbollmann-v

README.md

@@ -99,7 +99,7 @@ To run Kafbat UI, you can use either a pre-built Docker image or build it (or a
 
 ## Quick start (Demo run)
 
-```
+```bash
 docker run -it -p 8080:8080 -e DYNAMIC_CONFIG_ENABLED=true ghcr.io/kafbat/kafka-ui
 ```
 
@@ -109,7 +109,7 @@ The command is sufficient to try things out. When you're done trying things out,
 
 ## Persistent installation
 
-```
+```yml
 services:
   kafbat-ui:
     container_name: kafbat-ui
@@ -158,3 +158,7 @@ Please refer to [contributing guide](https://ui.docs.kafbat.io/development/contr
 
 As we're fully independent, team members contribute in their free time.
 Your support is crucial for us, if you wish to sponsor us, take a look [here](https://github.com/sponsors/kafbat) 
+
+# Powered by
+
+[![JetBrains logo.](https://resources.jetbrains.com/storage/products/company/brand/logos/jetbrains.svg)](https://jb.gg/OpenSourceSupport)

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

@@ -39,8 +39,9 @@ public class RetryingKafkaConnectClient extends KafkaConnectClientApi {
 
   public RetryingKafkaConnectClient(ClustersProperties.ConnectCluster config,
                                     @Nullable ClustersProperties.TruststoreConfig truststoreConfig,
-                                    DataSize maxBuffSize) {
-    super(new RetryingApiClient(config, truststoreConfig, maxBuffSize));
+                                    DataSize maxBuffSize,
+                                    Duration responseTimeout) {
+    super(new RetryingApiClient(config, truststoreConfig, maxBuffSize, responseTimeout));
   }
 
   private static Retry conflictCodeRetry() {
@@ -318,14 +319,16 @@ private static class RetryingApiClient extends ApiClient {
 
     public RetryingApiClient(ClustersProperties.ConnectCluster config,
                              ClustersProperties.TruststoreConfig truststoreConfig,
-                             DataSize maxBuffSize) {
-      super(buildWebClient(maxBuffSize, config, truststoreConfig), null, null);
+                             DataSize maxBuffSize,
+                             Duration responseTimeout) {
+      super(buildWebClient(maxBuffSize, responseTimeout, config, truststoreConfig), null, null);
       setBasePath(config.getAddress());
       setUsername(config.getUsername());
       setPassword(config.getPassword());
     }
 
     public static WebClient buildWebClient(DataSize maxBuffSize,
+                                           Duration responseTimeout,
                                            ClustersProperties.ConnectCluster config,
                                            ClustersProperties.TruststoreConfig truststoreConfig) {
       return new WebClientConfigurator()
@@ -341,6 +344,7 @@ public static WebClient buildWebClient(DataSize maxBuffSize,
               config.getPassword()
           )
           .configureBufferSize(maxBuffSize)
+          .configureResponseTimeout(responseTimeout)
           .build();
     }
   }

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

@@ -77,6 +77,7 @@ public static class PollingProperties {
     Integer pollTimeoutMs;
     Integer maxPageSize;
     Integer defaultPageSize;
+    Integer responseTimeoutMs;
   }
 
   @Data

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

@@ -13,6 +13,7 @@
 public class WebclientProperties {
 
   String maxInMemoryBufferSize;
+  Integer responseTimeoutMs;
 
   @PostConstruct
   public void validate() {

api/src/main/java/io/kafbat/ui/config/auth/LdapProperties.java

@@ -20,7 +20,7 @@ public class LdapProperties {
 
   @Value("${oauth2.ldap.activeDirectory:false}")
   private boolean isActiveDirectory;
-  @Value("${oauth2.ldap.activeDirectory.domain:@null}")
+  @Value("${oauth2.ldap.activeDirectory.domain:#{null}}")
   private String activeDirectoryDomain;
 
 }

api/src/main/java/io/kafbat/ui/config/auth/LdapSecurityConfig.java

@@ -12,6 +12,7 @@
 import java.util.stream.Stream;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -163,6 +164,10 @@ public SecurityWebFilterChain configureLdap(ServerHttpSecurity http) {
   }
 
   private ActiveDirectoryLdapAuthenticationProvider activeDirectoryProvider(LdapAuthoritiesPopulator populator) {
+    if (StringUtils.isBlank(props.getActiveDirectoryDomain())) {
+      throw new IllegalArgumentException("Active Directory domain is required but not specified");
+    }
+
     ActiveDirectoryLdapAuthenticationProvider provider = new ActiveDirectoryLdapAuthenticationProvider(
         props.getActiveDirectoryDomain(),
         props.getUrls()

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

@@ -16,6 +16,7 @@
 import io.kafbat.ui.util.KafkaServicesValidation;
 import io.kafbat.ui.util.ReactiveFailover;
 import io.kafbat.ui.util.WebClientConfigurator;
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -37,13 +38,18 @@
 public class KafkaClusterFactory {
 
   private static final DataSize DEFAULT_WEBCLIENT_BUFFER = DataSize.parse("20MB");
+  private static final Duration DEFAULT_RESPONSE_TIMEOUT = Duration.ofSeconds(20);
 
   private final DataSize webClientMaxBuffSize;
+  private final Duration responseTimeout;
 
   public KafkaClusterFactory(WebclientProperties webclientProperties) {
     this.webClientMaxBuffSize = Optional.ofNullable(webclientProperties.getMaxInMemoryBufferSize())
         .map(DataSize::parse)
         .orElse(DEFAULT_WEBCLIENT_BUFFER);
+    this.responseTimeout = Optional.ofNullable(webclientProperties.getResponseTimeoutMs())
+        .map(Duration::ofMillis)
+        .orElse(DEFAULT_RESPONSE_TIMEOUT);
   }
 
   public KafkaCluster create(ClustersProperties properties,
@@ -147,7 +153,8 @@ private ReactiveFailover<KafkaConnectClientApi> connectClient(ClustersProperties
         url -> new RetryingKafkaConnectClient(
             connectCluster.toBuilder().address(url).build(),
             cluster.getSsl(),
-            webClientMaxBuffSize
+            webClientMaxBuffSize,
+            responseTimeout
         ),
         ReactiveFailover.CONNECTION_REFUSED_EXCEPTION_FILTER,
         "No alive connect instances available",

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

@@ -29,7 +29,7 @@ class KafkaConfigSanitizer {
       .addAll(kafkaConfigKeysToSanitize())
       .add(
           "basic.auth.user.info",  /* For Schema Registry credentials */
-          "password", "secret", "token", "key", ".*credentials.*",   /* General credential patterns */
+          "password", "secret", "token", "key", ".*credentials.*", "passphrase",   /* General credential patterns */
           "aws.access.*", "aws.secret.*", "aws.session.*",   /* AWS-related credential patterns */
           "connection.uri" /* mongo credential patterns */
       )

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

@@ -10,6 +10,7 @@
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
 import java.io.FileInputStream;
 import java.security.KeyStore;
+import java.time.Duration;
 import java.util.function.Consumer;
 import javax.annotation.Nullable;
 import javax.net.ssl.KeyManagerFactory;
@@ -144,6 +145,11 @@ public WebClientConfigurator configureCodecs(Consumer<ClientCodecConfigurer> con
     return this;
   }
 
+  public WebClientConfigurator configureResponseTimeout(Duration responseTimeout) {
+    httpClient = httpClient.responseTimeout(responseTimeout);
+    return this;
+  }
+
   public WebClient build() {
     return builder.clientConnector(new ReactorClientHttpConnector(httpClient)).build();
   }

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

@@ -27,6 +27,7 @@ void obfuscateCredentials() {
     assertThat(sanitizer.sanitize("main.consumer.sasl.jaas.config", "secret")).isEqualTo("******");
     assertThat(sanitizer.sanitize("database.password", "secret")).isEqualTo("******");
     assertThat(sanitizer.sanitize("basic.auth.user.info", "secret")).isEqualTo("******");
+    assertThat(sanitizer.sanitize("private.key.passphrase", "secret")).isEqualTo("******");
 
     //AWS var sanitizing
     assertThat(sanitizer.sanitize("aws.access.key.id", "secret")).isEqualTo("******");