ct: add test config for fixtures

Some tests will want to start the Redpanda application with certain
features turned off. Rather than exposing this abililty to users with a
cluster property, this plumbs a struct through the application to be
used by test fixtures.

This is done in this commit to disable the flush loop when running
replicated_metastore_test, which will exercise flushes of its own.

Now that this is configurable and set appropriately for
replicated_metastore_test, the default behavior is also switched to
_not_ skipping the flush loop.

Author: andrwng

src/v/cloud_topics/BUILD

@@ -24,6 +24,18 @@ redpanda_cc_library(
     ],
 )
 
+redpanda_cc_library(
+    name = "test_fixture_cfg",
+    hdrs = [
+        "test_fixture_cfg.h",
+    ],
+    visibility = [
+        "//src/v/cloud_topics/tests:__pkg__",
+        "//src/v/redpanda:__pkg__",
+        "//src/v/redpanda/tests:__pkg__",
+    ],
+)
+
 redpanda_cc_library(
     name = "types",
     srcs = [
@@ -80,6 +92,7 @@ redpanda_cc_library(
         ":cluster_services_interface",
         ":data_plane_impl",
         "//src/v/cloud_topics/housekeeper:manager",
+        "//src/v/cloud_topics/level_one/metastore:flush_loop",
         "//src/v/cloud_topics/level_one/metastore:topic_purger",
         "//src/v/cloud_topics/level_zero/gc:level_zero_gc",
         "//src/v/cloud_topics/manager",

src/v/cloud_topics/app.h

@@ -65,7 +65,7 @@ class app : public ssx::sharded_service_container {
       ss::sharded<rpc::connection_cache>*,
       cloud_storage_clients::bucket_name,
       ss::sharded<storage::api>*,
-      bool skip_flush_loop = true);
+      bool skip_flush_loop = false);
 
     ss::future<> start();
 

src/v/cloud_topics/level_one/metastore/tests/replicated_metastore_test.cc

@@ -41,9 +41,16 @@ class ReplicatedMetastoreTest
 
     bool is_lsm_backend() const { return GetParam() == metastore_backend::lsm; }
 
+    cloud_topics::test_fixture_cfg fixture_cfg() const {
+        return {
+          .use_lsm_metastore = is_lsm_backend(),
+          // Skip flushing since tests may exercise flushing.
+          .skip_flush_loop = true,
+        };
+    }
     void SetUp() override {
         for (size_t i = 0; i < num_brokers; i++) {
-            add_node(is_lsm_backend());
+            add_node(fixture_cfg());
         }
         wait_for_all_members(5s).get();
     }
@@ -1007,7 +1014,7 @@ TEST_P(ReplicatedMetastoreTest, TestBasicFlushAndRestore) {
     // Restart all nodes.
     // NOTE: the added nodes get the same node IDs 0, 1, 2.
     for (size_t i = 0; i < num_brokers; i++) {
-        add_node(is_lsm_backend());
+        add_node(fixture_cfg());
     }
     wait_for_all_members(5s).get();
 

src/v/cloud_topics/test_fixture_cfg.h

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2025 Redpanda Data, Inc.
+ *
+ * Licensed as a Redpanda Enterprise file under the Redpanda Community
+ * License (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
+ */
+
+#pragma once
+
+namespace cloud_topics {
+
+// Configuration for cloud topics test fixtures.
+struct test_fixture_cfg {
+    bool use_lsm_metastore{true};
+    bool skip_flush_loop{false};
+};
+
+} // namespace cloud_topics

src/v/cloud_topics/tests/BUILD

@@ -7,6 +7,7 @@ redpanda_test_cc_library(
     deps = [
         "//src/v/cloud_io/tests:s3_imposter",
         "//src/v/cloud_topics:app",
+        "//src/v/cloud_topics:test_fixture_cfg",
         "//src/v/cloud_topics/level_one/metastore:simple_stm",
         "//src/v/cluster/tests:cluster_test_fixture",
         "//src/v/kafka/server/tests:kafka_test_utils",

src/v/cloud_topics/tests/cluster_fixture.h

@@ -14,6 +14,7 @@
 #include "cloud_topics/app.h"
 #include "cloud_topics/level_one/metastore/lsm/stm.h"
 #include "cloud_topics/level_one/metastore/simple_stm.h"
+#include "cloud_topics/test_fixture_cfg.h"
 #include "cluster/tests/cluster_test_fixture.h"
 #include "kafka/server/tests/produce_consume_utils.h"
 #include "model/fundamental.h"
@@ -39,7 +40,7 @@ class cluster_fixture
             remove_node_application(id);
         }
     }
-    void add_node(bool use_lsm_metastore = true) {
+    void add_node(cloud_topics::test_fixture_cfg ct_test_cfg = {}) {
         static constexpr int kafka_port_base = 9092;
         static constexpr int rpc_port_base = 11000;
         auto [s3_conf, a_conf, cs_conf] = get_cloud_storage_configurations(
@@ -60,7 +61,7 @@ class cluster_fixture
           /*cloud_topics_enabled=*/true,
           /*cluster_linking_enabled=*/false,
           /*seed_node_id=*/model::node_id{0},
-          use_lsm_metastore);
+          ct_test_cfg);
     }
 
     ss::future<kafka_produce_transport*> make_producer(model::node_id id) {

src/v/cluster/tests/cluster_test_fixture.h

@@ -122,7 +122,7 @@ class cluster_test_fixture {
       bool iceberg_enabled = false,
       bool cloud_topics_enabled = false,
       bool cluster_linking_enabled = false,
-      bool use_lsm_metastore = false) {
+      cloud_topics::test_fixture_cfg ct_test_cfg = {}) {
         return std::make_unique<redpanda_thread_fixture>(
           node_id,
           kafka_port,
@@ -142,7 +142,7 @@ class cluster_test_fixture {
           iceberg_enabled,
           cloud_topics_enabled,
           cluster_linking_enabled,
-          use_lsm_metastore);
+          ct_test_cfg);
     }
 
     void add_node(
@@ -163,7 +163,7 @@ class cluster_test_fixture {
       bool iceberg_enabled = false,
       bool cloud_topics_enabled = false,
       bool cluster_linking_enabled = false,
-      bool use_lsm_metastore = true) {
+      cloud_topics::test_fixture_cfg ct_test_cfg = {}) {
         _instances.emplace(
           node_id,
           make_redpanda_fixture(
@@ -182,7 +182,7 @@ class cluster_test_fixture {
             iceberg_enabled,
             cloud_topics_enabled,
             cluster_linking_enabled,
-            use_lsm_metastore));
+            ct_test_cfg));
     }
 
     application* get_node_application(model::node_id id) {
@@ -225,7 +225,7 @@ class cluster_test_fixture {
       bool cloud_topics_enabled = false,
       bool cluster_linking_enabled = false,
       model::node_id seed_node_id = model::node_id{0},
-      bool use_lsm_metastore = true) {
+      cloud_topics::test_fixture_cfg ct_test_cfg = {}) {
         std::vector<config::seed_server> seeds = {};
         if (!empty_seed_starts_cluster_val || node_id != 0) {
             seeds.push_back(
@@ -250,7 +250,7 @@ class cluster_test_fixture {
           iceberg_enabled,
           cloud_topics_enabled,
           cluster_linking_enabled,
-          use_lsm_metastore);
+          ct_test_cfg);
         return get_node_application(node_id);
     }
 

src/v/redpanda/BUILD

@@ -30,6 +30,7 @@ redpanda_cc_library(
         "//src/v/cloud_storage",
         "//src/v/cloud_storage_clients",
         "//src/v/cloud_topics:app",
+        "//src/v/cloud_topics:test_fixture_cfg",
         "//src/v/cloud_topics/level_one/metastore:leader_router",
         "//src/v/cloud_topics/level_one/metastore:simple_stm",
         "//src/v/cloud_topics/level_one/metastore/lsm:stm",

src/v/redpanda/application.h

@@ -14,6 +14,7 @@
 #include "base/seastarx.h"
 #include "cloud_storage/fwd.h"
 #include "cloud_topics/app.h"
+#include "cloud_topics/test_fixture_cfg.h"
 #include "cluster/archival/fwd.h"
 #include "cluster/config_manager.h"
 #include "cluster/fwd.h"
@@ -97,7 +98,9 @@ class application : public ssx::sharded_service_container {
       std::optional<YAML::Node> audit_log_client_cfg = std::nullopt);
     void check_environment();
     void wire_up_and_start(
-      ::stop_signal&, bool test_mode = false, bool use_lsm_metastore = true);
+      ::stop_signal&,
+      bool test_mode = false,
+      cloud_topics::test_fixture_cfg ct_test_cfg = {});
     void post_start_tasks();
 
     void init_crashtracker(::stop_signal& app_signal);
@@ -224,13 +227,16 @@ class application : public ssx::sharded_service_container {
     void start_bootstrap_services();
 
     // Constructs services across shards meant for Redpanda runtime.
-    void
-    wire_up_runtime_services(model::node_id node_id, ::stop_signal& app_signal);
+    void wire_up_runtime_services(
+      model::node_id node_id,
+      ::stop_signal& app_signal,
+      cloud_topics::test_fixture_cfg ct_test_cfg);
     void configure_admin_server(model::node_id);
     void wire_up_redpanda_services(
       model::node_id,
       ::stop_signal& app_signal,
-      std::optional<cloud_storage_clients::bucket_name>& bucket_name);
+      std::optional<cloud_storage_clients::bucket_name>& bucket_name,
+      cloud_topics::test_fixture_cfg ct_test_cfg);
 
     void load_feature_table_snapshot();
 
@@ -239,7 +245,9 @@ class application : public ssx::sharded_service_container {
     // Starts the services meant for Redpanda runtime. Must be called after
     // having constructed the subsystems via the corresponding `wire_up` calls.
     void start_runtime_services(
-      cluster::cluster_discovery&, ::stop_signal&, bool use_lsm_metastore);
+      cluster::cluster_discovery&,
+      ::stop_signal&,
+      cloud_topics::test_fixture_cfg ct_test_cfg);
     void start_kafka(const model::node_id&, ::stop_signal&);
     void add_runtime_rpc_services(rpc::rpc_server&, bool start_raft_rpc_early);
 

src/v/redpanda/application_bootstrap.cc

@@ -385,7 +385,9 @@ void application::start_bootstrap_services() {
 }
 
 void application::wire_up_and_start(
-  ::stop_signal& app_signal, bool test_mode, bool use_lsm_metastore) {
+  ::stop_signal& app_signal,
+  bool test_mode,
+  cloud_topics::test_fixture_cfg ct_test_cfg) {
     // Setup the app level abort service
     construct_service(_as).get();
 
@@ -510,7 +512,7 @@ void application::wire_up_and_start(
       node_id,
       storage.local().get_cluster_uuid());
 
-    wire_up_runtime_services(node_id, app_signal);
+    wire_up_runtime_services(node_id, app_signal, ct_test_cfg);
 
     if (test_mode) {
         // When running inside a unit test fixture, we may fast-forward
@@ -545,7 +547,7 @@ void application::wire_up_and_start(
         controller->set_ready().get();
     }
 
-    start_runtime_services(cd, app_signal, use_lsm_metastore);
+    start_runtime_services(cd, app_signal, ct_test_cfg);
 
     if (_proxy_config && !config::node().recovery_mode_enabled) {
         _proxy->start().get();