Fix Kafka version fallback for MSK clusters

Signed-off-by: srinivas chennupati <[email protected]>

Author: reachsrinivaschennupati-tech

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

@@ -102,6 +102,8 @@
 @AllArgsConstructor
 public class ReactiveAdminClient implements Closeable {
   private static final String DEFAULT_UNKNOWN_VERSION = "Unknown";
+  private static final String INTER_BROKER_PROTOCOL_VERSION_CONFIG = "inter.broker.protocol.version";
+  private static final String LOG_MESSAGE_FORMAT_VERSION_CONFIG = "log.message.format.version";
 
   public enum SupportedFeature {
     INCREMENTAL_ALTER_CONFIGS(2.3f),
@@ -164,12 +166,9 @@ private static Mono<ConfigRelatedInfo> extract(AdminClient ac) {
                 .flatMap(tuple -> {
                   List<ConfigEntry> configs = tuple.getT1();
                   FeatureMetadata featureMetadata = tuple.getT2();
-                  Optional<String> version = Optional.empty();
+                  Optional<String> version = extractKafkaVersion(configs);
                   boolean topicDeletionEnabled = true;
                   for (ConfigEntry entry : configs) {
-                    if (entry.name().contains("inter.broker.protocol.version")) {
-                      version = Optional.ofNullable(entry.value());
-                    }
                     if (entry.name().equals("delete.topic.enable") && entry.value() != null) {
                       topicDeletionEnabled = Boolean.parseBoolean(entry.value());
                     }
@@ -193,6 +192,21 @@ private static Mono<ConfigRelatedInfo> extract(AdminClient ac) {
           })
           .cache(UPDATE_DURATION);
     }
+
+  }
+
+  @VisibleForTesting
+  static Optional<String> extractKafkaVersion(List<ConfigEntry> configs) {
+    Optional<String> logMessageFormatVersion = Optional.empty();
+    for (ConfigEntry entry : configs) {
+      if (entry.name().equals(INTER_BROKER_PROTOCOL_VERSION_CONFIG) && entry.value() != null) {
+        return Optional.of(entry.value());
+      }
+      if (entry.name().equals(LOG_MESSAGE_FORMAT_VERSION_CONFIG) && entry.value() != null) {
+        logMessageFormatVersion = Optional.of(entry.value());
+      }
+    }
+    return logMessageFormatVersion;
   }
 
   public static Mono<ReactiveAdminClient> create(AdminClient adminClient, ClustersProperties.AdminClient properties) {

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

@@ -0,0 +1,46 @@
+package io.kafbat.ui.service;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.junit.jupiter.api.Test;
+
+class ReactiveAdminClientVersionTest {
+
+  @Test
+  void extractKafkaVersionUsesInterBrokerProtocolVersionFirst() {
+    var configs = List.of(
+        new ConfigEntry("log.message.format.version", "3.8-IV0"),
+        new ConfigEntry("inter.broker.protocol.version", "3.9-IV0")
+    );
+
+    assertThat(ReactiveAdminClient.extractKafkaVersion(configs)).contains("3.9-IV0");
+  }
+
+  @Test
+  void extractKafkaVersionFallsBackToLogMessageFormatVersion() {
+    var configs = List.of(
+        new ConfigEntry("log.message.format.version", "3.8-IV0")
+    );
+
+    assertThat(ReactiveAdminClient.extractKafkaVersion(configs)).contains("3.8-IV0");
+  }
+
+  @Test
+  void extractKafkaVersionFallsBackWhenInterBrokerProtocolVersionIsNull() {
+    var configs = List.of(
+        new ConfigEntry("inter.broker.protocol.version", null),
+        new ConfigEntry("log.message.format.version", "3.8-IV0")
+    );
+
+    assertThat(ReactiveAdminClient.extractKafkaVersion(configs)).contains("3.8-IV0");
+  }
+
+  @Test
+  void extractKafkaVersionReturnsEmptyWhenVersionConfigsAreMissing() {
+    var configs = List.of(new ConfigEntry("delete.topic.enable", "true"));
+
+    assertThat(ReactiveAdminClient.extractKafkaVersion(configs)).isEmpty();
+  }
+}