Skip to content

Commit

Permalink
kafka: ignore case in parse_and_set_bool()
Browse files Browse the repository at this point in the history
Kafka ignores case when setting a boolean property:
https://github.com/apache/kafka/blob/master/clients/src/main/...
...java/org/apache/kafka/common/config/ConfigDef.java#L690-L710

Realign ourselves with Kafka by transforming the string value to
lowercase before parsing it in `parse_and_set_bool()`.
  • Loading branch information
WillemKauf committed Oct 1, 2024
1 parent e2a47cb commit 0faad94
Show file tree
Hide file tree
Showing 2 changed files with 41 additions and 2 deletions.
13 changes: 11 additions & 2 deletions src/v/kafka/server/handlers/configs/config_utils.h
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@

#include <absl/container/node_hash_set.h>

#include <algorithm>
#include <optional>

namespace kafka {
Expand Down Expand Up @@ -627,10 +628,18 @@ inline void parse_and_set_bool(

if (op == config_resource_operation::set && value) {
try {
property.value = string_switch<bool>(*value)
// Ignore case.
auto str_value = std::move(*value);
std::transform(
str_value.begin(),
str_value.end(),
str_value.begin(),
[](const auto& c) { return std::tolower(c); });

property.value = string_switch<bool>(str_value)
.match("true", true)
.match("false", false);
auto v_error = validator(tn, *value, property.value);
auto v_error = validator(tn, str_value, property.value);
if (v_error) {
throw validation_error{*v_error};
}
Expand Down
30 changes: 30 additions & 0 deletions src/v/kafka/server/tests/alter_config_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -1053,3 +1053,33 @@ FIXTURE_TEST(
describe_resp);
}
}

FIXTURE_TEST(
test_shadow_indexing_uppercase_alter_config, alter_config_test_fixture) {
wait_for_controller_leadership().get();
model::topic test_tp{"topic-1"};
create_topic(test_tp, 6);
using map_t = absl::flat_hash_map<
ss::sstring,
std::pair<std::optional<ss::sstring>, kafka::config_resource_operation>>;
map_t properties;
// Case is on purpose.
properties.emplace(
"redpanda.remote.write",
std::make_pair("True", kafka::config_resource_operation::set));
properties.emplace(
"redpanda.remote.read",
std::make_pair("TRUE", kafka::config_resource_operation::set));

auto resp = incremental_alter_configs(
make_incremental_alter_topic_config_resource_cv(test_tp, properties));

BOOST_REQUIRE_EQUAL(resp.data.responses.size(), 1);
BOOST_REQUIRE_EQUAL(
resp.data.responses[0].error_code, kafka::error_code::none);
auto describe_resp = describe_configs(test_tp);
assert_property_value(
test_tp, "redpanda.remote.write", "true", describe_resp);
assert_property_value(
test_tp, "redpanda.remote.read", "true", describe_resp);
}

0 comments on commit 0faad94

Please sign in to comment.