kafka: auto-create internal topics with their intended configs

The auto create topics path would always create topics with default
cluster configurations. This is very incorrect for certain special
topics such as `_schemas`, `__consumer_offsets`, or our internal
`_redpanda.audit_log` topic.

Use the newly available config accessors for these specific topics, and
use them when autocreating these topics to ensure we are constructing
these topics with the correct properties.

Author: WillemKauf

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/tests/metadata_test.cc

@@ -20,6 +20,7 @@
 #include "kafka/server/handlers/details/security.h"
 #include "kafka/server/handlers/metadata.h"
 #include "model/fundamental.h"
+#include "model/namespace.h"
 #include "model/timeout_clock.h"
 #include "redpanda/tests/fixture.h"
 #include "security/acl.h"
@@ -550,6 +551,73 @@ FIXTURE_TEST(metadata_autocreate, metadata_fixture) {
     }
 }
 
+FIXTURE_TEST(metadata_autocreate_internal_topics, metadata_fixture) {
+    // Internal topics auto-created in response to a metadata request must
+    // get their owning subsystem's configuration, not cluster defaults.
+    auto undo = set_auto_create_topics(true);
+    wait_for_controller_leadership().get();
+
+    auto client = make_kafka_client().get();
+    client.connect().get();
+    auto close = ss::defer([&client] { client.stop().get(); });
+
+    auto req = kafka::metadata_request{.data{
+      .topics = {{
+        {.name{model::kafka_consumer_offsets_topic}},
+        {.name{model::schema_registry_internal_tp.topic}},
+        {.name{model::kafka_audit_logging_topic}},
+      }},
+      .allow_auto_topic_creation = true,
+      .include_cluster_authorized_operations = false,
+      .include_topic_authorized_operations = false}};
+    auto resp = client.dispatch(std::move(req), kafka::api_version{9}).get();
+
+    // Only check topic-level errors: partitions may transiently report
+    // leader_not_available until leadership metadata catches up with the
+    // freshly created topics.
+    BOOST_REQUIRE_EQUAL(resp.data.topics.size(), 3);
+    for (const auto& topic : resp.data.topics) {
+        BOOST_REQUIRE_EQUAL(topic.error_code, kafka::error_code::none);
+    }
+
+    const auto& topics_state = app.controller->get_topics_state().local();
+
+    auto co_cfg = topics_state.get_topic_cfg(model::kafka_consumer_offsets_nt);
+    BOOST_REQUIRE(co_cfg.has_value());
+    BOOST_CHECK_EQUAL(
+      co_cfg->partition_count,
+      config::shard_local_cfg().group_topic_partitions());
+    BOOST_CHECK(
+      co_cfg->properties.cleanup_policy_bitflags
+      == model::cleanup_policy_bitflags::compaction);
+
+    auto schemas_cfg = topics_state.get_topic_cfg(
+      {model::kafka_namespace, model::schema_registry_internal_tp.topic});
+    BOOST_REQUIRE(schemas_cfg.has_value());
+    BOOST_CHECK_EQUAL(schemas_cfg->partition_count, 1);
+    BOOST_CHECK(
+      schemas_cfg->properties.cleanup_policy_bitflags
+      == model::cleanup_policy_bitflags::compaction);
+    BOOST_CHECK(
+      schemas_cfg->properties.compression == model::compression::none);
+    BOOST_CHECK(schemas_cfg->properties.retention_duration.is_disabled());
+    BOOST_CHECK(schemas_cfg->properties.retention_bytes.is_disabled());
+
+    auto audit_cfg = topics_state.get_topic_cfg(model::kafka_audit_logging_nt);
+    BOOST_REQUIRE(audit_cfg.has_value());
+    BOOST_CHECK_EQUAL(
+      audit_cfg->partition_count,
+      config::shard_local_cfg().audit_log_num_partitions());
+    BOOST_CHECK(
+      audit_cfg->properties.cleanup_policy_bitflags
+      == model::cleanup_policy_bitflags::deletion);
+    BOOST_REQUIRE(
+      audit_cfg->properties.retention_duration.has_optional_value());
+    BOOST_CHECK(
+      audit_cfg->properties.retention_duration.value()
+      == std::chrono::milliseconds(604800000));
+}
+
 FIXTURE_TEST(metadata_v12_unauthorized, metadata_fixture) {
     using kafka::api_version;
     constexpr auto min_version = api_version{12};