Minor issues fixes (#1646)

* Fixed minor issues

* fixed review bug

* fixed bug

Co-authored-by: German Osin <germanosin@Germans-MacBook-Pro.local>
Co-authored-by: Roman Zabaluev <rzabaluev@provectus.com>

kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/KafkaConnectClients.java

@@ -6,6 +6,10 @@ import java.util.concurrent.ConcurrentHashMap;
 
 public final class KafkaConnectClients {
 
+  private KafkaConnectClients() {
+
+  }
+
   private static final Map<String, KafkaConnectClientApi> CACHE = new ConcurrentHashMap<>();
 
   public static KafkaConnectClientApi withBaseUrl(String basePath) {

kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/RetryingKafkaConnectClient.java

@@ -20,7 +20,6 @@ import org.springframework.web.reactive.function.client.WebClientResponseExcepti
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.util.retry.Retry;
-import reactor.util.retry.RetryBackoffSpec;
 
 @Slf4j
 public class RetryingKafkaConnectClient extends KafkaConnectClientApi {
@@ -32,7 +31,7 @@ public class RetryingKafkaConnectClient extends KafkaConnectClientApi {
   }
 
   private static Retry conflictCodeRetry() {
-    return RetryBackoffSpec
+    return Retry
         .fixedDelay(MAX_RETRIES, RETRIES_DELAY)
         .filter(e -> e instanceof WebClientResponseException.Conflict)
         .onRetryExhaustedThrow((spec, signal) ->

kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java

@@ -61,14 +61,14 @@ public class ClustersProperties {
 
   private void validateClusterNames() {
     // if only one cluster provided it is ok not to set name
-    if (clusters.size() == 1 && StringUtils.isEmpty(clusters.get(0).getName())) {
+    if (clusters.size() == 1 && !StringUtils.hasText(clusters.get(0).getName())) {
       clusters.get(0).setName("Default");
       return;
     }
 
     Set<String> clusterNames = new HashSet<>();
     for (Cluster clusterProperties : clusters) {
-      if (StringUtils.isEmpty(clusterProperties.getName())) {
+      if (!StringUtils.hasText(clusterProperties.getName())) {
         throw new IllegalStateException(
             "Application config isn't valid. "
                 + "Cluster names should be provided in case of multiple clusters present");
@@ -79,5 +79,4 @@ public class ClustersProperties {
       }
     }
   }
-
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/Config.java

@@ -1,6 +1,5 @@
 package com.provectus.kafka.ui.config;
 
-import com.fasterxml.jackson.databind.Module;
 import com.provectus.kafka.ui.model.JmxConnectionInfo;
 import com.provectus.kafka.ui.util.JmxPoolFactory;
 import java.util.Collections;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ReadOnlyModeFilter.java

@@ -45,7 +45,7 @@ public class ReadOnlyModeFilter implements WebFilter {
             () -> new ClusterNotFoundException(
                 String.format("No cluster for name '%s'", clusterName)));
 
-    if (!kafkaCluster.getReadOnly()) {
+    if (!kafkaCluster.isReadOnly()) {
       return chain.filter(exchange);
     }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/auth/AbstractAuthSecurityConfig.java

@@ -2,7 +2,11 @@ package com.provectus.kafka.ui.config.auth;
 
 abstract class AbstractAuthSecurityConfig {
 
-  public static final String[] AUTH_WHITELIST = {
+  protected AbstractAuthSecurityConfig() {
+
+  }
+
+  protected static final String[] AUTH_WHITELIST = {
       "/css/**",
       "/js/**",
       "/media/**",

kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/auth/LdapSecurityConfig.java

@@ -43,7 +43,7 @@ public class LdapSecurityConfig extends AbstractAuthSecurityConfig {
   public ReactiveAuthenticationManager authenticationManager(BaseLdapPathContextSource contextSource) {
     BindAuthenticator ba = new BindAuthenticator(contextSource);
     if (ldapUserDnPattern != null) {
-      ba.setUserDnPatterns(new String[]{ldapUserDnPattern});
+      ba.setUserDnPatterns(new String[] {ldapUserDnPattern});
     }
     if (userFilterSearchFilter != null) {
       LdapUserSearch userSearch =

kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/auth/OAuthSecurityConfig.java

@@ -39,14 +39,14 @@ public class OAuthSecurityConfig extends AbstractAuthSecurityConfig {
         .authenticated();
 
     if (IS_OAUTH2_PRESENT && OAuth2ClasspathGuard.shouldConfigure(this.context)) {
-      OAuth2ClasspathGuard.configure(this.context, http);
+      OAuth2ClasspathGuard.configure(http);
     }
 
     return http.csrf().disable().build();
   }
 
   private static class OAuth2ClasspathGuard {
-    static void configure(ApplicationContext context, ServerHttpSecurity http) {
+    static void configure(ServerHttpSecurity http) {
       http
           .oauth2Login()
           .and()

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/AuthController.java

@@ -15,8 +15,8 @@ import reactor.core.publisher.Mono;
 @Slf4j
 public class AuthController {
 
-  @GetMapping(value = "/auth", produces = { "text/html" })
+  @GetMapping(value = "/auth", produces = {"text/html"})
-  private Mono<byte[]> getAuth(ServerWebExchange exchange) {
+  public Mono<byte[]> getAuth(ServerWebExchange exchange) {
     Mono<CsrfToken> token = exchange.getAttributeOrDefault(CsrfToken.class.getName(), Mono.empty());
     return token
         .map(AuthController::csrfToken)

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/StaticController.java

@@ -1,13 +1,11 @@
 package com.provectus.kafka.ui.controller;
 
 import com.provectus.kafka.ui.util.ResourceUtil;
-import java.util.Optional;
 import java.util.concurrent.atomic.AtomicReference;
 import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.boot.autoconfigure.web.ServerProperties;
 import org.springframework.core.io.Resource;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -24,7 +22,7 @@ public class StaticController {
   private Resource indexFile;
   private final AtomicReference<String> renderedIndexFile = new AtomicReference<>();
 
-  @GetMapping(value = "/index.html", produces = { "text/html" })
+  @GetMapping(value = "/index.html", produces = {"text/html"})
   public Mono<ResponseEntity<String>> getIndex(ServerWebExchange exchange) {
     return Mono.just(ResponseEntity.ok(getRenderedIndexFile(exchange)));
   }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java

@@ -101,10 +101,9 @@ public class TopicsController extends AbstractController implements TopicsApi {
       String clusterName, String topicName,
       Mono<PartitionsIncreaseDTO> partitionsIncrease,
       ServerWebExchange exchange) {
-    return partitionsIncrease.flatMap(
+    return partitionsIncrease.flatMap(partitions ->
-        partitions ->
+            topicsService.increaseTopicPartitions(getCluster(clusterName), topicName, partitions)
-            topicsService.increaseTopicPartitions(getCluster(clusterName), topicName, partitions))
+        ).map(ResponseEntity::ok);
-        .map(ResponseEntity::ok);
   }
 
   @Override

kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/AbstractEmitter.java

@@ -19,7 +19,7 @@ public abstract class AbstractEmitter {
   private final RecordSerDe recordDeserializer;
   private final ConsumingStats consumingStats = new ConsumingStats();
 
-  public AbstractEmitter(RecordSerDe recordDeserializer) {
+  protected AbstractEmitter(RecordSerDe recordDeserializer) {
     this.recordDeserializer = recordDeserializer;
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/ConsumingStats.java

@@ -17,13 +17,13 @@ class ConsumingStats {
   void sendConsumingEvt(FluxSink<TopicMessageEventDTO> sink,
                         ConsumerRecords<Bytes, Bytes> polledRecords,
                         long elapsed) {
-    for (ConsumerRecord<Bytes, Bytes> record : polledRecords) {
+    for (ConsumerRecord<Bytes, Bytes> rec : polledRecords) {
-      for (Header header : record.headers()) {
+      for (Header header : rec.headers()) {
         bytes +=
             (header.key() != null ? header.key().getBytes().length : 0L)
                 + (header.value() != null ? header.value().length : 0L);
       }
-      bytes += record.serializedKeySize() + record.serializedValueSize();
+      bytes += rec.serializedKeySize() + rec.serializedValueSize();
     }
     this.records += polledRecords.count();
     this.elapsed += elapsed;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/emitter/MessageFilters.java

@@ -18,6 +18,9 @@ public class MessageFilters {
 
   private static GroovyScriptEngineImpl GROOVY_ENGINE;
 
+  private MessageFilters() {
+  }
+
   public static Predicate<TopicMessageDTO> createMsgFilter(String query, MessageFilterTypeDTO type) {
     switch (type) {
       case STRING_CONTAINS:

kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/SchemaTypeIsNotSupportedException.java

@@ -1,12 +0,0 @@
-package com.provectus.kafka.ui.exception;
-
-public class SchemaTypeIsNotSupportedException extends UnprocessableEntityException {
-
-  private static final String REQUIRED_SCHEMA_REGISTRY_VERSION = "5.5.0";
-
-  public SchemaTypeIsNotSupportedException() {
-    super(String.format("Current version of Schema Registry does "
-            + "not support provided schema type,"
-            + " version %s or later is required here.", REQUIRED_SCHEMA_REGISTRY_VERSION));
-  }
-}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/exception/SchemaTypeNotSupportedException.java

@@ -0,0 +1,12 @@
+package com.provectus.kafka.ui.exception;
+
+public class SchemaTypeNotSupportedException extends UnprocessableEntityException {
+
+  private static final String REQUIRED_SCHEMA_REGISTRY_VERSION = "5.5.0";
+
+  public SchemaTypeNotSupportedException() {
+    super(String.format("Current version of Schema Registry does "
+        + "not support provided schema type,"
+        + " version %s or later is required here.", REQUIRED_SCHEMA_REGISTRY_VERSION));
+  }
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ConsumerGroupMapper.java

@@ -17,6 +17,9 @@ import org.apache.kafka.common.TopicPartition;
 
 public class ConsumerGroupMapper {
 
+  private ConsumerGroupMapper() {
+  }
+
   public static ConsumerGroupDTO toDto(InternalConsumerGroup c) {
     return convertToConsumerGroup(c, new ConsumerGroupDTO());
   }
@@ -47,7 +50,7 @@ public class ConsumerGroupMapper {
       for (TopicPartition topicPartition : member.getAssignment()) {
         final ConsumerGroupTopicPartitionDTO partition = partitionMap.computeIfAbsent(
             topicPartition,
-            (tp) -> new ConsumerGroupTopicPartitionDTO()
+            tp -> new ConsumerGroupTopicPartitionDTO()
                 .topic(tp.topic())
                 .partition(tp.partition())
         );
@@ -99,12 +102,18 @@ public class ConsumerGroupMapper {
   private static ConsumerGroupStateDTO mapConsumerGroupState(
       org.apache.kafka.common.ConsumerGroupState state) {
     switch (state) {
-      case DEAD: return ConsumerGroupStateDTO.DEAD;
+      case DEAD:
-      case EMPTY: return ConsumerGroupStateDTO.EMPTY;
+        return ConsumerGroupStateDTO.DEAD;
-      case STABLE: return ConsumerGroupStateDTO.STABLE;
+      case EMPTY:
-      case PREPARING_REBALANCE: return ConsumerGroupStateDTO.PREPARING_REBALANCE;
+        return ConsumerGroupStateDTO.EMPTY;
-      case COMPLETING_REBALANCE: return ConsumerGroupStateDTO.COMPLETING_REBALANCE;
+      case STABLE:
-      default: return ConsumerGroupStateDTO.UNKNOWN;
+        return ConsumerGroupStateDTO.STABLE;
+      case PREPARING_REBALANCE:
+        return ConsumerGroupStateDTO.PREPARING_REBALANCE;
+      case COMPLETING_REBALANCE:
+        return ConsumerGroupStateDTO.COMPLETING_REBALANCE;
+      default:
+        return ConsumerGroupStateDTO.UNKNOWN;
     }
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/CleanupPolicy.java

@@ -1,6 +1,5 @@
 package com.provectus.kafka.ui.model;
 
-import com.provectus.kafka.ui.exception.IllegalEntityStateException;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -11,24 +10,24 @@ public enum CleanupPolicy {
   COMPACT_DELETE(Arrays.asList("compact,delete", "delete,compact")),
   UNKNOWN("unknown");
 
-  private final List<String> cleanUpPolicy;
+  private final List<String> policies;
 
-  CleanupPolicy(String cleanUpPolicy) {
+  CleanupPolicy(String policy) {
-    this(Collections.singletonList(cleanUpPolicy));
+    this(Collections.singletonList(policy));
   }
 
-  CleanupPolicy(List<String> cleanUpPolicy) {
+  CleanupPolicy(List<String> policies) {
-    this.cleanUpPolicy = cleanUpPolicy;
+    this.policies = policies;
   }
 
-  public String getCleanUpPolicy() {
+  public String getPolicy() {
-    return cleanUpPolicy.get(0);
+    return policies.get(0);
   }
 
   public static CleanupPolicy fromString(String string) {
     return Arrays.stream(CleanupPolicy.values())
         .filter(v ->
-            v.cleanUpPolicy.stream().anyMatch(
+            v.policies.stream().anyMatch(
                 s -> s.equals(string.replace(" ", "")
                 )
             )

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalClusterMetrics.java

@@ -38,7 +38,7 @@ public class InternalClusterMetrics {
 
   // zk stats
   @Deprecated //use 'zookeeperStatus' field with enum type instead
-  private final int zooKeeperStatus;
+  private final int zooKeeperStatusEnum;
   private final ServerStatusDTO zookeeperStatus;
   private final Throwable lastZookeeperException;
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalClusterState.java

@@ -67,7 +67,7 @@ public class InternalClusterState {
     inSyncReplicasCount = partitionsStats.getInSyncReplicasCount();
     outOfSyncReplicasCount = partitionsStats.getOutOfSyncReplicasCount();
     underReplicatedPartitionCount = partitionsStats.getUnderReplicatedPartitionCount();
-    readOnly = cluster.getReadOnly();
+    readOnly = cluster.isReadOnly();
   }
 
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalConsumerGroup.java

@@ -65,21 +65,21 @@ public class InternalConsumerGroup {
 
   // removes data for all partitions that are not fit filter
   public InternalConsumerGroup retainDataForPartitions(Predicate<TopicPartition> partitionsFilter) {
-    var offsets = getOffsets().entrySet().stream()
+    var offsetsMap = getOffsets().entrySet().stream()
         .filter(e -> partitionsFilter.test(e.getKey()))
         .collect(Collectors.toMap(
             Map.Entry::getKey,
             Map.Entry::getValue
         ));
 
-    var members = getMembers().stream()
+    var nonEmptyMembers = getMembers().stream()
         .map(m -> filterConsumerMemberTopic(m, partitionsFilter))
         .filter(m -> !m.getAssignment().isEmpty())
         .collect(Collectors.toList());
 
     return toBuilder()
-        .offsets(offsets)
+        .offsets(offsetsMap)
-        .members(members)
+        .members(nonEmptyMembers)
         .build();
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java

@@ -30,6 +30,6 @@ public class KafkaCluster {
   private final String protobufMessageName;
   private final Map<String, String> protobufMessageNameByTopic;
   private final Properties properties;
-  private final Boolean readOnly;
+  private final boolean readOnly;
-  private final Boolean disableLogDirsCollection;
+  private final boolean disableLogDirsCollection;
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/schemaregistry/InternalNewSchema.java

@@ -2,7 +2,6 @@ package com.provectus.kafka.ui.model.schemaregistry;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.provectus.kafka.ui.model.SchemaTypeDTO;
-import com.provectus.kafka.ui.model.SchemaTypeDTO;
 import lombok.Data;
 
 @Data

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/RecordSerDe.java

@@ -14,12 +14,18 @@ public interface RecordSerDe {
   @Value
   @Builder
   class DeserializedKeyValue {
-    @Nullable String key;
+    @Nullable
-    @Nullable String value;
+    String key;
-    @Nullable MessageFormat keyFormat;
+    @Nullable
-    @Nullable MessageFormat valueFormat;
+    String value;
-    @Nullable String keySchemaId;
+    @Nullable
-    @Nullable String valueSchemaId;
+    MessageFormat keyFormat;
+    @Nullable
+    MessageFormat valueFormat;
+    @Nullable
+    String keySchemaId;
+    @Nullable
+    String valueSchemaId;
   }
 
   DeserializedKeyValue deserialize(ConsumerRecord<Bytes, Bytes> msg);

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/AvroMessageReader.java

@@ -6,8 +6,8 @@ import io.confluent.kafka.schemaregistry.avro.AvroSchemaUtils;
 import io.confluent.kafka.schemaregistry.client.SchemaMetadata;
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
 import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
+import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import io.confluent.kafka.serializers.KafkaAvroSerializer;
-import io.confluent.kafka.serializers.KafkaAvroSerializerConfig;
 import java.io.IOException;
 import java.util.Map;
 import org.apache.kafka.common.serialization.Serializer;
@@ -27,8 +27,8 @@ public class AvroMessageReader extends MessageReader<Object> {
     serializer.configure(
         Map.of(
             "schema.registry.url", "wontbeused",
-            KafkaAvroSerializerConfig.AUTO_REGISTER_SCHEMAS, false,
+            AbstractKafkaSchemaSerDeConfig.AUTO_REGISTER_SCHEMAS, false,
-            KafkaAvroSerializerConfig.USE_LATEST_VERSION, true
+            AbstractKafkaSchemaSerDeConfig.USE_LATEST_VERSION, true
         ),
         isKey
     );

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/JsonSchemaMessageReader.java

@@ -10,8 +10,8 @@ import io.confluent.kafka.schemaregistry.client.SchemaMetadata;
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
 import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
 import io.confluent.kafka.schemaregistry.json.JsonSchema;
+import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer;
-import io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializerConfig;
 import java.io.IOException;
 import java.util.Map;
 import org.apache.kafka.common.serialization.Serializer;
@@ -33,8 +33,8 @@ public class JsonSchemaMessageReader extends MessageReader<JsonNode> {
     serializer.configure(
         Map.of(
             "schema.registry.url", "wontbeused",
-            KafkaJsonSchemaSerializerConfig.AUTO_REGISTER_SCHEMAS, false,
+            AbstractKafkaSchemaSerDeConfig.AUTO_REGISTER_SCHEMAS, false,
-            KafkaJsonSchemaSerializerConfig.USE_LATEST_VERSION, true
+            AbstractKafkaSchemaSerDeConfig.USE_LATEST_VERSION, true
         ),
         isKey
     );
@@ -69,10 +69,10 @@ public class JsonSchemaMessageReader extends MessageReader<JsonNode> {
      * possible in our case. So, we just skip all infer logic and pass schema directly.
      */
     @Override
-    public byte[] serialize(String topic, JsonNode record) {
+    public byte[] serialize(String topic, JsonNode rec) {
       return super.serializeImpl(
-          super.getSubjectName(topic, isKey, record, schema),
+          super.getSubjectName(topic, isKey, rec, schema),
-          record,
+          rec,
           (JsonSchema) schema
       );
     }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/ProtobufMessageReader.java

@@ -8,8 +8,8 @@ import io.confluent.kafka.schemaregistry.client.SchemaMetadata;
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
 import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
 import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchema;
+import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer;
-import io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializerConfig;
 import java.io.IOException;
 import java.util.Map;
 import org.apache.kafka.common.serialization.Serializer;
@@ -28,8 +28,8 @@ public class ProtobufMessageReader extends MessageReader<Message> {
     serializer.configure(
         Map.of(
             "schema.registry.url", "wontbeused",
-            KafkaProtobufSerializerConfig.AUTO_REGISTER_SCHEMAS, false,
+            AbstractKafkaSchemaSerDeConfig.AUTO_REGISTER_SCHEMAS, false,
-            KafkaProtobufSerializerConfig.USE_LATEST_VERSION, true
+            AbstractKafkaSchemaSerDeConfig.USE_LATEST_VERSION, true
         ),
         isKey
     );

kafka-ui-api/src/main/java/com/provectus/kafka/ui/serde/schemaregistry/SchemaRegistryAwareRecordSerDe.java

@@ -113,7 +113,7 @@ public class SchemaRegistryAwareRecordSerDe implements RecordSerDe {
                                          DeserializedKeyValueBuilder builder) {
     Optional<Integer> schemaId = extractSchemaIdFromMsg(rec, isKey);
     Optional<MessageFormat> format = schemaId.flatMap(this::getMessageFormatBySchemaId);
-    if (format.isPresent() && schemaRegistryFormatters.containsKey(format.get())) {
+    if (schemaId.isPresent() && format.isPresent() && schemaRegistryFormatters.containsKey(format.get())) {
       var formatter = schemaRegistryFormatters.get(format.get());
       try {
         var deserialized = formatter.format(rec.topic(), isKey ? rec.key().get() : rec.value().get());
@@ -135,12 +135,13 @@ public class SchemaRegistryAwareRecordSerDe implements RecordSerDe {
 
     // fallback
     if (isKey) {
-      builder.key(FALLBACK_FORMATTER.format(rec.topic(), isKey ? rec.key().get() : rec.value().get()));
+      builder.key(FALLBACK_FORMATTER.format(rec.topic(), rec.key().get()));
       builder.keyFormat(FALLBACK_FORMATTER.getFormat());
     } else {
-      builder.value(FALLBACK_FORMATTER.format(rec.topic(), isKey ? rec.key().get() : rec.value().get()));
+      builder.value(FALLBACK_FORMATTER.format(rec.topic(), rec.value().get()));
       builder.valueFormat(FALLBACK_FORMATTER.getFormat());
     }
+
   }
 
   @Override
@@ -202,14 +203,14 @@ public class SchemaRegistryAwareRecordSerDe implements RecordSerDe {
 
     final MessageSchemaDTO keySchema = new MessageSchemaDTO()
         .name(maybeKeySchema.map(
-            (s) -> schemaSubject(topic, true)
+            s -> schemaSubject(topic, true)
         ).orElse("unknown"))
         .source(MessageSchemaDTO.SourceEnum.SCHEMA_REGISTRY)
         .schema(sourceKeySchema);
 
     final MessageSchemaDTO valueSchema = new MessageSchemaDTO()
         .name(maybeValueSchema.map(
-            (s) -> schemaSubject(topic, false)
+            s -> schemaSubject(topic, false)
         ).orElse("unknown"))
         .source(MessageSchemaDTO.SourceEnum.SCHEMA_REGISTRY)
         .schema(sourceValueSchema);

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ConsumerGroupService.java

@@ -11,7 +11,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.UUID;
-import java.util.function.Function;
+import java.util.function.ToIntFunction;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 import lombok.RequiredArgsConstructor;
@@ -141,18 +141,25 @@ public class ConsumerGroupService {
       case NAME:
         return Comparator.comparing(ConsumerGroupDescription::groupId);
       case STATE:
-        Function<ConsumerGroupDescription, Integer> statesPriorities = cg -> {
+        ToIntFunction<ConsumerGroupDescription> statesPriorities = cg -> {
           switch (cg.state()) {
-            case STABLE: return 0;
+            case STABLE:
-            case COMPLETING_REBALANCE: return 1;
+              return 0;
-            case PREPARING_REBALANCE: return 2;
+            case COMPLETING_REBALANCE:
-            case EMPTY: return 3;
+              return 1;
-            case DEAD: return 4;
+            case PREPARING_REBALANCE:
-            case UNKNOWN: return 5;
+              return 2;
-            default: return 100;
+            case EMPTY:
+              return 3;
+            case DEAD:
+              return 4;
+            case UNKNOWN:
+              return 5;
+            default:
+              return 100;
           }
         };
-        return Comparator.comparingInt(statesPriorities::apply);
+        return Comparator.comparingInt(statesPriorities);
       case MEMBERS:
         return Comparator.comparingInt(cg -> -cg.members().size());
       default:

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/FeatureService.java

@@ -44,7 +44,7 @@ public class FeatureService {
     if (controller != null) {
       features.add(
           isTopicDeletionEnabled(cluster, controller)
-              .flatMap(r -> r ? Mono.just(Feature.TOPIC_DELETION) : Mono.empty())
+              .flatMap(r -> Boolean.TRUE.equals(r) ? Mono.just(Feature.TOPIC_DELETION) : Mono.empty())
       );
     }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConfigSanitizer.java

@@ -30,7 +30,7 @@ class KafkaConfigSanitizer extends Sanitizer {
       var keysToSanitize = new HashSet<>(
           patternsToSanitize.isEmpty() ? DEFAULT_PATTERNS_TO_SANITIZE : patternsToSanitize);
       keysToSanitize.addAll(kafkaConfigKeysToSanitize());
-      setKeysToSanitize(keysToSanitize.toArray(new String[]{}));
+      setKeysToSanitize(keysToSanitize.toArray(new String[] {}));
     }
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaConnectService.java

@@ -103,7 +103,7 @@ public class KafkaConnectService {
   }
 
   private Predicate<FullConnectorInfoDTO> matchesSearchTerm(final String search) {
-    return (connector) -> getSearchValues(connector)
+    return connector -> getSearchValues(connector)
         .anyMatch(value -> value.contains(
             StringUtils.defaultString(
                     search,
@@ -158,7 +158,7 @@ public class KafkaConnectService {
             connector
                 .flatMap(c -> connectorExists(cluster, connectName, c.getName())
                     .map(exists -> {
-                      if (exists) {
+                      if (Boolean.TRUE.equals(exists)) {
                         throw new ValidationException(
                             String.format("Connector with name %s already exists", c.getName()));
                       }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/MetricsService.java

@@ -60,7 +60,7 @@ public class MetricsService {
   }
 
   private Mono<InternalLogDirStats> getLogDirInfo(KafkaCluster cluster, ReactiveAdminClient c) {
-    if (cluster.getDisableLogDirsCollection() == null || !cluster.getDisableLogDirsCollection()) {
+    if (!cluster.isDisableLogDirsCollection()) {
       return c.describeLogDirs().map(InternalLogDirStats::new);
     }
     return Mono.just(InternalLogDirStats.empty());

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/SchemaRegistryService.java

@@ -5,7 +5,7 @@ import static org.springframework.http.HttpStatus.UNPROCESSABLE_ENTITY;
 
 import com.provectus.kafka.ui.exception.SchemaFailedToDeleteException;
 import com.provectus.kafka.ui.exception.SchemaNotFoundException;
-import com.provectus.kafka.ui.exception.SchemaTypeIsNotSupportedException;
+import com.provectus.kafka.ui.exception.SchemaTypeNotSupportedException;
 import com.provectus.kafka.ui.exception.UnprocessableEntityException;
 import com.provectus.kafka.ui.exception.ValidationException;
 import com.provectus.kafka.ui.mapper.ClusterMapper;
@@ -212,7 +212,7 @@ public class SchemaRegistryService {
         .onStatus(UNPROCESSABLE_ENTITY::equals,
             r -> r.bodyToMono(ErrorResponse.class)
                 .flatMap(x -> Mono.error(isUnrecognizedFieldSchemaTypeMessage(x.getMessage())
-                    ? new SchemaTypeIsNotSupportedException()
+                    ? new SchemaTypeNotSupportedException()
                     : new UnprocessableEntityException(x.getMessage()))))
         .bodyToMono(SubjectIdResponse.class);
   }
@@ -294,7 +294,9 @@ public class SchemaRegistryService {
   }
 
   public String formatted(String str, Object... args) {
-    return new Formatter().format(str, args).toString();
+    try (Formatter formatter = new Formatter()) {
+      return formatter.format(str, args).toString();
+    }
   }
 
   private void setBasicAuthIfEnabled(InternalSchemaRegistry schemaRegistry, HttpHeaders headers) {

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java

@@ -206,7 +206,7 @@ public class TopicsService {
                     .thenReturn(topicName))
                 .retryWhen(Retry.fixedDelay(recreateMaxRetries,
                         Duration.ofSeconds(recreateDelayInSeconds))
-                                    .filter(throwable -> throwable instanceof TopicExistsException)
+                    .filter(TopicExistsException.class::isInstance)
                     .onRetryExhaustedThrow((a, b) ->
                         new TopicRecreationException(topicName,
                             recreateMaxRetries * recreateDelayInSeconds)))
@@ -403,10 +403,11 @@ public class TopicsService {
           );
           return ac.createPartitions(newPartitionsMap)
               .then(loadTopic(cluster, topicName));
-        })
+        }).map(t -> new PartitionsIncreaseResponseDTO()
-        .map(t -> new PartitionsIncreaseResponseDTO()
             .topicName(t.getName())
-            .totalPartitionsCount(t.getPartitionCount())));
+            .totalPartitionsCount(t.getPartitionCount())
+        )
+    );
   }
 
   public Mono<Void> deleteTopic(KafkaCluster cluster, String topicName) {

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ZookeeperService.java

@@ -15,7 +15,6 @@ import org.apache.zookeeper.ZooKeeper;
 import org.springframework.stereotype.Service;
 import org.springframework.util.StringUtils;
 import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Schedulers;
 
 @Service
 @RequiredArgsConstructor
@@ -82,7 +81,8 @@ public class ZookeeperService {
 
   private ZooKeeper createClient(KafkaCluster cluster) {
     try {
-      return new ZooKeeper(cluster.getZookeeper(), 60 * 1000, watchedEvent -> {});
+      return new ZooKeeper(cluster.getZookeeper(), 60 * 1000, watchedEvent -> {
+      });
     } catch (IOException e) {
       log.error("Error while creating a zookeeper client for cluster [{}]",
           cluster.getName());

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlApiClient.java

@@ -101,7 +101,7 @@ public class KsqlApiClient {
     if (parsed.getStatements().size() > 1) {
       throw new ValidationException("Only single statement supported now");
     }
-    if (parsed.getStatements().size() == 0) {
+    if (parsed.getStatements().isEmpty()) {
       throw new ValidationException("No valid ksql statement found");
     }
     if (KsqlGrammar.isSelect(parsed.getStatements().get(0))) {

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/KsqlGrammar.java

@@ -18,6 +18,9 @@ import org.antlr.v4.runtime.atn.PredictionMode;
 
 class KsqlGrammar {
 
+  private KsqlGrammar() {
+  }
+
   @Value
   static class KsqlStatements {
     List<KsqlGrammarParser.SingleStatementContext> statements;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/response/DynamicParser.java

@@ -12,6 +12,9 @@ import java.util.stream.StreamSupport;
 
 class DynamicParser {
 
+  private DynamicParser() {
+  }
+
   static KsqlResponseTable parseArray(String tableName, JsonNode array) {
     return parseArray(tableName, getFieldNamesFromArray(array), array);
   }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/ksql/response/ResponseParser.java

@@ -14,6 +14,9 @@ import org.springframework.web.reactive.function.client.WebClientResponseExcepti
 
 public class ResponseParser {
 
+  private ResponseParser() {
+  }
+
   public static Optional<KsqlApiClient.KsqlResponseTable> parseSelectResponse(JsonNode jsonNode) {
     // in response we getting either header record or row data
     if (arrayFieldNonEmpty(jsonNode, "header")) {

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/ClusterUtil.java

@@ -18,6 +18,9 @@ import org.apache.kafka.common.utils.Bytes;
 @Slf4j
 public class ClusterUtil {
 
+  private ClusterUtil() {
+  }
+
   private static final ZoneId UTC_ZONE_ID = ZoneId.of("UTC");
 
   public static int convertToIntServerStatus(ServerStatusDTO serverStatus) {

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/JmxPoolFactory.java

@@ -21,7 +21,7 @@ public class JmxPoolFactory extends BaseKeyedPooledObjectFactory<JmxConnectionIn
   public JMXConnector create(JmxConnectionInfo info) throws Exception {
     Map<String, Object> env = new HashMap<>();
     if (StringUtils.isNotEmpty(info.getUsername()) && StringUtils.isNotEmpty(info.getPassword())) {
-      env.put("jmx.remote.credentials", new String[]{info.getUsername(), info.getPassword()});
+      env.put("jmx.remote.credentials", new String[] {info.getUsername(), info.getPassword()});
     }
 
     if (info.isSsl()) {

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/KafkaConstants.java

@@ -30,20 +30,21 @@ import java.util.Map;
 
 public final class KafkaConstants {
 
+  private static final String LONG_MAX_STRING = Long.valueOf(Long.MAX_VALUE).toString();
+
   public static final Map<String, String> TOPIC_DEFAULT_CONFIGS = Map.ofEntries(
       new AbstractMap.SimpleEntry<>(CLEANUP_POLICY_CONFIG, CLEANUP_POLICY_DELETE),
       new AbstractMap.SimpleEntry<>(COMPRESSION_TYPE_CONFIG, "producer"),
       new AbstractMap.SimpleEntry<>(DELETE_RETENTION_MS_CONFIG, "86400000"),
       new AbstractMap.SimpleEntry<>(FILE_DELETE_DELAY_MS_CONFIG, "60000"),
-      new AbstractMap.SimpleEntry<>(FLUSH_MESSAGES_INTERVAL_CONFIG, "9223372036854775807"),
+      new AbstractMap.SimpleEntry<>(FLUSH_MESSAGES_INTERVAL_CONFIG, LONG_MAX_STRING),
-      new AbstractMap.SimpleEntry<>(FLUSH_MS_CONFIG, "9223372036854775807"),
+      new AbstractMap.SimpleEntry<>(FLUSH_MS_CONFIG, LONG_MAX_STRING),
       new AbstractMap.SimpleEntry<>("follower.replication.throttled.replicas", ""),
       new AbstractMap.SimpleEntry<>(INDEX_INTERVAL_BYTES_CONFIG, "4096"),
       new AbstractMap.SimpleEntry<>("leader.replication.throttled.replicas", ""),
-      new AbstractMap.SimpleEntry<>(MAX_COMPACTION_LAG_MS_CONFIG, "9223372036854775807"),
+      new AbstractMap.SimpleEntry<>(MAX_COMPACTION_LAG_MS_CONFIG, LONG_MAX_STRING),
       new AbstractMap.SimpleEntry<>(MAX_MESSAGE_BYTES_CONFIG, "1000012"),
-      new AbstractMap.SimpleEntry<>(MESSAGE_TIMESTAMP_DIFFERENCE_MAX_MS_CONFIG,
+      new AbstractMap.SimpleEntry<>(MESSAGE_TIMESTAMP_DIFFERENCE_MAX_MS_CONFIG, LONG_MAX_STRING),
-          "9223372036854775807"),
       new AbstractMap.SimpleEntry<>(MESSAGE_TIMESTAMP_TYPE_CONFIG, "CreateTime"),
       new AbstractMap.SimpleEntry<>(MIN_CLEANABLE_DIRTY_RATIO_CONFIG, "0.5"),
       new AbstractMap.SimpleEntry<>(MIN_COMPACTION_LAG_MS_CONFIG, "0"),

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/OffsetsSeekBackward.java

@@ -51,7 +51,7 @@ public class OffsetsSeekBackward extends OffsetsSeek {
         consumerPosition.getSeekTo().entrySet().stream()
             .collect(Collectors.toMap(
                 Map.Entry::getKey,
-                e -> e.getValue()
+                Map.Entry::getValue
             ));
     Map<TopicPartition, Long> offsetsForTimestamps = consumer.offsetsForTimes(timestampsToSearch)
         .entrySet().stream()

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/AvroJsonSchemaConverter.java

@@ -59,7 +59,8 @@ public class AvroJsonSchemaConverter implements JsonSchemaConverter<Schema> {
           }
         case ARRAY:
           return createArraySchema(name, schema, definitions);
-        default: throw new RuntimeException("Unknown type");
+        default:
+          throw new RuntimeException("Unknown type");
       }
     } else {
       return createUnionSchema(schema, definitions);
@@ -138,14 +139,18 @@ public class AvroJsonSchemaConverter implements JsonSchemaConverter<Schema> {
       case BYTES:
       case STRING:
         return new SimpleJsonType(JsonType.Type.STRING);
-      case NULL: return new SimpleJsonType(JsonType.Type.NULL);
+      case NULL:
-      case ARRAY: return new SimpleJsonType(JsonType.Type.ARRAY);
+        return new SimpleJsonType(JsonType.Type.NULL);
+      case ARRAY:
+        return new SimpleJsonType(JsonType.Type.ARRAY);
       case FIXED:
       case FLOAT:
       case DOUBLE:
         return new SimpleJsonType(JsonType.Type.NUMBER);
-      case BOOLEAN: return new SimpleJsonType(JsonType.Type.BOOLEAN);
+      case BOOLEAN:
-      default: return new SimpleJsonType(JsonType.Type.STRING);
+        return new SimpleJsonType(JsonType.Type.BOOLEAN);
+      default:
+        return new SimpleJsonType(JsonType.Type.STRING);
     }
   }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/JsonType.java

@@ -8,7 +8,7 @@ public abstract class JsonType {
 
   protected final Type type;
 
-  public JsonType(Type type) {
+  protected JsonType(Type type) {
     this.type = type;
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/jsonschema/ObjectFieldSchema.java

@@ -2,10 +2,7 @@ package com.provectus.kafka.ui.util.jsonschema;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ArrayNode;
-import com.fasterxml.jackson.databind.node.BooleanNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
-import com.fasterxml.jackson.databind.node.TextNode;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;