[v25.3.x] Fix audit log client on upgrade

src/v/kafka/client/cluster.cc

@@ -176,8 +176,9 @@ ss::future<> cluster::dispatch_and_apply_metadata_updates(
 
         co_await apply_metadata(std::move(mu));
     } catch (const broker_error& e) {
+        // TODO(CORE-14956) - We need to improve handling of broker errors
         vlog(
-          _logger.warn, "Failed to dispatch metadata request - {}", e.what());
+          _logger.debug, "Failed to dispatch metadata request - {}", e.what());
     }
 }
 
@@ -337,8 +338,10 @@ ss::future<metadata_response> cluster::dispatch_metadata_request(
       metadata_request{.data{
         .topics = std::move(topics_to_request),
         .allow_auto_topic_creation = false,
-        .include_cluster_authorized_operations = true,
-        .include_topic_authorized_operations = true}},
+        .include_cluster_authorized_operations = bool(
+          _config.include_authorized_operations),
+        .include_topic_authorized_operations = bool(
+          _config.include_authorized_operations)}},
       metadata_version);
     vassert(
       std::holds_alternative<kafka::metadata_response>(reply),
@@ -356,8 +359,9 @@ cluster::dispatch_describe_cluster_request(shared_broker_t broker) {
     auto request_version = co_await get_describe_cluster_request_version(
       broker, _as);
     auto reply = co_await broker->dispatch(
-      describe_cluster_request{
-        .data{.include_cluster_authorized_operations = true}},
+      describe_cluster_request{.data{
+        .include_cluster_authorized_operations = bool(
+          _config.include_authorized_operations)}},
       request_version);
     vassert(
       std::holds_alternative<kafka::describe_cluster_response>(reply),

src/v/kafka/client/configuration.cc

@@ -280,6 +280,8 @@ connection_configuration::from_config_store(const configuration& cfg) {
         };
     }
     ret.broker_tls = tls_configuration::from_tls_config(cfg.broker_tls.value());
+    ret.include_authorized_operations
+      = connection_configuration::include_authorized_ops_t::no;
 
     return ret;
 }

src/v/kafka/client/configuration.h

@@ -97,12 +97,18 @@ struct sasl_configuration {
  * are present,they will be used to connect to the brokers.
  */
 struct connection_configuration {
+    using include_authorized_ops_t
+      = ss::bool_class<struct include_authorized_ops_tag>;
     std::vector<net::unresolved_address> initial_brokers;
     std::optional<tls_configuration> broker_tls;
     std::optional<sasl_configuration> sasl_cfg;
     std::optional<ss::sstring> client_id;
     std::chrono::milliseconds max_metadata_age{10000};
     std::chrono::milliseconds connection_timeout{1000};
+    // If set to true, will request from the brokers the bitmask of authorized
+    // operations the authenticated user is permitted to perform
+    include_authorized_ops_t include_authorized_operations{
+      include_authorized_ops_t::yes};
 
     ss::sstring get_client_id() const {
         return client_id.value_or("kafka-client");

src/v/kafka/server/handlers/details/security.h

@@ -392,7 +392,7 @@ authorized_operations(authorized_function<T> fn, const T& resource) {
       ops.end(),
       std::back_inserter(allowed_operations),
       [&fn, &resource](security::acl_operation op) {
-          return fn(op, resource, authz_quiet::no, audit_authz_check::no);
+          return fn(op, resource, authz_quiet::yes, audit_authz_check::no);
       });
 
     return allowed_operations;