[CORE-16880] kafka: autocreate internal topics with correct properties

src/v/cluster/BUILD

@@ -1247,6 +1247,7 @@ redpanda_cc_library(
         ":errc",
         ":logger",
         "//src/v/base",
+        "//src/v/config",
         "@fmt",
     ],
     visibility = ["//visibility:public"],

src/v/cluster/controller.cc

@@ -1199,16 +1199,8 @@ ss::future<> controller::cluster_creation_hook(
  * how many replicas they should use.
  */
 int16_t controller::internal_topic_replication() const {
-    auto replication_factor
-      = (int16_t)config::shard_local_cfg().internal_topic_replication_factor();
-    if (replication_factor > (int16_t)_members_table.local().node_count()) {
-        // Fall back to r=1 if we do not have sufficient nodes
-        return 1;
-    } else {
-        // Respect `internal_topic_replication_factor` if enough
-        // nodes were available.
-        return replication_factor;
-    }
+    return cluster::internal_topic_replication(
+      _members_table.local().node_count());
 }
 
 ss::future<result<std::vector<partition_state>>>

src/v/cluster/members_table.cc

@@ -14,6 +14,7 @@
 #include "cluster/errc.h"
 #include "cluster/logger.h"
 #include "cluster/types.h"
+#include "config/configuration.h"
 #include "model/metadata.h"
 
 #include <fmt/format.h>
@@ -415,4 +416,16 @@ void members_table::notify_member_updated(
     }
 }
 
+int16_t internal_topic_replication(size_t node_count) {
+    auto replication_factor = static_cast<int16_t>(
+      config::shard_local_cfg().internal_topic_replication_factor());
+    if (replication_factor > static_cast<int16_t>(node_count)) {
+        // Fall back to r=1 if we do not have sufficient nodes
+        return 1;
+    }
+    // Respect `internal_topic_replication_factor` if enough
+    // nodes were available.
+    return replication_factor;
+}
+
 } // namespace cluster

src/v/cluster/members_table.h

@@ -115,4 +115,10 @@ class members_table {
 
     void notify_member_updated(model::node_id, model::membership_state);
 };
+
+/// Helper for subsystems that create internal topics, to discover how many
+/// replicas they should use: respects `internal_topic_replication_factor`
+/// if enough nodes are available, otherwise falls back to r=1.
+int16_t internal_topic_replication(size_t node_count);
+
 } // namespace cluster

src/v/kafka/server/BUILD

@@ -70,6 +70,7 @@ redpanda_cc_library(
         "//src/v/pandaproxy/schema_registry:types",
         "//src/v/security",
         "//src/v/serde:chrono",
+        "//src/v/ssx:sformat",
         "//src/v/strings:string_switch",
         "//src/v/utils:absl_sstring_hash",
         "//src/v/utils:tristate",
@@ -320,6 +321,7 @@ redpanda_cc_library(
         "//src/v/cluster:shard_table",
         "//src/v/cluster:state_machine_registry",
         "//src/v/cluster:tm_stm",
+        "//src/v/cluster:topic_configuration",
         "//src/v/cluster:topic_table",
         "//src/v/cluster:tx_protocol_types",
         "//src/v/cluster:types",

src/v/kafka/server/group_initializer.cc

@@ -12,6 +12,7 @@
 #include "kafka/server/group_initializer.h"
 
 #include "cluster/controller_api.h"
+#include "cluster/members_table.h"
 #include "cluster/topics_frontend.h"
 #include "kafka/protocol/logger.h"
 /*
@@ -60,37 +61,40 @@ ss::future<bool> group_initializer::assure_topic_exists(
     co_return true;
 }
 
+cluster::topic_configuration consumer_offsets_topic_configuration(
+  model::topic_namespace tp_ns, int16_t replication_factor) {
+    cluster::topic_configuration topic{
+      std::move(tp_ns.ns),
+      std::move(tp_ns.tp),
+      config::shard_local_cfg().group_topic_partitions(),
+      replication_factor};
+
+    topic.properties.cleanup_policy_bitflags
+      = model::cleanup_policy_bitflags::compaction;
+    // allow creation even if a consumer group migration is in progress
+    topic.is_migrated = true;
+    return topic;
+}
+
 /*
  * create the internal metadata topic for group membership
  */
 ss::future<bool> group_initializer::try_create_consumer_group_topic() {
-    // Attempt to use internal topic replication factor, if enough nodes found.
-    auto replication_factor
-      = (int16_t)config::shard_local_cfg().internal_topic_replication_factor();
     if (!_members_table.local_is_initialized()) {
         vlog(
           klog.warn,
           "can not create consumer offsets topic topic - members_table not "
           "initialized");
         return ssx::now(false);
     }
-    if (
-      static_cast<size_t>(replication_factor)
-      > _members_table.local().node_count()) {
-        replication_factor = 1;
-    }
+    // Attempt to use internal topic replication factor, if enough nodes found.
+    auto replication_factor = cluster::internal_topic_replication(
+      _members_table.local().node_count());
 
     // the new internal metadata topic for group membership
-    cluster::topic_configuration topic{
-      _coordinator_ntp_mapper.ns(),
-      _coordinator_ntp_mapper.topic(),
-      config::shard_local_cfg().group_topic_partitions(),
-      replication_factor};
-
-    topic.properties.cleanup_policy_bitflags
-      = model::cleanup_policy_bitflags::compaction;
-    // allow cretation even if a consumer group migration is in progress
-    topic.is_migrated = true;
+    auto topic = consumer_offsets_topic_configuration(
+      {_coordinator_ntp_mapper.ns(), _coordinator_ntp_mapper.topic()},
+      replication_factor);
 
     if (!_topics_frontend.local_is_initialized()) {
         vlog(

src/v/kafka/server/group_initializer.h

@@ -10,11 +10,19 @@
  */
 #pragma once
 
+#include "cluster/topic_configuration.h"
 #include "kafka/server/coordinator_ntp_mapper.h"
 /*
  * This class creates consumer offsets topic
  */
 namespace kafka {
+
+/// \brief The intended configuration of the consumer group metadata topic:
+/// compacted, with group_topic_partitions partitions. Creation is permitted
+/// while a consumer group migration is in progress.
+cluster::topic_configuration consumer_offsets_topic_configuration(
+  model::topic_namespace tp_ns, int16_t replication_factor);
+
 class group_initializer {
 public:
     group_initializer(

src/v/kafka/server/handlers/configs/config_response_utils.cc

@@ -136,18 +136,6 @@ bool config_property_requested(
                 != configuration_keys->end();
 }
 
-template<typename T>
-ss::sstring describe_as_string(const T& t) {
-    if constexpr (::detail::is_specialization_of_v<T, std::chrono::duration>) {
-        return ssx::sformat("{}", t.count());
-    } else {
-        return ssx::sformat("{}", t);
-    }
-}
-
-// Instantiate explicitly for unit testing
-template ss::sstring describe_as_string(const int&);
-
 // Kafka protocol defines integral types by sizes. See
 // https://kafka.apache.org/protocol.html
 // Therefore we should also use type sizes for integrals and use Java type sizes
@@ -465,15 +453,6 @@ template void add_topic_config_if_requested<int, describe_int_t>(
   describe_int_t&& describe_f,
   bool hide_default_override = false);
 
-template<typename T>
-static ss::sstring maybe_print_tristate(const tristate<T>& tri) {
-    if (tri.is_disabled() || !tri.has_optional_value()) {
-        return "-1";
-    }
-
-    return describe_as_string(tri.value());
-}
-
 template<typename T>
 static void add_topic_config(
   config_response_container_t& result,

src/v/kafka/server/handlers/configs/config_response_utils.h

@@ -11,14 +11,18 @@
 
 #pragma once
 
+#include "base/type_traits.h"
 #include "cluster/types.h"
 #include "container/chunked_vector.h"
 #include "kafka/protocol/describe_configs.h"
 #include "kafka/protocol/schemata/create_topics_response.h"
 #include "model/compression.h"
 #include "model/fundamental.h"
 #include "pandaproxy/schema_registry/subject_name_strategy.h"
+#include "ssx/sformat.h"
+#include "utils/tristate.h"
 
+#include <chrono>
 #include <iterator>
 #include <optional>
 
@@ -42,6 +46,28 @@ struct config_response {
 using config_response_container_t = chunked_vector<config_response>;
 using config_key_t = std::optional<chunked_vector<ss::sstring>>;
 
+/// Serialize a topic property value the way it appears in Kafka
+/// configuration responses (durations as their millisecond count).
+template<typename T>
+ss::sstring describe_as_string(const T& t) {
+    if constexpr (::detail::is_specialization_of_v<T, std::chrono::duration>) {
+        return ssx::sformat("{}", t.count());
+    } else {
+        return ssx::sformat("{}", t);
+    }
+}
+
+/// Serialize a tristate topic property value the way it appears in Kafka
+/// configuration responses: -1 when disabled or not set.
+template<typename T>
+ss::sstring maybe_print_tristate(const tristate<T>& tri) {
+    if (tri.is_disabled() || !tri.has_optional_value()) {
+        return "-1";
+    }
+
+    return describe_as_string(tri.value());
+}
+
 /// This abstract interface was added in order to decouple metadata_cache from
 /// the make_topic_configs method below.  This allows us to create our own
 /// adapters for unit tests

src/v/kafka/server/handlers/metadata.cc

@@ -9,6 +9,7 @@
 
 #include "kafka/server/handlers/metadata.h"
 
+#include "cluster/members_table.h"
 #include "cluster/metadata_cache.h"
 #include "cluster/topics_frontend.h"
 #include "cluster/types.h"
@@ -20,17 +21,20 @@
 #include "kafka/protocol/types.h"
 #include "kafka/server/errors.h"
 #include "kafka/server/fwd.h"
+#include "kafka/server/group_initializer.h"
 #include "kafka/server/handlers/describe_cluster.h"
 #include "kafka/server/handlers/details/leader_epoch.h"
 #include "kafka/server/handlers/details/security.h"
 #include "kafka/server/handlers/topics/topic_utils.h"
+#include "kafka/server/handlers/topics/types.h"
 #include "kafka/server/response.h"
 #include "model/errc.h"
 #include "model/metadata.h"
 #include "model/namespace.h"
 #include "model/timeout_clock.h"
 #include "random/generators.h"
 #include "security/acl.h"
+#include "security/audit/audit_log_topic.h"
 
 #include <seastar/core/coroutine.hh>
 #include <seastar/core/future-util.hh>
@@ -206,6 +210,49 @@ metadata_response::topic make_topic_response_from_topic_metadata(
 }
 
 namespace {
+/// Internal topics requested by name are created with their owning
+/// subsystem's configuration rather than cluster defaults, mirroring Apache
+/// Kafka's special-casing of internal topics during metadata-driven topic
+/// auto-creation.
+cluster::topic_configuration
+autocreate_topic_configuration(request_context& ctx, model::topic topic) {
+    if (topic == model::kafka_consumer_offsets_topic) {
+        return consumer_offsets_topic_configuration(
+          model::kafka_consumer_offsets_nt,
+          cluster::internal_topic_replication(
+            ctx.metadata_cache().node_count()));
+    }
+    if (topic == model::schema_registry_internal_tp.topic) {
+        return schema_registry_topic_configuration(
+          cluster::internal_topic_replication(
+            ctx.metadata_cache().node_count()));
+    }
+    if (topic == model::kafka_audit_logging_topic) {
+        auto replication_factor
+          = config::shard_local_cfg().audit_log_replication_factor().value_or(
+            cluster::internal_topic_replication(
+              ctx.metadata_cache().node_count()));
+        cluster::topic_configuration cfg{
+          model::kafka_namespace,
+          std::move(topic),
+          config::shard_local_cfg().audit_log_num_partitions(),
+          replication_factor};
+        cfg.properties = security::audit::audit_log_topic_properties();
+        return cfg;
+    }
+    // default topic configuration
+    cluster::topic_configuration cfg{
+      model::kafka_namespace,
+      std::move(topic),
+      config::shard_local_cfg().default_topic_partitions(),
+      config::shard_local_cfg().default_topic_replication()};
+    // Need to respect the default_redpanda_storage_mode when autocreating a
+    // topic.
+    cfg.properties.storage_mode
+      = config::shard_local_cfg().default_redpanda_storage_mode();
+    return cfg;
+}
+
 ss::future<metadata_response::topic> create_topic(
   request_context& ctx,
   model::topic topic,
@@ -221,20 +268,10 @@ ss::future<metadata_response::topic> create_topic(
         t.error_code = error_code::broker_not_available;
         co_return t;
     }
-    // default topic configuration
-    cluster::topic_configuration cfg{
-      model::kafka_namespace,
-      topic,
-      config::shard_local_cfg().default_topic_partitions(),
-      config::shard_local_cfg().default_topic_replication()};
-    // Need to respect the default_redpanda_storage_mode when autocreating a
-    // topic.
-    cfg.properties.storage_mode
-      = config::shard_local_cfg().default_redpanda_storage_mode();
     auto tout = config::shard_local_cfg().internal_rpc_request_timeout_ms();
     try {
         auto res = co_await ctx.topics_frontend().autocreate_topics(
-          {std::move(cfg)}, tout);
+          {autocreate_topic_configuration(ctx, topic)}, tout);
         vassert(res.size() == 1, "expected single result");
         // error, neither success nor topic exists
         if (!(res[0].ec == cluster::errc::success
@@ -251,10 +288,11 @@ ss::future<metadata_response::topic> create_topic(
           ctx.controller_api(),
           tout + model::timeout_clock::now());
 
-        auto tp_md = ctx.metadata_cache().get_topic_metadata(res[0].tp_ns);
+        auto tp_md = ctx.metadata_cache().get_topic_metadata(
+          model::topic_namespace_view(model::kafka_namespace, topic));
         if (!tp_md) {
             metadata_response::topic t;
-            t.name = std::move(res[0].tp_ns.tp);
+            t.name = std::move(topic);
             t.error_code = error_code::invalid_topic_exception;
             co_return t;
         }

src/v/kafka/server/handlers/topics/types.cc

@@ -385,6 +385,33 @@ cluster::topic_configuration to_topic_config(
     return cfg;
 }
 
+cluster::topic_configuration
+schema_registry_topic_configuration(int16_t replication_factor) {
+    // Create the base topic configuration to get the cluster defaults
+    auto cfg = to_topic_config(
+      model::kafka_namespace,
+      model::schema_registry_internal_tp.topic,
+      /*partition_count=*/1,
+      replication_factor,
+      {});
+    // Now update the properties
+    cfg.properties.cleanup_policy_bitflags
+      = model::cleanup_policy_bitflags::compaction;
+    cfg.properties.compression = model::compression::none;
+    cfg.properties.retention_bytes = tristate<size_t>{disable_tristate};
+    cfg.properties.retention_duration = tristate<std::chrono::milliseconds>{
+      disable_tristate};
+    cfg.properties.retention_local_target_bytes = tristate<size_t>{
+      disable_tristate};
+    cfg.properties.retention_local_target_ms
+      = tristate<std::chrono::milliseconds>{disable_tristate};
+    cfg.properties.initial_retention_local_target_bytes = tristate<size_t>{
+      disable_tristate};
+    cfg.properties.initial_retention_local_target_ms
+      = tristate<std::chrono::milliseconds>{disable_tristate};
+    return cfg;
+}
+
 static std::vector<kafka::creatable_topic_configs>
 convert_topic_configs(config_response_container_t&& topic_cfgs) {
     auto configs = std::vector<kafka::creatable_topic_configs>();