cluster: Add topic_table test that checks

how the clear_partition_bootstrap_params_cmd command is handled

Signed-off-by: Evgeny Lazin <4lazin@gmail.com>

Author: Lazin

src/v/cluster/tests/topic_table_test.cc

@@ -810,3 +810,65 @@ FIXTURE_TEST(
     BOOST_REQUIRE(params1.has_value());
     BOOST_REQUIRE_EQUAL(params1->start_offset, model::offset(2000));
 }
+
+FIXTURE_TEST(test_clear_bootstrap_params, topic_table_fixture) {
+    // First, set some bootstrap params for multiple topics
+    absl::btree_map<model::partition_id, cluster::partition_bootstrap_params>
+      params1;
+    params1[model::partition_id(0)] = cluster::partition_bootstrap_params(
+      model::offset(1000), model::term_id(5));
+    params1[model::partition_id(1)] = cluster::partition_bootstrap_params(
+      model::offset(2000), model::term_id(10));
+
+    cluster::set_partition_bootstrap_params_cmd_data data1{
+      .tp_ns = make_tp_ns("topic1"), .partition_params = std::move(params1)};
+    cluster::set_partition_bootstrap_params_cmd cmd1(
+      make_tp_ns("topic1"), std::move(data1));
+
+    auto res = table.local().apply(std::move(cmd1), model::offset(0)).get();
+    BOOST_REQUIRE_EQUAL(res, cluster::errc::success);
+
+    absl::btree_map<model::partition_id, cluster::partition_bootstrap_params>
+      params2;
+    params2[model::partition_id(0)] = cluster::partition_bootstrap_params(
+      model::offset(3000), model::term_id(15));
+
+    cluster::set_partition_bootstrap_params_cmd_data data2{
+      .tp_ns = make_tp_ns("topic2"), .partition_params = std::move(params2)};
+    cluster::set_partition_bootstrap_params_cmd cmd2(
+      make_tp_ns("topic2"), std::move(data2));
+
+    res = table.local().apply(std::move(cmd2), model::offset(1)).get();
+    BOOST_REQUIRE_EQUAL(res, cluster::errc::success);
+
+    // Verify params are stored
+    auto ntp1_0 = model::ntp(
+      test_ns, model::topic("topic1"), model::partition_id(0));
+    auto ntp1_1 = model::ntp(
+      test_ns, model::topic("topic1"), model::partition_id(1));
+    auto ntp2_0 = model::ntp(
+      test_ns, model::topic("topic2"), model::partition_id(0));
+
+    BOOST_REQUIRE(
+      table.local().get_partition_bootstrap_params(ntp1_0).has_value());
+    BOOST_REQUIRE(
+      table.local().get_partition_bootstrap_params(ntp1_1).has_value());
+    BOOST_REQUIRE(
+      table.local().get_partition_bootstrap_params(ntp2_0).has_value());
+
+    // Now clear all bootstrap params
+    cluster::clear_partition_bootstrap_params_cmd_data clear_data{};
+    cluster::clear_partition_bootstrap_params_cmd clear_cmd(
+      int8_t{0}, std::move(clear_data));
+
+    res = table.local().apply(std::move(clear_cmd), model::offset(2)).get();
+    BOOST_REQUIRE_EQUAL(res, cluster::errc::success);
+
+    // Verify all params have been cleared
+    BOOST_REQUIRE(
+      !table.local().get_partition_bootstrap_params(ntp1_0).has_value());
+    BOOST_REQUIRE(
+      !table.local().get_partition_bootstrap_params(ntp1_1).has_value());
+    BOOST_REQUIRE(
+      !table.local().get_partition_bootstrap_params(ntp2_0).has_value());
+}