Skip to content

Keep Kafka version fallback on metadata.version #1905

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
reachsrinivaschennupati-tech wants to merge 1 commit into
base: main
Choose a base branch
from
+59 −4
Open

Keep Kafka version fallback on metadata.version #1905

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 13 additions & 4 deletions api/src/main/java/io/kafbat/ui/service/ReactiveAdminClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,7 @@
@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";

public enum SupportedFeature {
INCREMENTAL_ALTER_CONFIGS(2.3f),
Expand Down Expand Up @@ -164,12 +165,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());
}
Expand All @@ -193,6 +191,17 @@ private static Mono<ConfigRelatedInfo> extract(AdminClient ac) {
})
.cache(UPDATE_DURATION);
}

}

@VisibleForTesting
static Optional<String> extractKafkaVersion(List<ConfigEntry> configs) {
for (ConfigEntry entry : configs) {
if (entry.name().equals(INTER_BROKER_PROTOCOL_VERSION_CONFIG) && entry.value() != null) {
return Optional.of(entry.value());
}
}
return Optional.empty();
Comment on lines +197 to +204

@coderabbitai coderabbitai Bot Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== repo files matching ReactiveAdminClient =="
fd -a 'ReactiveAdminClient|ReactiveAdminClientVersionTest' . || true

echo "== relevant snippets =="
for f in $(fd 'ReactiveAdminClient|ReactiveAdminClientVersionTest' .); do
  echo "--- $f"
  wc -l "$f"
  rg -n "extractKafkaVersion|INTER_BROKER_PROTOCOL_VERSION_CONFIG|LOG_MESSAGE_FORMAT_VERSION|inter.broker.protocol.version|log.message.format.version|kafkaVersion|kafka version" "$f" -C 8 || true
done

echo "== git diff summary/name =="
git diff --stat || true
git diff -- api/src/main/java/io/kafbat/ui/service/ReactiveAdminClient.java api/src/test/java/io/kafbat/ui/service/ReactiveAdminClientVersionTest.java || true

Repository: kafbat/kafka-ui

Length of output: 6387

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== all tracked occurrences of Kafka version config fallback =="
rg -n "log\.message\.format\.version|inter\.broker\.protocol\.version|KafkaVersion|extractKafkaVersion|KafkaVersion::parse|inter.broker.protocol.version|log.message.format.version" . || true

echo "== Behavioral probe: deterministic fallback behavior in source text =="
python3 - <<'PY'
import re
from pathlib import Path

client = Path("api/src/main/java/io/kafbat/ui/service/ReactiveAdminClient.java").read_text()
test = Path("api/src/test/java/io/kafbat/ui/service/ReactiveAdminClientVersionTest.java").read_text()

m = re.search(r"static Optional<String> extractKafkaVersion\(\).*?\n  \}", client, re.S)
method = m.group(0) if m else ""
has_inter_only_return = bool(re.search(r"entry\.name\(\)\.equals\(INTER_BROKER_PROTOCOL_VERSION_CONFIG\)\s*&&\s*entry\.value\(\)\s*!=\s*null", method)) and re.search(r"return Optional\.of\(entry\.value\(\)\);", method)

has_log_fallback = "log.message.format.version" in method or "LOG_MESSAGE_FORMAT_VERSION" in method

test_ignores_log_single = 'new ConfigEntry("log.message.format.version", "3.8-IV0")' in test and 'assertThat(ReactiveAdminClient.extractKafkaVersion(configs)).isEmpty()' in test
test_returns_empty_when_inter_null = 'new ConfigEntry("inter.broker.protocol.version", null)' in test and 'new ConfigEntry("log.message.format.version", "3.8-IV0")' in test and test.endswith('assertThat(ReactiveAdminClient.extractKafkaVersion(configs)).isEmpty();\n}\n')

print({"method_present": bool(m), "inter_broker_only_return": has_inter_only_return, "log_format_or_constant_in_method": has_log_fallback, "test_ignores_log_single": test_ignores_log_single, "test_returns_empty_when_inter_null": test_returns_empty_when_inter_null})
PY

Repository: kafbat/kafka-ui

Length of output: 3336

Restore the stated Kafka version fallback.

extractKafkaVersion(...) only returns inter.broker.protocol.version, so log.message.format.version is ignored even when the preferred config is missing or null. Update the helper and tests to select the log-message format version only after no valid inter-broker value exists.

📍 Affects 2 files
  • api/src/main/java/io/kafbat/ui/service/ReactiveAdminClient.java#L197-L204 (this comment)
  • api/src/test/java/io/kafbat/ui/service/ReactiveAdminClientVersionTest.java#L21-L38
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@api/src/main/java/io/kafbat/ui/service/ReactiveAdminClient.java` around lines
197 - 204, The extractKafkaVersion helper must fall back to
log.message.format.version when no non-null inter.broker.protocol.version value
exists; update it to preserve the preferred-config priority, then update
ReactiveAdminClientVersionTest.java lines 21-38 to cover the missing and null
preferred values and verify the fallback selection.

}

public static Mono<ReactiveAdminClient> create(AdminClient adminClient, ClustersProperties.AdminClient properties) {
Expand Down
46 changes: 46 additions & 0 deletions api/src/test/java/io/kafbat/ui/service/ReactiveAdminClientVersionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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 extractKafkaVersionUsesInterBrokerProtocolVersion() {
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 extractKafkaVersionIgnoresLogMessageFormatVersion() {
var configs = List.of(
new ConfigEntry("log.message.format.version", "3.8-IV0")
);

assertThat(ReactiveAdminClient.extractKafkaVersion(configs)).isEmpty();
}

@Test
void extractKafkaVersionReturnsEmptyWhenInterBrokerProtocolVersionIsNull() {
var configs = List.of(
new ConfigEntry("inter.broker.protocol.version", null),
new ConfigEntry("log.message.format.version", "3.8-IV0")
);

assertThat(ReactiveAdminClient.extractKafkaVersion(configs)).isEmpty();
}

@Test
void extractKafkaVersionReturnsEmptyWhenVersionConfigsAreMissing() {
var configs = List.of(new ConfigEntry("delete.topic.enable", "true"));

assertThat(ReactiveAdminClient.extractKafkaVersion(configs)).isEmpty();
}
}
Loading