Default timeout for http requests to prevent hanging - ability to configurate in properties

Author: o-afanasenko

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

@@ -68,6 +68,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 responseTimeout;
 
   @PostConstruct
   public void validate() {

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.getResponseTimeout())
+        .map(Duration::ofMillis)
+        .orElse(DEFAULT_RESPONSE_TIMEOUT);
   }
 
   public KafkaCluster create(ClustersProperties properties,
@@ -145,7 +151,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/util/WebClientConfigurator.java

@@ -32,8 +32,7 @@ public class WebClientConfigurator {
   private final WebClient.Builder builder = WebClient.builder();
   private HttpClient httpClient = HttpClient
       .create()
-      .proxyWithSystemProperties()
-      .responseTimeout(Duration.ofSeconds(10));
+      .proxyWithSystemProperties();
 
   public WebClientConfigurator() {
     configureObjectMapper(defaultOM());
@@ -146,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();
   }