cloud_topics: gate tiered_cloud on full upgrade and add tiered_v1/tiered_v2 storage modes

src/v/cloud_topics/app.cc

@@ -194,7 +194,8 @@ ss::future<> app::construct(
       metadata_cache,
       &controller->get_shard_table(),
       &controller->get_partition_manager(),
-      connection_cache);
+      connection_cache,
+      &controller->get_feature_table());
 
     co_await construct_service(
       topic_manifest_upload_mgr, std::ref(*remote), bucket);

src/v/cloud_topics/level_zero/notifier/BUILD

@@ -20,6 +20,7 @@ redpanda_cc_library(
         "//src/v/cluster:partition_leaders_table",
         "//src/v/cluster:shard_table",
         "//src/v/cluster:state_machine_registry",
+        "//src/v/features",
         "//src/v/model",
         "//src/v/rpc",
         "//src/v/ssx:future_util",

src/v/cloud_topics/level_zero/notifier/level_zero_notifier.cc

@@ -21,6 +21,7 @@
 #include "cluster/partition_leaders_table.h"
 #include "cluster/partition_manager.h"
 #include "cluster/shard_table.h"
+#include "features/feature_table.h"
 #include "model/timeout_clock.h"
 #include "rpc/connection_cache.h"
 #include "ssx/future-util.h"
@@ -51,15 +52,23 @@ level_zero_notifier::level_zero_notifier(
   ss::sharded<cluster::shard_table>* shard_table,
   ss::sharded<cluster::partition_manager>* partition_manager,
   ss::sharded<rpc::connection_cache>* connections,
+  ss::sharded<features::feature_table>* features,
   std::chrono::milliseconds retry_backoff)
   : _self(self)
   , _leaders(leaders)
   , _metadata(metadata)
   , _shard_table(shard_table)
   , _partition_manager(partition_manager)
   , _connections(connections)
+  , _features(features)
   , _retry_backoff(retry_backoff) {}
 
+bool level_zero_notifier::notifications_enabled() const {
+    return _features != nullptr
+           && _features->local().is_active(
+             features::feature::tiered_cloud_topics);
+}
+
 std::optional<model::ntp>
 level_zero_notifier::resolve_ntp(const model::topic_id_partition& tidp) const {
     auto tns = _metadata->local().get_name_by_id(tidp.topic_id);
@@ -78,6 +87,15 @@ ss::future<> level_zero_notifier::stop() {
 ss::future<std::expected<void, ctp_stm_api_errc>>
 level_zero_notifier::set_min_allowed_local_threshold(
   model::topic_id_partition tidp, kafka::offset new_floor) {
+    if (!notifications_enabled()) {
+        vlog(
+          cd_log.debug,
+          "{} set_min_allowed_local_threshold: skipping notification, the "
+          "tiered_cloud_topics feature is not active yet, new floor {}",
+          tidp,
+          new_floor);
+        co_return std::expected<void, ctp_stm_api_errc>{};
+    }
     auto ntp = resolve_ntp(tidp);
     if (!ntp.has_value()) {
         vlog(
@@ -104,6 +122,15 @@ level_zero_notifier::set_min_allowed_local_threshold(
 ss::future<std::expected<void, ctp_stm_api_errc>>
 level_zero_notifier::set_min_allowed_local_threshold_locally(
   model::topic_id_partition tidp, kafka::offset new_floor) {
+    if (!notifications_enabled()) {
+        vlog(
+          cd_log.debug,
+          "{} set_min_allowed_local_threshold_locally: skipping notification, "
+          "the tiered_cloud_topics feature is not active yet, new floor {}",
+          tidp,
+          new_floor);
+        co_return std::expected<void, ctp_stm_api_errc>{};
+    }
     auto ntp = resolve_ntp(tidp);
     if (!ntp.has_value()) {
         vlog(

src/v/cloud_topics/level_zero/notifier/level_zero_notifier.h

@@ -12,6 +12,7 @@
 
 #include "cloud_topics/level_zero/stm/ctp_stm_api.h"
 #include "cluster/fwd.h"
+#include "features/fwd.h"
 #include "model/fundamental.h"
 #include "rpc/fwd.h"
 #include "ssx/semaphore.h"
@@ -35,6 +36,18 @@ namespace cloud_topics {
 /// replicates the floor through ctp_stm_api, retrying transient failures up to
 /// max_attempts. The replication result is returned to the caller, which
 /// decides whether a failure is fatal -- the call is not fire-and-forget.
+///
+/// While the tiered_cloud_topics feature is not active (the cluster is not
+/// fully upgraded to v26.2) every notification attempt is a no-op that
+/// reports success. The gate is not hypothetical: storage.mode=cloud topics
+/// may exist during the upgrade (only tiered_cloud creation is blocked) and
+/// L1 compaction of a cloud topic hands the notifier a new floor like any
+/// other, so without the gate the set_min_allowed_local_threshold stm
+/// command would be replicated to pre-v26.2 replicas that cannot apply it.
+/// Skipping the floor update is safe in the meantime: cloud-mode reads are
+/// routed through the last reconciled offset rather than the floor, and no
+/// tiered_cloud topic (the only reader of local data below the floor) can
+/// exist until the feature is active.
 class level_zero_notifier
   : public ss::peering_sharded_service<level_zero_notifier> {
 public:
@@ -54,6 +67,7 @@ class level_zero_notifier
       ss::sharded<cluster::shard_table>* shard_table,
       ss::sharded<cluster::partition_manager>* partition_manager,
       ss::sharded<rpc::connection_cache>* connections,
+      ss::sharded<features::feature_table>* features,
       std::chrono::milliseconds retry_backoff = default_retry_backoff);
 
     ss::future<> stop();
@@ -85,6 +99,10 @@ class level_zero_notifier
     replicate(model::ntp ntp, ctp_stm_api& api, kafka::offset new_floor);
 
 private:
+    // True once the tiered_cloud_topics feature is active (the cluster is
+    // fully upgraded to v26.2). A null feature table reads as inactive.
+    bool notifications_enabled() const;
+
     // Resolve a topic_id_partition to an ntp via the metadata cache. Returns
     // nullopt when the topic id is unknown (e.g. deleted topic or stale
     // metadata on this node).
@@ -122,6 +140,7 @@ class level_zero_notifier
     ss::sharded<cluster::shard_table>* _shard_table;
     ss::sharded<cluster::partition_manager>* _partition_manager;
     ss::sharded<rpc::connection_cache>* _connections;
+    ss::sharded<features::feature_table>* _features;
     std::chrono::milliseconds _retry_backoff;
     ssx::semaphore _inflight{
       max_concurrent_replications, "level_zero_notifier::inflight"};

src/v/cloud_topics/level_zero/notifier/tests/BUILD

@@ -13,6 +13,7 @@ redpanda_cc_gtest(
         "//src/v/cloud_topics/level_zero/stm:ctp_stm_api",
         "//src/v/cloud_topics/level_zero/stm:ctp_stm_factory",
         "//src/v/cluster:state_machine_registry",
+        "//src/v/features",
         "//src/v/model",
         "//src/v/raft/tests:raft_fixture",
         "//src/v/rpc",

src/v/cloud_topics/level_zero/notifier/tests/level_zero_notifier_test.cc

@@ -13,6 +13,7 @@
 #include "cloud_topics/level_zero/stm/ctp_stm.h"
 #include "cloud_topics/level_zero/stm/ctp_stm_api.h"
 #include "cloud_topics/logger.h"
+#include "features/feature_table.h"
 #include "model/fundamental.h"
 #include "raft/tests/raft_fixture.h"
 #include "rpc/errc.h"
@@ -54,7 +55,14 @@ TEST_F_CORO(level_zero_notifier_fixture, replicate_to_leader_succeeds) {
     co_await wait_for_leader(raft::default_timeout());
 
     ct::level_zero_notifier notifier(
-      self_node, nullptr, nullptr, nullptr, nullptr, nullptr, tiny_backoff);
+      self_node,
+      nullptr,
+      nullptr,
+      nullptr,
+      nullptr,
+      nullptr,
+      nullptr,
+      tiny_backoff);
     auto leader_api = api(node(*get_leader()));
     auto res = co_await notifier.replicate(
       test_ntp, leader_api, kafka::offset(42));
@@ -80,7 +88,14 @@ TEST_F_CORO(level_zero_notifier_fixture, gives_up_on_follower) {
     ASSERT_TRUE_CORO(follower != nullptr);
 
     ct::level_zero_notifier notifier(
-      self_node, nullptr, nullptr, nullptr, nullptr, nullptr, tiny_backoff);
+      self_node,
+      nullptr,
+      nullptr,
+      nullptr,
+      nullptr,
+      nullptr,
+      nullptr,
+      tiny_backoff);
     auto follower_api = api(*follower);
     auto res = co_await notifier.replicate(
       test_ntp, follower_api, kafka::offset(7));
@@ -89,6 +104,38 @@ TEST_F_CORO(level_zero_notifier_fixture, gives_up_on_follower) {
     ASSERT_FALSE_CORO(res.has_value());
 }
 
+// While the tiered_cloud_topics feature is not active, notification attempts
+// are no-ops that report success without touching any of the notifier's
+// dependencies (all null here, so reaching past the gate would crash).
+TEST_F_CORO(
+  level_zero_notifier_fixture, notification_is_noop_until_feature_active) {
+    ss::sharded<features::feature_table> features;
+    co_await features.start();
+
+    ct::level_zero_notifier notifier(
+      self_node,
+      nullptr,
+      nullptr,
+      nullptr,
+      nullptr,
+      nullptr,
+      &features,
+      tiny_backoff);
+    const model::topic_id_partition tidp(
+      model::topic_id(), model::partition_id(0));
+
+    auto res = co_await notifier.set_min_allowed_local_threshold(
+      tidp, kafka::offset(42));
+    ASSERT_TRUE_CORO(res.has_value());
+
+    auto local_res = co_await notifier.set_min_allowed_local_threshold_locally(
+      tidp, kafka::offset(42));
+    ASSERT_TRUE_CORO(local_res.has_value());
+
+    co_await notifier.stop();
+    co_await features.stop();
+}
+
 TEST(level_zero_notifier_routing, map_transport_error) {
     using ct::notifier_detail::map_transport_error;
     EXPECT_EQ(

src/v/cluster_link/manager.cc

@@ -92,6 +92,7 @@ manager::manager(
   std::unique_ptr<partition_metadata_provider> partition_metadata_provider,
   std::unique_ptr<kafka_rpc_client_service> kafka_rpc_client_service,
   std::unique_ptr<members_table_provider> members_table_provider,
+  ss::sharded<features::feature_table>* feature_table,
   ss::lowres_clock::duration task_reconciler_interval,
   config::binding<int16_t> default_topic_replication,
   ss::scheduling_group scheduling_group)
@@ -102,6 +103,7 @@ manager::manager(
   , _topic_creator(std::move(topic_creator))
   , _security_service(std::move(security_service))
   , _registry(std::move(registry))
+  , _feature_table(feature_table)
   , _link_factory(std::move(link_factory))
   , _cluster_factory(std::move(cluster_factory))
   , _group_router(std::move(group_router))
@@ -999,6 +1001,7 @@ ss::future<> manager::on_controller_leadership(::model::term_id term) {
           _topic_creator.get(),
           _topic_metadata_cache.get(),
           _registry.get(),
+          _feature_table,
           topic_reconciler_interval,
           _default_topic_replication,
           _scheduling_group);

src/v/cluster_link/manager.h

@@ -21,6 +21,7 @@
 #include "cluster_link/topic_reconciler.h"
 #include "cluster_link/types.h"
 #include "container/chunked_vector.h"
+#include "features/fwd.h"
 #include "kafka/data/rpc/deps.h"
 #include "kafka/data/rpc/fwd.h"
 #include "model/fundamental.h"
@@ -56,6 +57,7 @@ class manager {
       std::unique_ptr<partition_metadata_provider> partition_metadata_provider,
       std::unique_ptr<kafka_rpc_client_service> kafka_rpc_client_service,
       std::unique_ptr<members_table_provider> members_table_provider,
+      ss::sharded<features::feature_table>* feature_table,
       ss::lowres_clock::duration task_reconciler_interval,
       config::binding<int16_t> default_topic_replication,
       ss::scheduling_group scheduling_group);
@@ -236,6 +238,7 @@ class manager {
     std::unique_ptr<kafka::data::rpc::topic_creator> _topic_creator;
     std::unique_ptr<security_service> _security_service;
     std::unique_ptr<link_registry> _registry;
+    ss::sharded<features::feature_table>* _feature_table;
     std::unique_ptr<link_factory> _link_factory;
     std::unique_ptr<cluster_factory> _cluster_factory;
     std::unique_ptr<topic_reconciler> _topic_reconciler;

src/v/cluster_link/model/types.h

@@ -70,6 +70,7 @@ inline auto default_synced_topic_properties = std::to_array<std::string_view>({
   kafka::topic_property_min_compaction_lag_ms,
   kafka::topic_property_max_compaction_lag_ms,
   kafka::topic_property_redpanda_storage_mode,
+  kafka::topic_property_redpanda_storage_mode_impl,
 });
 
 /// List of topic properties that are not permitted to be synced

src/v/cluster_link/service.cc

@@ -1299,6 +1299,7 @@ ss::future<> service::maybe_start_manager() {
         _hm_frontend),
       kafka_rpc_client_service::make_default(_kafka_data_rpc_client),
       members_table_provider::make_default(&_controller->get_members_table()),
+      &_controller->get_feature_table(),
       30s, // Temporary until we have a proper configuration for this
       config::shard_local_cfg().default_topic_replication.bind(),
       _scheduling_group);

src/v/cluster_link/tests/BUILD

@@ -20,6 +20,7 @@ redpanda_test_cc_library(
         "//src/v/cluster_link/replication/tests:deps_test_impl",
         "//src/v/config",
         "//src/v/container:chunked_vector",
+        "//src/v/features",
         "//src/v/kafka/client/test:cluster_mock",
         "//src/v/kafka/data/rpc",
         "//src/v/kafka/data/rpc/test:test_deps",
@@ -138,7 +139,9 @@ redpanda_cc_gtest(
         ":test_deps",
         "//src/v/cluster:cluster_link_table",
         "//src/v/cluster_link:impl",
+        "//src/v/cluster_link/utils:topic_properties_utils",
         "//src/v/config",
+        "//src/v/features",
         "//src/v/test_utils:gtest",
         "@googletest//:gtest",
     ],

src/v/cluster_link/tests/deps.cc

@@ -58,6 +58,9 @@ ss::future<> cluster_link_manager_test_fixture::wire_up_and_start(
     _ftpc = ftpc.get();
 
     _lf = lf.get();
+    co_await _feature_table.start();
+    co_await _feature_table.invoke_on_all(
+      [](features::feature_table& ft) { ft.testing_activate_all(); });
     co_await _manager.start_single(
       _self,
       ss::sharded_parameter([&fplc]() { return std::move(fplc); }),
@@ -101,6 +104,7 @@ ss::future<> cluster_link_manager_test_fixture::wire_up_and_start(
           _fmtp = fmtp.get();
           return fmtp;
       }),
+      &_feature_table,
       1s,
       _default_topic_replication.bind(),
       ss::default_scheduling_group());
@@ -119,6 +123,7 @@ ss::future<> cluster_link_manager_test_fixture::reset() {
     _notification_cleanups.clear();
     _lf = nullptr;
     co_await _manager.stop();
+    co_await _feature_table.stop();
     _fss = nullptr;
     _tmc = nullptr;
     _fpm = nullptr;

src/v/cluster_link/tests/deps.h

@@ -19,6 +19,7 @@
 #include "cluster_link/utils.h"
 #include "config/mock_property.h"
 #include "container/chunked_vector.h"
+#include "features/feature_table.h"
 #include "kafka/client/test/cluster_mock.h"
 #include "kafka/data/rpc/deps.h"
 #include "kafka/data/rpc/test/deps.h"
@@ -794,6 +795,7 @@ class cluster_link_manager_test_fixture {
     test_kafka_rpc_client_service* _tkrcs{nullptr};
     fake_members_table_provider* _fmtp{nullptr};
     ss::sharded<manager> _manager;
+    ss::sharded<features::feature_table> _feature_table;
     config::mock_property<int16_t> _default_topic_replication{1};
 
     ::model::node_id _self;

src/v/cluster_link/tests/link_test.cc

@@ -194,6 +194,7 @@ class link_test : public link_test_base {
           std::make_unique<test_partition_metadata_provider>(),
           std::make_unique<test_kafka_rpc_client_service>(_tmc),
           std::make_unique<fake_members_table_provider>(),
+          nullptr,
           task_reconciler_interval,
           _default_topic_replication.bind(),
           ss::default_scheduling_group());
@@ -440,6 +441,7 @@ class evil_link_test : public link_test_base {
           std::make_unique<test_partition_metadata_provider>(),
           std::make_unique<test_kafka_rpc_client_service>(_tmc),
           std::make_unique<fake_members_table_provider>(),
+          nullptr,
           task_reconciler_interval,
           _default_topic_replication.bind(),
           ss::default_scheduling_group());