Merge branch 'main' into issues/729

Author: zambrinf

.github/FUNDING.yml

@@ -1 +1,2 @@
 github: [kafbat]
+open_collective: kafka-ui

api/pom.xml

@@ -329,7 +329,7 @@
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-surefire-plugin</artifactId>
                 <configuration>
-                    <argLine>@{argLine} --illegal-access=permit</argLine>
+                    <argLine>@{argLine}</argLine>
                 </configuration>
             </plugin>
             <plugin>

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

@@ -12,7 +12,7 @@
 import io.kafbat.ui.connect.model.ConnectorTopics;
 import io.kafbat.ui.connect.model.NewConnector;
 import io.kafbat.ui.connect.model.TaskStatus;
-import io.kafbat.ui.exception.KafkaConnectConflictReponseException;
+import io.kafbat.ui.exception.KafkaConnectConflictResponseException;
 import io.kafbat.ui.exception.ValidationException;
 import io.kafbat.ui.util.WebClientConfigurator;
 import jakarta.validation.constraints.NotNull;
@@ -48,7 +48,7 @@ private static Retry conflictCodeRetry() {
         .fixedDelay(MAX_RETRIES, RETRIES_DELAY)
         .filter(e -> e instanceof WebClientResponseException.Conflict)
         .onRetryExhaustedThrow((spec, signal) ->
-            new KafkaConnectConflictReponseException(
+            new KafkaConnectConflictResponseException(
                 (WebClientResponseException.Conflict) signal.failure()));
   }
 

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

@@ -1,8 +1,6 @@
 package io.kafbat.ui.config.auth;
 
-import io.kafbat.ui.util.EmptyRedirectStrategy;
 import io.kafbat.ui.util.StaticFileWebFilter;
-import java.net.URI;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
@@ -12,8 +10,6 @@
 import org.springframework.security.config.web.server.SecurityWebFiltersOrder;
 import org.springframework.security.config.web.server.ServerHttpSecurity;
 import org.springframework.security.web.server.SecurityWebFilterChain;
-import org.springframework.security.web.server.authentication.RedirectServerAuthenticationSuccessHandler;
-import org.springframework.security.web.server.authentication.logout.RedirectServerLogoutSuccessHandler;
 import org.springframework.security.web.server.util.matcher.ServerWebExchangeMatchers;
 
 @Configuration

api/src/main/java/io/kafbat/ui/config/auth/condition/ActiveDirectoryCondition.java

@@ -40,7 +40,7 @@ public Mono<Void> handle(WebFilterExchange exchange, Authentication authenticati
         requestUri.getPath(), requestUri.getQuery());
 
     final UriComponents baseUrl = UriComponentsBuilder
-        .fromHttpUrl(fullUrl)
+        .fromUriString(fullUrl)
         .replacePath("/")
         .replaceQuery(null)
         .fragment(null)

api/src/main/java/io/kafbat/ui/config/auth/logout/CognitoLogoutSuccessHandler.java

@@ -53,7 +53,7 @@ private Map<TopicPartition, Long> firstOffsetsForPolling(Consumer<?, ?> consumer
                                                            Collection<TopicPartition> partitions) {
     try {
       // we try to use offsetsForTimes() to find earliest offsets, since for
-      // some topics (like compacted) beginningOffsets() ruturning 0 offsets
+      // some topics (like compacted) beginningOffsets() returning 0 offsets
       // even when effectively first offset can be very high
       var offsets = consumer.offsetsForTimes(
           partitions.stream().collect(Collectors.toMap(p -> p, p -> 0L))

api/src/main/java/io/kafbat/ui/emitter/OffsetsInfo.java

@@ -1,7 +1,10 @@
 package io.kafbat.ui.exception;
 
+import lombok.Getter;
+
+@Getter
 public class CelException extends CustomBaseException {
-  private String celOriginalExpression;
+  private final String celOriginalExpression;
 
   public CelException(String celOriginalExpression, String errorMessage) {
     super("CEL error. Original expression: %s. Error message: %s".formatted(celOriginalExpression, errorMessage));

api/src/main/java/io/kafbat/ui/emitter/ResultSizeLimiter.java

@@ -4,25 +4,19 @@
 import org.slf4j.LoggerFactory;
 import org.springframework.http.HttpStatus;
 
-
 public enum ErrorCode {
 
-  FORBIDDEN(403, HttpStatus.FORBIDDEN),
-
   UNEXPECTED(5000, HttpStatus.INTERNAL_SERVER_ERROR),
   KSQL_API_ERROR(5001, HttpStatus.INTERNAL_SERVER_ERROR),
   BINDING_FAIL(4001, HttpStatus.BAD_REQUEST),
   NOT_FOUND(404, HttpStatus.NOT_FOUND),
   VALIDATION_FAIL(4002, HttpStatus.BAD_REQUEST),
   READ_ONLY_MODE_ENABLE(4003, HttpStatus.METHOD_NOT_ALLOWED),
   CONNECT_CONFLICT_RESPONSE(4004, HttpStatus.CONFLICT),
-  DUPLICATED_ENTITY(4005, HttpStatus.CONFLICT),
   UNPROCESSABLE_ENTITY(4006, HttpStatus.UNPROCESSABLE_ENTITY),
   CLUSTER_NOT_FOUND(4007, HttpStatus.NOT_FOUND),
   TOPIC_NOT_FOUND(4008, HttpStatus.NOT_FOUND),
   SCHEMA_NOT_FOUND(4009, HttpStatus.NOT_FOUND),
-  CONNECT_NOT_FOUND(4010, HttpStatus.NOT_FOUND),
-  KSQLDB_NOT_FOUND(4011, HttpStatus.NOT_FOUND),
   DIR_NOT_FOUND(4012, HttpStatus.BAD_REQUEST),
   TOPIC_OR_PARTITION_NOT_FOUND(4013, HttpStatus.BAD_REQUEST),
   INVALID_REQUEST(4014, HttpStatus.BAD_REQUEST),

api/src/main/java/io/kafbat/ui/exception/CelException.java

@@ -54,18 +54,18 @@ private Mono<ServerResponse> renderErrorResponse(ServerRequest request) {
     Throwable throwable = getError(request);
 
     // validation and params binding errors
-    if (throwable instanceof WebExchangeBindException) {
-      return render((WebExchangeBindException) throwable, request);
+    if (throwable instanceof WebExchangeBindException webExchangeBindException) {
+      return render(webExchangeBindException, request);
     }
 
     // requests mapping & access errors
-    if (throwable instanceof ResponseStatusException) {
-      return render((ResponseStatusException) throwable, request);
+    if (throwable instanceof ResponseStatusException responseStatusException) {
+      return render(responseStatusException, request);
     }
 
     // custom exceptions
-    if (throwable instanceof CustomBaseException) {
-      return render((CustomBaseException) throwable, request);
+    if (throwable instanceof CustomBaseException customBaseException) {
+      return render(customBaseException, request);
     }
 
     return renderDefault(throwable, request);
@@ -102,8 +102,7 @@ private Mono<ServerResponse> render(CustomBaseException baseException, ServerReq
 
   private Mono<ServerResponse> render(WebExchangeBindException exception, ServerRequest request) {
     Map<String, Set<String>> fieldErrorsMap = exception.getFieldErrors().stream()
-        .collect(Collectors
-            .toMap(FieldError::getField, f -> Set.of(extractFieldErrorMsg(f)), Sets::union));
+        .collect(Collectors.toMap(FieldError::getField, f -> Set.of(extractFieldErrorMsg(f)), Sets::union));
 
     var fieldsErrors = fieldErrorsMap.entrySet().stream()
         .map(e -> {
@@ -151,9 +150,7 @@ private String requestId(ServerRequest request) {
   }
 
   private Consumer<HttpHeaders> headers(ServerRequest request) {
-    return (HttpHeaders headers) -> {
-      CorsGlobalConfiguration.fillCorsHeader(headers, request.exchange().getRequest());
-    };
+    return (HttpHeaders headers) -> CorsGlobalConfiguration.fillCorsHeader(headers, request.exchange().getRequest());
   }
 
   private BigDecimal currentTimestamp() {

api/src/main/java/io/kafbat/ui/exception/ConnectNotFoundException.java

@@ -1,11 +1,10 @@
 package io.kafbat.ui.exception;
 
-
 import org.springframework.web.reactive.function.client.WebClientResponseException;
 
-public class KafkaConnectConflictReponseException extends CustomBaseException {
+public class KafkaConnectConflictResponseException extends CustomBaseException {
 
-  public KafkaConnectConflictReponseException(WebClientResponseException.Conflict e) {
+  public KafkaConnectConflictResponseException(WebClientResponseException.Conflict e) {
     super("Kafka Connect responded with 409 (Conflict) code. Response body: "
         + e.getResponseBodyAsString());
   }

api/src/main/java/io/kafbat/ui/exception/DuplicateEntityException.java

@@ -97,23 +97,15 @@ private static BrokerDTO mapCoordinator(Node node) {
     return new BrokerDTO().host(node.host()).id(node.id()).port(node.port());
   }
 
-  private static ConsumerGroupStateDTO mapConsumerGroupState(
-      org.apache.kafka.common.ConsumerGroupState state) {
-    switch (state) {
-      case DEAD:
-        return ConsumerGroupStateDTO.DEAD;
-      case EMPTY:
-        return ConsumerGroupStateDTO.EMPTY;
-      case STABLE:
-        return ConsumerGroupStateDTO.STABLE;
-      case PREPARING_REBALANCE:
-        return ConsumerGroupStateDTO.PREPARING_REBALANCE;
-      case COMPLETING_REBALANCE:
-        return ConsumerGroupStateDTO.COMPLETING_REBALANCE;
-      default:
-        return ConsumerGroupStateDTO.UNKNOWN;
-    }
+  private static ConsumerGroupStateDTO mapConsumerGroupState(org.apache.kafka.common.ConsumerGroupState state) {
+    return switch (state) {
+      case DEAD -> ConsumerGroupStateDTO.DEAD;
+      case EMPTY -> ConsumerGroupStateDTO.EMPTY;
+      case STABLE -> ConsumerGroupStateDTO.STABLE;
+      case PREPARING_REBALANCE -> ConsumerGroupStateDTO.PREPARING_REBALANCE;
+      case COMPLETING_REBALANCE -> ConsumerGroupStateDTO.COMPLETING_REBALANCE;
+      default -> ConsumerGroupStateDTO.UNKNOWN;
+    };
   }
 
-
 }

api/src/main/java/io/kafbat/ui/exception/ErrorCode.java

@@ -42,7 +42,7 @@ private BrokersLogdirsDTO toBrokerLogDirs(Integer broker, String dirName,
 
   private BrokerTopicLogdirsDTO toTopicLogDirs(Integer broker, String name,
                                                List<Map.Entry<TopicPartition,
-                                                   DescribeLogDirsResponse.ReplicaInfo>> partitions) {
+                                               DescribeLogDirsResponse.ReplicaInfo>> partitions) {
     BrokerTopicLogdirsDTO topic = new BrokerTopicLogdirsDTO();
     topic.setName(name);
     topic.setPartitions(
@@ -54,8 +54,7 @@ private BrokerTopicLogdirsDTO toTopicLogDirs(Integer broker, String name,
   }
 
   private BrokerTopicPartitionLogdirDTO topicPartitionLogDir(Integer broker, Integer partition,
-                                                             DescribeLogDirsResponse.ReplicaInfo
-                                                                 replicaInfo) {
+                                                             DescribeLogDirsResponse.ReplicaInfo replicaInfo) {
     BrokerTopicPartitionLogdirDTO logDir = new BrokerTopicPartitionLogdirDTO();
     logDir.setBroker(broker);
     logDir.setPartition(partition);