fixes

Author: germanosin

api/build.gradle

@@ -107,7 +107,7 @@ generateGrammarSource {
 }
 
 tasks.withType(JavaCompile) {
-    options.compilerArgs << "-Xlint:deprecation"
+    options.compilerArgs << "-Xlint:deprecation" << "-Xlint:unchecked"
 }
 
 

api/src/main/java/io/kafbat/ui/serdes/builtin/sr/Serialize.java

@@ -96,6 +96,7 @@ static byte[] serializeProto(SchemaRegistryClient srClient,
 
   @KafkaClientInternalsDependant("AbstractKafkaAvroSerializer::serializeImpl")
   @SneakyThrows
+  @SuppressWarnings("unchecked")
   static byte[] serializeAvro(AvroSchema schema, int schemaId, String input) {
     var avroObject = JsonAvroConversion.convertJsonToAvro(input, schema.rawSchema());
     try (var out = new ByteArrayOutputStream()) {

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

@@ -377,6 +377,7 @@ public Mono<TopicDescription> describeTopic(String topic) {
    * This method converts input map into Mono[Map] ignoring keys for which KafkaFutures
    * finished with <code>classes</code> exceptions and empty Monos.
    */
+  @SuppressWarnings("unchecked")
   @SafeVarargs
   static <K, V> Mono<Map<K, V>> toMonoWithExceptionFilter(Map<K, KafkaFuture<V>> values,
                                                           Class<? extends KafkaException>... classes) {

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

@@ -34,6 +34,7 @@ public Mono<Statistics> updateCache(KafkaCluster c) {
     return getStatistics(c).doOnSuccess(m -> cache.replace(c, m));
   }
 
+  @SuppressWarnings("unchecked")
   private Mono<Statistics> getStatistics(KafkaCluster cluster) {
     return adminClientService.get(cluster).flatMap(ac ->
             ac.describeCluster().flatMap(description ->

api/src/main/java/io/kafbat/ui/util/jsonschema/JsonAvroConversion.java

@@ -195,6 +195,7 @@ private static Object convert(JsonNode node, Schema avroSchema) {
 
   // converts output of KafkaAvroDeserializer (with AVRO_USE_LOGICAL_TYPE_CONVERTERS flat enabled!) into json.
   // Note: conversion should be compatible with AvroJsonSchemaConverter logic!
+  @SuppressWarnings("unchecked")
   public static JsonNode convertAvroToJson(Object obj, Schema avroSchema) {
     if (obj == null) {
       return NullNode.getInstance();
@@ -213,7 +214,7 @@ public static JsonNode convertAvroToJson(Object obj, Schema avroSchema) {
       }
       case MAP -> {
         ObjectNode node = MAPPER.createObjectNode();
-        ((Map) obj).forEach((k, v) -> node.set(k.toString(), convertAvroToJson(v, avroSchema.getValueType())));
+        ((Map<?, ?>) obj).forEach((k, v) -> node.set(k.toString(), convertAvroToJson(v, avroSchema.getValueType())));
         yield node;
       }
       case ARRAY -> {

api/src/test/java/io/kafbat/ui/AbstractIntegrationTest.java

@@ -40,8 +40,11 @@ public abstract class AbstractIntegrationTest {
 
   private static final String CONFLUENT_PLATFORM_VERSION = "7.8.0";
 
-  public static final ConfluentKafkaContainer kafka = new ConfluentKafkaContainer(
-      DockerImageName.parse("confluentinc/cp-kafka").withTag(CONFLUENT_PLATFORM_VERSION))
+  public static final ConfluentKafkaContainer kafkaOriginal = new ConfluentKafkaContainer(
+      DockerImageName.parse("confluentinc/cp-kafka").withTag(CONFLUENT_PLATFORM_VERSION));
+
+  public static final ConfluentKafkaContainer kafka = kafkaOriginal
+      .withListener("0.0.0.0:9095", () -> kafkaOriginal.getNetworkAliases().getFirst() + ":9095")
       .withNetwork(Network.SHARED);
 
   public static final SchemaRegistryContainer schemaRegistry =

api/src/test/java/io/kafbat/ui/container/KafkaConnectContainer.java

@@ -19,7 +19,7 @@ public KafkaConnectContainer(String version) {
 
 
   public KafkaConnectContainer withKafka(ConfluentKafkaContainer kafka) {
-    String bootstrapServers = kafka.getNetworkAliases().getFirst() + ":9092";
+    String bootstrapServers = kafka.getNetworkAliases().getFirst() + ":9095";
     return withKafka(kafka.getNetwork(), bootstrapServers);
   }
 

api/src/test/java/io/kafbat/ui/container/KsqlDbContainer.java

@@ -22,7 +22,7 @@ public KsqlDbContainer(DockerImageName imageName) {
 
   public KsqlDbContainer withKafka(ConfluentKafkaContainer kafka) {
     dependsOn(kafka);
-    String bootstrapServers = kafka.getNetworkAliases().getFirst() + ":9092";
+    String bootstrapServers = kafka.getNetworkAliases().getFirst() + ":9095";
     return withKafka(kafka.getNetwork(), bootstrapServers);
   }
 

api/src/test/java/io/kafbat/ui/container/SchemaRegistryContainer.java

@@ -3,7 +3,6 @@
 import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
 import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.containers.Network;
 import org.testcontainers.kafka.ConfluentKafkaContainer;
 
@@ -16,7 +15,7 @@ public SchemaRegistryContainer(String version) {
   }
 
   public SchemaRegistryContainer withKafka(ConfluentKafkaContainer kafka) {
-    String bootstrapServers = kafka.getNetworkAliases().get(0) + ":9092";
+    String bootstrapServers = kafka.getNetworkAliases().getFirst() + ":9095";
     return withKafka(kafka.getNetwork(), bootstrapServers);
   }
 

api/src/test/java/io/kafbat/ui/service/audit/AuditServiceTest.java

@@ -26,6 +26,7 @@
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Signal;
 
+@SuppressWarnings("unchecked")
 class AuditServiceTest {
 
   @Test
@@ -59,6 +60,7 @@ void auditCallsWriterMethodDependingOnSignal() {
     verify(auditWriter).write(any(), any(), eq(th));
   }
 
+  @SuppressWarnings("unchecked")
   @Nested
   class CreateAuditWriter {