Merge branch 'main' into main

Author: germanosin

api/build.gradle

@@ -62,6 +62,15 @@ dependencies {
     implementation libs.netty.common
     implementation libs.netty.handler
 
+
+    // Google Managed Service for Kafka IAM support
+    implementation (libs.google.managed.kafka.login.handler) {
+        exclude group: 'com.google.oauth-client', module: 'google-oauth-client'
+    }
+    implementation (libs.google.oauth.client) {
+        because("CVE Fix: It is excluded above because of a vulnerability")
+    }
+
     // Annotation processors
     implementation libs.lombok
     implementation libs.mapstruct

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

@@ -7,6 +7,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableCollection;
 import com.google.common.collect.ImmutableSet;
+import com.google.protobuf.NullValue;
 import dev.cel.common.CelAbstractSyntaxTree;
 import dev.cel.common.CelOptions;
 import dev.cel.common.CelValidationException;
@@ -26,6 +27,7 @@
 import io.kafbat.ui.exception.CelException;
 import io.kafbat.ui.model.TopicMessageDTO;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -38,11 +40,13 @@
 @Slf4j
 @UtilityClass
 public class MessageFilters {
+
   private static final String CEL_RECORD_VAR_NAME = "record";
   private static final String CEL_RECORD_TYPE_NAME = TopicMessageDTO.class.getSimpleName();
 
   private static final CelCompiler CEL_COMPILER = createCompiler();
   private static final CelRuntime CEL_RUNTIME = createRuntime();
+  private static final Object CELL_NULL_VALUE = NullValue.NULL_VALUE;
 
   private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
@@ -188,10 +192,34 @@ private static Object parseToJsonOrReturnAsIs(@Nullable String str) {
     }
 
     try {
-      return OBJECT_MAPPER.readValue(str, new TypeReference<Map<String, Object>>() {
-      });
+      //@formatter:off
+      var map = OBJECT_MAPPER.readValue(str, new TypeReference<Map<String, Object>>() {});
+      //@formatter:on
+      return replaceCelNulls(map);
     } catch (JsonProcessingException e) {
       return str;
     }
   }
+
+  @SuppressWarnings("unchecked")
+  private static Map<String, Object> replaceCelNulls(Map<String, Object> map) {
+    var result = new LinkedHashMap<String, Object>();
+
+    for (var entry : map.entrySet()) {
+      String key = entry.getKey();
+      Object value = entry.getValue();
+
+      if (value == null) {
+        result.put(key, CELL_NULL_VALUE);
+      } else if (value instanceof Map<?, ?>) {
+        var inner = (Map<String, Object>) value;
+        result.put(key, replaceCelNulls(inner));
+      } else {
+        result.put(key, value);
+      }
+    }
+
+    return result;
+  }
+
 }

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

@@ -29,7 +29,7 @@ class KafkaConfigSanitizer {
       .addAll(kafkaConfigKeysToSanitize())
       .add(
           "basic.auth.user.info",  /* For Schema Registry credentials */
-          "password", "secret", "token", "key", ".*credentials.*",   /* General credential patterns */
+          "password", "secret", "token", "key", ".*credentials.*", "passphrase",   /* General credential patterns */
           "aws.access.*", "aws.secret.*", "aws.session.*",   /* AWS-related credential patterns */
           "connection.uri" /* mongo credential patterns */
       )

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

@@ -13,6 +13,7 @@
 import io.kafbat.ui.exception.NotFoundException;
 import io.kafbat.ui.exception.ValidationException;
 import io.kafbat.ui.util.KafkaVersion;
+import io.kafbat.ui.util.MetadataVersion;
 import io.kafbat.ui.util.annotation.KafkaClientInternalsDependant;
 import java.io.Closeable;
 import java.time.Duration;
@@ -49,6 +50,8 @@
 import org.apache.kafka.clients.admin.DescribeClusterOptions;
 import org.apache.kafka.clients.admin.DescribeClusterResult;
 import org.apache.kafka.clients.admin.DescribeConfigsOptions;
+import org.apache.kafka.clients.admin.FeatureMetadata;
+import org.apache.kafka.clients.admin.FinalizedVersionRange;
 import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsSpec;
 import org.apache.kafka.clients.admin.ListOffsetsResult;
 import org.apache.kafka.clients.admin.ListTopicsOptions;
@@ -96,6 +99,7 @@
 @Slf4j
 @AllArgsConstructor
 public class ReactiveAdminClient implements Closeable {
+  private static final String DEFAULT_UNKNOWN_VERSION = "Unknown";
 
   public enum SupportedFeature {
     INCREMENTAL_ALTER_CONFIGS(2.3f),
@@ -114,8 +118,8 @@ public enum SupportedFeature {
       this.predicate = (admin, ver) -> Mono.just(ver != null && ver >= fromVersion);
     }
 
-    static Mono<Set<SupportedFeature>> forVersion(AdminClient ac, String kafkaVersionStr) {
-      @Nullable Float kafkaVersion = KafkaVersion.parse(kafkaVersionStr).orElse(null);
+    static Mono<Set<SupportedFeature>> forVersion(AdminClient ac, Optional<String> kafkaVersionStr) {
+      @Nullable Float kafkaVersion = kafkaVersionStr.flatMap(KafkaVersion::parse).orElse(null);
       return Flux.fromArray(SupportedFeature.values())
           .flatMap(f -> f.predicate.apply(ac, kafkaVersion).map(enabled -> Tuples.of(f, enabled)))
           .filter(Tuple2::getT2)
@@ -150,18 +154,28 @@ private static Mono<ConfigRelatedInfo> extract(AdminClient ac) {
                 .orElse(desc.getNodes().iterator().next().id());
             return loadBrokersConfig(ac, List.of(targetNodeId))
                 .map(map -> map.isEmpty() ? List.<ConfigEntry>of() : map.get(targetNodeId))
-                .flatMap(configs -> {
-                  String version = "1.0-UNKNOWN";
+                .zipWith(toMono(ac.describeFeatures().featureMetadata()))
+                .flatMap(tuple -> {
+                  List<ConfigEntry> configs = tuple.getT1();
+                  FeatureMetadata featureMetadata = tuple.getT2();
+                  Optional<String> version = Optional.empty();
                   boolean topicDeletionEnabled = true;
                   for (ConfigEntry entry : configs) {
                     if (entry.name().contains("inter.broker.protocol.version")) {
-                      version = entry.value();
+                      version = Optional.of(entry.value());
                     }
                     if (entry.name().equals("delete.topic.enable")) {
                       topicDeletionEnabled = Boolean.parseBoolean(entry.value());
                     }
                   }
-                  final String finalVersion = version;
+                  if (version.isEmpty()) {
+                    FinalizedVersionRange metadataVersion =
+                        featureMetadata.finalizedFeatures().get("metadata.version");
+                    if (metadataVersion != null) {
+                      version = MetadataVersion.findVersion(metadataVersion.maxVersionLevel());
+                    }
+                  }
+                  final String finalVersion = version.orElse(DEFAULT_UNKNOWN_VERSION);
                   final boolean finalTopicDeletionEnabled = topicDeletionEnabled;
                   return SupportedFeature.forVersion(ac, version)
                       .map(features -> new ConfigRelatedInfo(finalVersion, features, finalTopicDeletionEnabled));

api/src/main/java/io/kafbat/ui/util/MetadataVersion.java

@@ -0,0 +1,48 @@
+package io.kafbat.ui.util;
+
+import java.util.Arrays;
+import java.util.Optional;
+
+public enum MetadataVersion {
+  IBP_3_0_IV1(1, "3.0-IV1"),
+  IBP_3_1_IV0(2, "3.1-IV0"),
+  IBP_3_2_IV0(3, "3.2-IV0"),
+  IBP_3_3_IV0(4, "3.3-IV0"),
+  IBP_3_3_IV1(5, "3.3-IV1"),
+  IBP_3_3_IV2(6, "3.3-IV2"),
+  IBP_3_3_IV3(7, "3.3-IV3"),
+  IBP_3_4_IV0(8, "3.4-IV0"),
+  IBP_3_5_IV0(9, "3.5-IV0"),
+  IBP_3_5_IV1(10, "3.5-IV1"),
+  IBP_3_5_IV2(11, "3.5-IV2"),
+  IBP_3_6_IV0(12, "3.6-IV0"),
+  IBP_3_6_IV1(13, "3.6-IV1"),
+  IBP_3_6_IV2(14, "3.6-IV2"),
+  IBP_3_7_IV0(15, "3.7-IV0"),
+  IBP_3_7_IV1(16, "3.7-IV1"),
+  IBP_3_7_IV2(17, "3.7-IV2"),
+  IBP_3_7_IV3(18, "3.7-IV3"),
+  IBP_3_7_IV4(19, "3.7-IV4"),
+  IBP_3_8_IV0(20, "3.8-IV0"),
+  IBP_3_9_IV0(21, "3.9-IV0"),
+  IBP_4_0_IV0(22, "4.0-IV0"),
+  IBP_4_0_IV1(23, "4.0-IV1"),
+  IBP_4_0_IV2(24, "4.0-IV2"),
+  IBP_4_0_IV3(25, "4.0-IV3"),
+  IBP_4_1_IV0(26, "4.1-IV0");
+
+  private final int featureLevel;
+  private final String release;
+
+  MetadataVersion(int featureLevel, String release) {
+    this.featureLevel = featureLevel;
+    this.release = release;
+  }
+
+  public static Optional<String> findVersion(int featureLevel) {
+    return Arrays.stream(values())
+        .filter(v -> v.featureLevel == featureLevel)
+        .findFirst().map(v -> v.release);
+  }
+
+}

api/src/test/java/io/kafbat/ui/emitter/MessageFiltersTest.java

@@ -201,6 +201,18 @@ void filterSpeedIsAtLeast5kPerSec() {
       assertThat(took).isLessThan(1000);
       assertThat(matched).isPositive();
     }
+
+    @Test
+    void nullFiltering() {
+      String msg = "{ \"field\": { \"inner\": null } }";
+
+      var f = celScriptFilter("record.value.field.inner == null");
+      assertTrue(f.test(msg().content(msg)));
+
+      f = celScriptFilter("record.value.field.inner != null");
+      assertFalse(f.test(msg().content(msg)));
+    }
+
   }
 
   @Test

api/src/test/java/io/kafbat/ui/service/KafkaConfigSanitizerTest.java

@@ -27,6 +27,7 @@ void obfuscateCredentials() {
     assertThat(sanitizer.sanitize("main.consumer.sasl.jaas.config", "secret")).isEqualTo("******");
     assertThat(sanitizer.sanitize("database.password", "secret")).isEqualTo("******");
     assertThat(sanitizer.sanitize("basic.auth.user.info", "secret")).isEqualTo("******");
+    assertThat(sanitizer.sanitize("private.key.passphrase", "secret")).isEqualTo("******");
 
     //AWS var sanitizing
     assertThat(sanitizer.sanitize("aws.access.key.id", "secret")).isEqualTo("******");