kafka: fix config source for cleanup policy

Currently Redpanda returns source=1 (topic-specific overwrite) for the
cleanup.policy config. This is different to Apache Kafka, which returns
source=5 (default config), even though the config value they return
(delete) is the same, so I believe this behaviour is a bug and this
commit fixes it by using the hide_default_override helper field.

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

@@ -520,7 +520,8 @@ config_response_container_t make_topic_configs(
       maybe_make_documentation(
         include_documentation,
         config::shard_local_cfg().log_cleanup_policy.desc()),
-      &describe_as_string<model::cleanup_policy_bitflags>);
+      &describe_as_string<model::cleanup_policy_bitflags>,
+      true);
 
     const std::string_view docstring{
       topic_properties.is_compacted()

tests/rptest/tests/controller_snapshot_test.py

@@ -9,6 +9,7 @@
 
 from rptest.tests.redpanda_test import RedpandaTest
 from rptest.services.redpanda import RESTART_LOG_ALLOW_LIST, RedpandaInstaller
+from rptest.services.redpanda_installer import ver_triple, RedpandaVersionTriple
 from rptest.services.cluster import cluster
 from rptest.services.admin import Admin
 from rptest.services.admin_ops_fuzzer import AdminOperationsFuzzer
@@ -131,10 +132,18 @@ def __init__(self, redpanda, node):
             'internal_rpc_status', 'internal_rpc_port'
         ]
         self.brokers = dict()
-        for broker in admin.get_brokers(node=node):
+        brokers_resp = admin.get_brokers(node=node)
+        for broker in brokers_resp:
             self.brokers[broker['node_id']] = dict(
                 (k, broker.get(k)) for k in broker_fields)
 
+        version_line = brokers_resp[0]['version']
+        assert all(
+            broker.get('version', version_line) == version_line
+            for broker in brokers_resp)
+
+        self.version = ver_triple(version_line)
+
     def _check_features(self, other, allow_upgrade_changes=False):
         for k, v in self.features_response.items():
             if k == 'features':
@@ -175,6 +184,16 @@ def to_set(configs):
         for t in self.topics:
             symdiff = to_set(self.topic_configs[t]) ^ to_set(
                 other.topic_configs[t])
+
+            # Version 24.1.0 includes a bugfix to change the source type of cleanup.policy from
+            # DYNAMIC_TOPIC_CONFIG to DEFAULT_CONFIG if the cleanup.policy wasn't specified as
+            # a config value when the topic is created.
+            if self.version < RedpandaVersionTriple((24, 1, 0)) and \
+                other.version >= RedpandaVersionTriple((24, 1, 0)):
+
+                symdiff -= set({('cleanup.policy', ('delete',
+                                                    'DYNAMIC_TOPIC_CONFIG'))})
+
             assert len(symdiff) == 0, f"configs differ, symdiff: {symdiff}"
 
             partitions = self.topic_partitions[t]

tests/rptest/tests/describe_topics_test.py

@@ -76,8 +76,7 @@ def test_describe_topics_with_documentation_and_types(self):
             "cleanup.policy":
             ConfigProperty(config_type="STRING",
                            value="DELETE",
-                           doc_string="Default topic cleanup policy",
-                           source_type="DYNAMIC_TOPIC_CONFIG"),
+                           doc_string="Default topic cleanup policy"),
             "compression.type":
             ConfigProperty(config_type="STRING",
                            value="producer",

tests/rptest/tests/topic_creation_test.py

@@ -652,7 +652,7 @@ def topic_alter_config_test(self):
         # overriden.
         topic_config = rpk.describe_topic_configs(topic)
         value, src = topic_config["cleanup.policy"]
-        assert value == "delete" and src == "DYNAMIC_TOPIC_CONFIG"
+        assert value == "delete" and src == "DEFAULT_CONFIG"
 
         kcl.alter_topic_config({"cleanup.policy": "compact"},
                                incremental=False,