partition_allocator/tests: add tests for internal topics

redpanda-data · Jun 30, 2023 · 5ca4dda · 5ca4dda
1 parent e3ae5c1
commit 5ca4dda
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 4 deletions.
diff --git a/src/v/cluster/tests/partition_allocator_fixture.h b/src/v/cluster/tests/partition_allocator_fixture.h
@@ -17,6 +17,7 @@
 #include "cluster/scheduling/allocation_strategy.h"
 #include "cluster/scheduling/partition_allocator.h"
 #include "config/configuration.h"
+#include "config/mock_property.h"
 #include "model/fundamental.h"
 #include "model/metadata.h"
 #include "net/unresolved_address.h"
@@ -37,7 +38,7 @@ struct partition_allocator_fixture {
         config::mock_binding<std::optional<int32_t>>(std::nullopt),
         config::mock_binding<uint32_t>(uint32_t{partitions_per_shard}),
         config::mock_binding<uint32_t>(uint32_t{partitions_reserve_shard0}),
-        config::mock_binding<std::vector<ss::sstring>>({}),
+        kafka_internal_topics.bind(),
         config::mock_binding<bool>(true)) {
         members.start().get0();
         ss::smp::invoke_on_all([] {
@@ -71,7 +72,7 @@ struct partition_allocator_fixture {
           broker.properties().cores,
           config::mock_binding<uint32_t>(uint32_t{partitions_per_shard}),
           config::mock_binding<uint32_t>(uint32_t{partitions_reserve_shard0}),
-          config::mock_binding<std::vector<ss::sstring>>({})));
+          kafka_internal_topics.bind()));
     }
 
     void saturate_all_machines() {
@@ -117,8 +118,13 @@ struct partition_allocator_fixture {
 
     cluster::allocation_request
     make_allocation_request(int partitions, uint16_t replication_factor) {
+        return make_allocation_request(tn, partitions, replication_factor);
+    }
+
+    cluster::allocation_request make_allocation_request(
+      model::topic_namespace tn, int partitions, uint16_t replication_factor) {
         cluster::allocation_request req(
-          tn, cluster::partition_allocation_domains::common);
+          std::move(tn), cluster::partition_allocation_domains::common);
         req.partitions.reserve(partitions);
         for (int i = 0; i < partitions; ++i) {
             req.partitions.emplace_back(
@@ -127,6 +133,7 @@ struct partition_allocator_fixture {
         return req;
     }
 
+    config::mock_property<std::vector<ss::sstring>> kafka_internal_topics{{}};
     model::topic_namespace tn{model::kafka_namespace, model::topic{"test"}};
     ss::sharded<cluster::members_table> members;
     cluster::partition_allocator allocator;

diff --git a/src/v/cluster/tests/partition_allocator_tests.cc b/src/v/cluster/tests/partition_allocator_tests.cc
@@ -85,7 +85,7 @@ FIXTURE_TEST(unregister_node, partition_allocator_fixture) {
     BOOST_REQUIRE_EQUAL(allocator.state().available_nodes(), 2);
 }
 
-FIXTURE_TEST(invalid_allocation_over_capacity, partition_allocator_fixture) {
+FIXTURE_TEST(allocation_over_capacity, partition_allocator_fixture) {
     register_node(0, 6);
     register_node(1, 6);
     register_node(2, 6);
@@ -96,6 +96,26 @@ FIXTURE_TEST(invalid_allocation_over_capacity, partition_allocator_fixture) {
       allocator.allocate(make_allocation_request(1, 1)).get().has_error());
     // group id hasn't changed
     BOOST_REQUIRE_EQUAL(allocator.state().last_group_id(), gr);
+
+    // Make the topic internal and retry, should work.
+    kafka_internal_topics.update({tn.tp()});
+    BOOST_REQUIRE(allocator.allocate(make_allocation_request(1, 1)).get());
+    BOOST_REQUIRE_GT(allocator.state().last_group_id(), gr);
+
+    // Undo the configuration, should fail again.
+    kafka_internal_topics.update({});
+    BOOST_REQUIRE(
+      allocator.allocate(make_allocation_request(1, 1)).get().has_error());
+
+    auto int_1 = model::topic_namespace{
+      model::ns{"redpanda"}, model::topic{"controller"}};
+    auto int_2 = model::topic_namespace{
+      model::ns{"kafka_internal"}, model::topic{"controller"}};
+    // Internal namespaces should work too.
+    BOOST_REQUIRE(
+      allocator.allocate(make_allocation_request(int_1, 1, 1)).get());
+    BOOST_REQUIRE(
+      allocator.allocate(make_allocation_request(int_2, 1, 1)).get());
 }
 
 FIXTURE_TEST(max_allocation, partition_allocator_fixture) {