Keep Kafka version fallback on metadata.version - #1905
Keep Kafka version fallback on metadata.version#1905reachsrinivaschennupati-tech wants to merge 1 commit into
Conversation
kapybro
Bot
added
status/triage/manual
|
AI Summary The issue addresses a problem where Kafka version fallback logic for Amazon MSK clusters doesn't properly use |
kapybro
Bot
changed the title
kapybro
Bot
added
area/brokers
📝 WalkthroughWalkthroughKafka version extraction now uses a named ChangesKafka version extraction
Estimated code review effort: 2 (Simple) | ~10 minutes Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hi reachsrinivaschennupati-tech! 👋
Welcome, and thank you for opening your first PR in the repo!
Please wait for triaging by our maintainers.
Please take a look at our contributing guide.
|
Hi maintainers, this PR is ready for review when you have a chance. The change is scoped to Kafka version detection for MSK-style broker configurations: Validation noted in the PR:
Happy to update the validation notes if there is a preferred test command for this module or if maintainers want the fallback precedence adjusted. |
|
please explain why you decided to use that property as a fallback? and what's the initial cause why this config entry isn't available/returned by MSK? |
|
@Haarolean Thankyou for reviewing this PR. The reason I used "log.message.format.version" as a fallback is that "inter.broker.protocol.version" is still treated as the primary and preferred source. The fallback only applies when "inter.broker.protocol.version" is not present in the broker config response. In managed Kafka environments such as MSK, some broker-level settings can be controlled by the service and may not always be exposed through the AdminClient config response the same way they are in a self-managed Kafka cluster. That appears to be the case this change is trying to handle: the primary config is unavailable, but we still need a reasonable version-related signal instead of immediately falling back to "metadata.version" or "Unknown". "log.message.format.version" is not intended to be a perfect replacement for "inter.broker.protocol.version". I chose it because it is still a Kafka broker compatibility/version-related setting and is commonly considered together with "inter.broker.protocol.version" during Kafka upgrade compatibility handling. So the precedence remains conservative:
This means existing behavior is preserved when "inter.broker.protocol.version" is available, and the fallback only improves the MSK case where that config entry is missing. Please let me know, If you prefer not to infer the displayed Kafka version from "log.message.format.version", I can adjust the PR to keep "metadata.version" as the next fallback instead. Thanks, |
|
This doesn't answer either of my questions. I asked why MSK doesn't return this config entry — the answer I got is "some broker-level settings may not always be exposed... that appears to be the case this change is trying to handle." You wrote this change; you shouldn't have to guess what case it handles. Concretely, why this fallback doesn't hold up:
Also, your offer to "adjust the PR to keep Happy to reconsider if you can reproduce the issue against a real MSK cluster, post the actual |
|
Further user feedback is requested. Please reply within 7 days or we might close the issue. |
Signed-off-by: reachsrinivaschennupati-tech <[email protected]>
55b52e0 to
1b9027d
Compare
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
reachsrinivaschennupati-tech
changed the title
|
@Haarolean I reworked this based on your feedback. The PR no longer uses What changed now:
I attempted the focused Gradle test locally, but this machine only has JDK 17 and the current project build requires source release 25, so Gradle stops at |
|
Thanks for the additional feedback! We'll get back to your issue soon. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with 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.
Inline comments:
In `@api/src/main/java/io/kafbat/ui/service/ReactiveAdminClient.java`:
- Around line 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.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 34cb1c02-8004-48b9-ad0c-8418c0afe893
📒 Files selected for processing (2)
api/src/main/java/io/kafbat/ui/service/ReactiveAdminClient.javaapi/src/test/java/io/kafbat/ui/service/ReactiveAdminClientVersionTest.java
| @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(); |
There was a problem hiding this comment.
🎯 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 || trueRepository: 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})
PYRepository: 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.
2 participants
What changed
This reworks the previous approach based on maintainer feedback. The PR no longer uses
log.message.format.versionas a Kafka version fallback.Instead, broker config version extraction is limited to the exact
inter.broker.protocol.versionconfig key. If that value is unavailable, the existingmetadata.versionfallback fromdescribeFeaturesremains the next source, and the existingUnknowndisplay fallback remains unchanged.Why
log.message.format.versionis deprecated for newer Kafka versions and is not a reliable source for the displayed Kafka broker version. Without real MSKDescribeConfigs/describeFeaturesoutput proving otherwise, keepingmetadata.versionas the fallback is the safer behavior.Tests
Added focused unit coverage for Kafka version extraction:
inter.broker.protocol.versionwhen presentlog.message.format.versionmetadata.versionfallback path to runLocal focused Gradle test was attempted with:
NPM_CONFIG_CACHE=/private/tmp/kafka-ui-npm-cache ./gradlew :api:test --tests io.kafbat.ui.service.ReactiveAdminClientVersionTestIt could not complete locally because this machine only has JDK 17 installed while the project currently compiles with source release 25.
Summary by CodeRabbit
Bug Fixes
Tests