refactor: use record classes

Author: yeikel

api/src/main/java/io/kafbat/ui/model/InternalClusterState.java

@@ -1,9 +1,6 @@
 package io.kafbat.ui.model;
 
 import com.google.common.base.Throwables;
-import io.kafbat.ui.model.BrokerDiskUsageDTO;
-import io.kafbat.ui.model.MetricsCollectionErrorDTO;
-import io.kafbat.ui.model.ServerStatusDTO;
 import java.math.BigDecimal;
 import java.util.List;
 import java.util.Optional;
@@ -40,8 +37,8 @@ public InternalClusterState(KafkaCluster cluster, Statistics statistics) {
             .stackTrace(Throwables.getStackTraceAsString(e)))
         .orElse(null);
     topicCount = statistics.getTopicDescriptions().size();
-    brokerCount = statistics.getClusterDescription().getNodes().size();
-    activeControllers = Optional.ofNullable(statistics.getClusterDescription().getController())
+    brokerCount = statistics.getClusterDescription().nodes().size();
+    activeControllers = Optional.ofNullable(statistics.getClusterDescription().controller())
         .map(Node::id)
         .orElse(null);
     version = statistics.getVersion();

api/src/main/java/io/kafbat/ui/model/InternalPartitionsOffsets.java

@@ -4,16 +4,11 @@
 import com.google.common.collect.Table;
 import java.util.Map;
 import java.util.Optional;
-import lombok.Value;
 import org.apache.kafka.common.TopicPartition;
 
-
 public class InternalPartitionsOffsets {
 
-  @Value
-  public static class Offsets {
-    Long earliest;
-    Long latest;
+  public record Offsets(Long earliest, Long latest) {
   }
 
   private final Table<String, Integer, Offsets> offsets = HashBasedTable.create();

api/src/main/java/io/kafbat/ui/model/InternalReplica.java

@@ -1,14 +1,7 @@
 package io.kafbat.ui.model;
 
 import lombok.Builder;
-import lombok.Data;
-import lombok.RequiredArgsConstructor;
 
-@Data
 @Builder
-@RequiredArgsConstructor
-public class InternalReplica {
-  private final int broker;
-  private final boolean leader;
-  private final boolean inSync;
+public record InternalReplica(int broker, boolean leader, boolean inSync) {
 }

api/src/main/java/io/kafbat/ui/model/InternalTopic.java

@@ -77,8 +77,8 @@ public static InternalTopic from(TopicDescription topicDescription,
 
           partitionsOffsets.get(topicDescription.name(), partition.partition())
               .ifPresent(offsets -> {
-                partitionDto.offsetMin(offsets.getEarliest());
-                partitionDto.offsetMax(offsets.getLatest());
+                partitionDto.offsetMin(offsets.earliest());
+                partitionDto.offsetMax(offsets.latest());
               });
 
           var segmentStats =

api/src/main/java/io/kafbat/ui/serdes/ConsumerRecordDeserializer.java

@@ -79,14 +79,14 @@ private void fillKey(TopicMessageDTO message, ConsumerRecord<Bytes, Bytes> rec)
     }
     try {
       var deserResult = keyDeserializer.deserialize(new RecordHeadersImpl(), rec.key().get());
-      message.setKey(deserResult.getResult());
+      message.setKey(deserResult.result());
       message.setKeySerde(keySerdeName);
-      message.setKeyDeserializeProperties(deserResult.getAdditionalProperties());
+      message.setKeyDeserializeProperties(deserResult.additionalProperties());
     } catch (Exception e) {
       log.trace("Error deserializing key for key topic: {}, partition {}, offset {}, with serde {}",
           rec.topic(), rec.partition(), rec.offset(), keySerdeName, e);
       var deserResult = fallbackKeyDeserializer.deserialize(new RecordHeadersImpl(), rec.key().get());
-      message.setKey(deserResult.getResult());
+      message.setKey(deserResult.result());
       message.setKeySerde(fallbackSerdeName);
     }
   }
@@ -98,15 +98,15 @@ private void fillValue(TopicMessageDTO message, ConsumerRecord<Bytes, Bytes> rec
     try {
       var deserResult = valueDeserializer.deserialize(
           new RecordHeadersImpl(rec.headers()), rec.value().get());
-      message.setContent(deserResult.getResult());
+      message.setContent(deserResult.result());
       message.setValueSerde(valueSerdeName);
-      message.setValueDeserializeProperties(deserResult.getAdditionalProperties());
+      message.setValueDeserializeProperties(deserResult.additionalProperties());
     } catch (Exception e) {
       log.trace("Error deserializing key for value topic: {}, partition {}, offset {}, with serde {}",
           rec.topic(), rec.partition(), rec.offset(), valueSerdeName, e);
       var deserResult = fallbackValueDeserializer.deserialize(
           new RecordHeadersImpl(rec.headers()), rec.value().get());
-      message.setContent(deserResult.getResult());
+      message.setContent(deserResult.result());
       message.setValueSerde(fallbackSerdeName);
     }
   }

api/src/main/java/io/kafbat/ui/serdes/CustomSerdeLoader.java

@@ -18,15 +18,10 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 import lombok.SneakyThrows;
-import lombok.Value;
-
 
 class CustomSerdeLoader {
 
-  @Value
-  static class CustomSerde {
-    Serde serde;
-    ClassLoader classLoader;
+  record CustomSerde(Serde serde, ClassLoader classLoader) {
   }
 
   // serde location -> classloader

api/src/main/java/io/kafbat/ui/serdes/SerdesInitializer.java

@@ -266,10 +266,10 @@ private SerdeInstance loadAndInitCustomSerde(ClustersProperties.SerdeConfig serd
         serdeConfig.getClassName(), serdeConfig.getFilePath(), serdeProps, clusterProps, globalProps);
     return new SerdeInstance(
         serdeConfig.getName(),
-        loaded.getSerde(),
+        loaded.serde(),
         nullablePattern(serdeConfig.getTopicKeysPattern()),
         nullablePattern(serdeConfig.getTopicValuesPattern()),
-        loaded.getClassLoader()
+        loaded.classLoader()
     );
   }
 

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

@@ -52,7 +52,7 @@ private Mono<List<ConfigEntry>> loadBrokersConfig(
   }
 
   private Flux<InternalBrokerConfig> getBrokersConfig(KafkaCluster cluster, Integer brokerId) {
-    if (statisticsCache.get(cluster).getClusterDescription().getNodes()
+    if (statisticsCache.get(cluster).getClusterDescription().nodes()
         .stream().noneMatch(node -> node.id() == brokerId)) {
       return Flux.error(
           new NotFoundException(String.format("Broker with id %s not found", brokerId)));
@@ -70,7 +70,7 @@ public Flux<InternalBroker> getBrokers(KafkaCluster cluster) {
     return adminClientService
         .get(cluster)
         .flatMap(ReactiveAdminClient::describeCluster)
-        .map(description -> description.getNodes().stream()
+        .map(description -> description.nodes().stream()
             .map(node -> new InternalBroker(node, partitionsDistribution, stats))
             .collect(Collectors.toList()))
         .flatMapMany(Flux::fromIterable);
@@ -113,7 +113,7 @@ private Mono<Map<Integer, Map<String, DescribeLogDirsResponse.LogDirInfo>>> getC
       KafkaCluster cluster, List<Integer> reqBrokers) {
     return adminClientService.get(cluster)
         .flatMap(admin -> {
-          List<Integer> brokers = statisticsCache.get(cluster).getClusterDescription().getNodes()
+          List<Integer> brokers = statisticsCache.get(cluster).getClusterDescription().nodes()
               .stream()
               .map(Node::id)
               .collect(Collectors.toList());

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

@@ -143,8 +143,8 @@ private SerdeDescriptionDTO toDto(SerdeInstance serdeInstance,
     return new SerdeDescriptionDTO()
         .name(serdeInstance.getName())
         .description(serdeInstance.description().orElse(null))
-        .schema(schemaOpt.map(SchemaDescription::getSchema).orElse(null))
-        .additionalProperties(schemaOpt.map(SchemaDescription::getAdditionalProperties).orElse(null))
+        .schema(schemaOpt.map(SchemaDescription::schema).orElse(null))
+        .additionalProperties(schemaOpt.map(SchemaDescription::additionalProperties).orElse(null))
         .preferred(preferred);
   }
 

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

@@ -61,7 +61,7 @@ private Mono<ClusterFeature> quotaManagement(ReactiveAdminClient adminClient) {
   }
 
   private Mono<ClusterFeature> aclEdit(ReactiveAdminClient adminClient, ClusterDescription clusterDescription) {
-    var authorizedOps = Optional.ofNullable(clusterDescription.getAuthorizedOperations()).orElse(Set.of());
+    var authorizedOps = Optional.ofNullable(clusterDescription.authorizedOperations()).orElse(Set.of());
     boolean canEdit = aclViewEnabled(adminClient)
         && (authorizedOps.contains(AclOperation.ALL) || authorizedOps.contains(AclOperation.ALTER));
     return canEdit