diff --git a/src/v/cloud_topics/app.cc b/src/v/cloud_topics/app.cc index ecfb5414ce2e4..1caf9ab12ba20 100644 --- a/src/v/cloud_topics/app.cc +++ b/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); diff --git a/src/v/cloud_topics/level_zero/notifier/BUILD b/src/v/cloud_topics/level_zero/notifier/BUILD index e57844f04b8aa..10130b64daf28 100644 --- a/src/v/cloud_topics/level_zero/notifier/BUILD +++ b/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", diff --git a/src/v/cloud_topics/level_zero/notifier/level_zero_notifier.cc b/src/v/cloud_topics/level_zero/notifier/level_zero_notifier.cc index 16fc9557b3215..e6ea4b6654195 100644 --- a/src/v/cloud_topics/level_zero/notifier/level_zero_notifier.cc +++ b/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,6 +52,7 @@ level_zero_notifier::level_zero_notifier( ss::sharded* shard_table, ss::sharded* partition_manager, ss::sharded* connections, + ss::sharded* features, std::chrono::milliseconds retry_backoff) : _self(self) , _leaders(leaders) @@ -58,8 +60,15 @@ level_zero_notifier::level_zero_notifier( , _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 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> 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{}; + } auto ntp = resolve_ntp(tidp); if (!ntp.has_value()) { vlog( @@ -104,6 +122,15 @@ level_zero_notifier::set_min_allowed_local_threshold( ss::future> 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{}; + } auto ntp = resolve_ntp(tidp); if (!ntp.has_value()) { vlog( diff --git a/src/v/cloud_topics/level_zero/notifier/level_zero_notifier.h b/src/v/cloud_topics/level_zero/notifier/level_zero_notifier.h index 2ccbeae809500..74ab0cef2e652 100644 --- a/src/v/cloud_topics/level_zero/notifier/level_zero_notifier.h +++ b/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 { public: @@ -54,6 +67,7 @@ class level_zero_notifier ss::sharded* shard_table, ss::sharded* partition_manager, ss::sharded* connections, + ss::sharded* 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* _shard_table; ss::sharded* _partition_manager; ss::sharded* _connections; + ss::sharded* _features; std::chrono::milliseconds _retry_backoff; ssx::semaphore _inflight{ max_concurrent_replications, "level_zero_notifier::inflight"}; diff --git a/src/v/cloud_topics/level_zero/notifier/tests/BUILD b/src/v/cloud_topics/level_zero/notifier/tests/BUILD index 8848126cbab0a..2d568c7815339 100644 --- a/src/v/cloud_topics/level_zero/notifier/tests/BUILD +++ b/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", diff --git a/src/v/cloud_topics/level_zero/notifier/tests/level_zero_notifier_test.cc b/src/v/cloud_topics/level_zero/notifier/tests/level_zero_notifier_test.cc index 6a27816e9338f..90c2a0dbdb77b 100644 --- a/src/v/cloud_topics/level_zero/notifier/tests/level_zero_notifier_test.cc +++ b/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; + 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( diff --git a/src/v/cluster_link/manager.cc b/src/v/cluster_link/manager.cc index 48d64ab452244..31ee0b4b5e9ba 100644 --- a/src/v/cluster_link/manager.cc +++ b/src/v/cluster_link/manager.cc @@ -93,6 +93,7 @@ manager::manager( std::unique_ptr kafka_rpc_client_service, std::unique_ptr members_table_provider, std::unique_ptr sr_preflight, + ss::sharded* feature_table, ss::lowres_clock::duration task_reconciler_interval, config::binding default_topic_replication, ss::scheduling_group scheduling_group) @@ -103,6 +104,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)) @@ -1000,6 +1002,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); diff --git a/src/v/cluster_link/manager.h b/src/v/cluster_link/manager.h index c4faf99f15f7b..17173454beb67 100644 --- a/src/v/cluster_link/manager.h +++ b/src/v/cluster_link/manager.h @@ -22,6 +22,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" @@ -58,6 +59,7 @@ class manager { std::unique_ptr kafka_rpc_client_service, std::unique_ptr members_table_provider, std::unique_ptr sr_preflight, + ss::sharded* feature_table, ss::lowres_clock::duration task_reconciler_interval, config::binding default_topic_replication, ss::scheduling_group scheduling_group); @@ -237,6 +239,7 @@ class manager { std::unique_ptr _topic_creator; std::unique_ptr _security_service; std::unique_ptr _registry; + ss::sharded* _feature_table; std::unique_ptr _link_factory; std::unique_ptr _cluster_factory; std::unique_ptr _topic_reconciler; diff --git a/src/v/cluster_link/model/types.h b/src/v/cluster_link/model/types.h index 56a2f11b3616c..3e19fab3a3fd7 100644 --- a/src/v/cluster_link/model/types.h +++ b/src/v/cluster_link/model/types.h @@ -70,6 +70,7 @@ inline auto default_synced_topic_properties = std::to_array({ 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 diff --git a/src/v/cluster_link/service.cc b/src/v/cluster_link/service.cc index 71b33b5df6c34..73a61b6d376c7 100644 --- a/src/v/cluster_link/service.cc +++ b/src/v/cluster_link/service.cc @@ -1306,6 +1306,7 @@ ss::future<> service::maybe_start_manager() { members_table_provider::make_default(&_controller->get_members_table()), sr_preflight_checker::make_default( *_schema_registry_dest, source_sr_prober::make_default()), + &_controller->get_feature_table(), 30s, // Temporary until we have a proper configuration for this config::shard_local_cfg().default_topic_replication.bind(), _scheduling_group); diff --git a/src/v/cluster_link/tests/BUILD b/src/v/cluster_link/tests/BUILD index f3bb0b156ad8d..2486fd29e0e51 100644 --- a/src/v/cluster_link/tests/BUILD +++ b/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", @@ -145,7 +146,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", ], diff --git a/src/v/cluster_link/tests/deps.cc b/src/v/cluster_link/tests/deps.cc index 3ac2eff2fe616..7d9d919b5e0e4 100644 --- a/src/v/cluster_link/tests/deps.cc +++ b/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); }), @@ -105,6 +108,7 @@ ss::future<> cluster_link_manager_test_fixture::wire_up_and_start( return sr_preflight_checker::make_default( _fake_schema_registry, std::make_unique()); }), + &_feature_table, 1s, _default_topic_replication.bind(), ss::default_scheduling_group()); @@ -123,6 +127,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; diff --git a/src/v/cluster_link/tests/deps.h b/src/v/cluster_link/tests/deps.h index fa936d207bf5c..35330b69851d8 100644 --- a/src/v/cluster_link/tests/deps.h +++ b/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" @@ -817,6 +818,7 @@ class cluster_link_manager_test_fixture { fake_members_table_provider* _fmtp{nullptr}; schema::fake_registry _fake_schema_registry; ss::sharded _manager; + ss::sharded _feature_table; config::mock_property _default_topic_replication{1}; ::model::node_id _self; diff --git a/src/v/cluster_link/tests/link_test.cc b/src/v/cluster_link/tests/link_test.cc index 676c16f9a6519..571f5a617a9c9 100644 --- a/src/v/cluster_link/tests/link_test.cc +++ b/src/v/cluster_link/tests/link_test.cc @@ -219,6 +219,7 @@ class link_test : public link_test_base { std::make_unique(), sr_preflight_checker::make_default( _fake_schema_registry, std::move(sr_prober)), + nullptr, task_reconciler_interval, _default_topic_replication.bind(), ss::default_scheduling_group()); @@ -468,6 +469,7 @@ class evil_link_test : public link_test_base { std::make_unique(), sr_preflight_checker::make_default( _fake_schema_registry, std::make_unique()), + nullptr, task_reconciler_interval, _default_topic_replication.bind(), ss::default_scheduling_group()); diff --git a/src/v/cluster_link/tests/topic_reconciler_test.cc b/src/v/cluster_link/tests/topic_reconciler_test.cc index c73b5f6f5433d..a17e86653bc4e 100644 --- a/src/v/cluster_link/tests/topic_reconciler_test.cc +++ b/src/v/cluster_link/tests/topic_reconciler_test.cc @@ -11,7 +11,9 @@ #include "cluster_link/tests/deps.h" #include "cluster_link/topic_reconciler.h" +#include "cluster_link/utils/topic_properties_utils.h" #include "config/mock_property.h" +#include "features/feature_table.h" #include "test_utils/async.h" #include "test_utils/test.h" @@ -49,10 +51,14 @@ class topic_reconciler_test : public seastar_test { }, _default_topic_replication.bind()); + // Left un-activated: tests exercising the tiered_cloud_topics gate + // activate features themselves; the other tests never hit it. + co_await _feature_table.start(); _reconciler = std::make_unique( _ftc.get(), _ftmc.get(), _link_registry.get(), + &_feature_table, 1s, _default_topic_replication.bind(), ss::default_scheduling_group()); @@ -75,6 +81,7 @@ class topic_reconciler_test : public seastar_test { virtual ss::future<> TearDownAsync() override { co_await _reconciler->stop(); + co_await _feature_table.stop(); _reconciler.reset(nullptr); _ftc.reset(nullptr); @@ -170,6 +177,9 @@ class topic_reconciler_test : public seastar_test { return _default_topic_replication; } +protected: + ss::sharded _feature_table; + private: ss::sharded _table; std::unique_ptr _link_registry; @@ -237,6 +247,41 @@ TEST_F_CORO(topic_reconciler_test, test_topic_creation_and_property_updates) { }); } +TEST_F_CORO(topic_reconciler_test, storage_mode_update_gated_until_upgrade) { + // The shadow-link apply path bypasses the kafka handlers, so it must + // enforce the tiered_cloud_topics upgrade gate itself: no mirror topic + // may move to the tiered_v2 impl while the cluster is not fully + // upgraded to v26.2. + cluster::topic_configuration cfg( + ::model::kafka_namespace, ::model::topic("mirror"), 1, 1); + cfg.properties.storage_mode = ::model::redpanda_storage_mode::cloud; + cluster::topic_properties_update update( + {::model::kafka_namespace, ::model::topic("mirror")}); + + EXPECT_THROW( + utils::maybe_append_storage_mode_update( + update, + std::make_optional("tiered"), + std::make_optional("tiered_v2"), + cfg, + _feature_table.local()), + kafka::validation_error); + + co_await _feature_table.invoke_on_all( + [](features::feature_table& ft) { ft.testing_activate_all(); }); + EXPECT_TRUE( + utils::maybe_append_storage_mode_update( + update, + std::make_optional("tiered"), + std::make_optional("tiered_v2"), + cfg, + _feature_table.local())); + EXPECT_EQ( + update.properties.storage_mode.value, + ::model::redpanda_storage_mode::tiered_cloud); + co_return; +} + TEST_F_CORO(topic_reconciler_test, test_topic_failure) { ::model::topic_namespace topic{ ::model::kafka_namespace, ::model::topic{"test_topic"}}; diff --git a/src/v/cluster_link/topic_reconciler.cc b/src/v/cluster_link/topic_reconciler.cc index e62b5c1b06ee8..9b629a570da8b 100644 --- a/src/v/cluster_link/topic_reconciler.cc +++ b/src/v/cluster_link/topic_reconciler.cc @@ -16,6 +16,7 @@ #include "cluster_link/logger.h" #include "cluster_link/model/types.h" #include "cluster_link/utils/topic_properties_utils.h" +#include "features/feature_table.h" #include "model/fundamental.h" #include "ssx/future-util.h" @@ -30,12 +31,14 @@ topic_reconciler::topic_reconciler( kafka::data::rpc::topic_creator* topic_creator, kafka::data::rpc::topic_metadata_cache* topic_metadata_cache, link_registry* link_registry, + ss::sharded* feature_table, ss::lowres_clock::duration run_interval, config::binding default_topic_replication, ss::scheduling_group scheduling_group) : _topic_creator(topic_creator) , _topic_metadata_cache(topic_metadata_cache) , _link_registry(link_registry) + , _feature_table(feature_table) , _run_interval(run_interval) , _default_topic_replication(std::move(default_topic_replication)) , _scheduling_group(scheduling_group) {} @@ -261,6 +264,24 @@ ss::future<> topic_reconciler::maybe_create_mirror_topic( _default_topic_replication()), topic_configs); + // This path bypasses the kafka create validators, so it must enforce + // the same upgrade gate: no tiered_v2 topic may come into existence + // while the cluster is not fully upgraded to v26.2. Skipping is safe - + // the reconciler retries and creates the mirror topic once the feature + // activates. + if ( + cfg.properties.storage_mode + == ::model::redpanda_storage_mode::tiered_cloud + && !_feature_table->local().is_active( + features::feature::tiered_cloud_topics)) { + vlog( + cllog.warn, + "Not creating mirror topic {}: cannot use the tiered_v2 storage " + "mode until the cluster is fully upgraded to at least v26.2.1", + topic); + co_return; + } + auto res = co_await _topic_creator->create_topic( {::model::kafka_namespace, topic}, mirror_topic_config.partition_count, @@ -302,8 +323,46 @@ topic_reconciler::maybe_create_update_mirror_topic( config_updated = true; } + // The storage mode is applied from the (mode, version) pair: the mode + // value alone is ambiguous because both tiered variants describe as + // 'tiered'. + auto find_config = [&mirror_topic_config]( + std::string_view name) -> std::optional { + auto it = mirror_topic_config.topic_configs.find(ss::sstring(name)); + if (it == mirror_topic_config.topic_configs.end()) { + return std::nullopt; + } + return it->second; + }; + try { + if ( + utils::maybe_append_storage_mode_update( + update, + find_config(kafka::topic_property_redpanda_storage_mode), + find_config(kafka::topic_property_redpanda_storage_mode_impl), + local_topic_config, + _feature_table->local())) { + config_updated = true; + } + } catch (const std::exception& e) { + vlog( + cllog.warn, + "Failed updating topic property {} for topic {}: {}", + kafka::topic_property_redpanda_storage_mode, + topic_name, + e); + } + for (const auto& [source_topic_config_name, source_topic_config_value] : mirror_topic_config.topic_configs) { + if ( + source_topic_config_name + == kafka::topic_property_redpanda_storage_mode + || source_topic_config_name + == kafka::topic_property_redpanda_storage_mode_impl) { + // Handled jointly above. + continue; + } // Need to check the current value of the local topic config // against that stored in the table, so I need to go from // string name to actual type diff --git a/src/v/cluster_link/topic_reconciler.h b/src/v/cluster_link/topic_reconciler.h index e04c39df19f98..aa6906865bd6f 100644 --- a/src/v/cluster_link/topic_reconciler.h +++ b/src/v/cluster_link/topic_reconciler.h @@ -12,9 +12,11 @@ #pragma once #include "cluster_link/deps.h" +#include "features/fwd.h" #include "kafka/data/rpc/deps.h" #include +#include namespace cluster_link { @@ -27,6 +29,7 @@ class topic_reconciler { kafka::data::rpc::topic_creator* topic_creator, kafka::data::rpc::topic_metadata_cache* topic_metadata_cache, link_registry* link_registry, + ss::sharded* feature_table, ss::lowres_clock::duration run_interval, config::binding default_topic_replication, ss::scheduling_group sg); @@ -75,6 +78,7 @@ class topic_reconciler { kafka::data::rpc::topic_creator* _topic_creator; kafka::data::rpc::topic_metadata_cache* _topic_metadata_cache; link_registry* _link_registry; + ss::sharded* _feature_table; ssx::mutex _reconciler_mutex{"cluster_link::task_reconciler"}; ss::timer _reconciler_timer; diff --git a/src/v/cluster_link/utils/BUILD b/src/v/cluster_link/utils/BUILD index ca28278c7f1da..62d1af68d9257 100644 --- a/src/v/cluster_link/utils/BUILD +++ b/src/v/cluster_link/utils/BUILD @@ -12,6 +12,7 @@ redpanda_cc_library( deps = [ "//src/v/base", "//src/v/cluster:types", + "//src/v/features", "//src/v/kafka/server:topic_config_utils", "//src/v/pandaproxy/schema_registry:types", "//src/v/reflection:type_traits", diff --git a/src/v/cluster_link/utils/topic_properties_utils.cc b/src/v/cluster_link/utils/topic_properties_utils.cc index 43d8733922607..47310899543b0 100644 --- a/src/v/cluster_link/utils/topic_properties_utils.cc +++ b/src/v/cluster_link/utils/topic_properties_utils.cc @@ -12,6 +12,7 @@ #include "cluster_link/utils/topic_properties_utils.h" #include "base/type_traits.h" +#include "features/feature_table.h" #include "kafka/server/handlers/configs/config_utils.h" #include "pandaproxy/schema_registry/types.h" #include "reflection/type_traits.h" @@ -333,14 +334,56 @@ bool maybe_append_update( kafka::max_compaction_lag_ms_validator); } - if (config_name == kafka::topic_property_redpanda_storage_mode) { - return parse_and_set( - topic_config.tp_ns, - update.properties.storage_mode, - config_value, - std::make_optional(topic_config.properties.storage_mode)); - } - return false; } + +bool maybe_append_storage_mode_update( + cluster::topic_properties_update& update, + const std::optional& mode_value, + const std::optional& impl_value, + const cluster::topic_configuration& topic_config, + const features::feature_table& feature_table) { + std::optional<::model::redpanda_storage_mode> mode; + if (impl_value.has_value()) { + // The impl property is exact and always present on sources that + // emit it, so it takes precedence over the ambiguous mode value. + mode = ::model::redpanda_storage_mode_from_impl_string(*impl_value); + if (!mode.has_value()) { + throw kafka::validation_error( + fmt::format("Unrecognized storage mode impl '{}'", *impl_value)); + } + } else if (mode_value.has_value()) { + // Sources that predate the impl property sync only the mode, where + // 'tiered' meant the classic tiered storage. + mode = ::model::redpanda_storage_mode_from_string(*mode_value); + if (!mode.has_value()) { + throw kafka::validation_error( + fmt::format("Unrecognized storage mode '{}'", *mode_value)); + } + } else { + return false; + } + auto current = topic_config.properties.storage_mode; + if (*mode == current) { + return false; + } + if ( + *mode == ::model::redpanda_storage_mode::tiered_cloud + && !feature_table.is_active(features::feature::tiered_cloud_topics)) { + throw kafka::validation_error( + "Cannot use the tiered_v2 storage mode until the cluster is fully " + "upgraded to at least v26.2.1"); + } + if (!kafka::is_storage_mode_transition_permitted(current, *mode)) { + throw kafka::validation_error( + fmt::format( + "Storage mode transition from {} to {} is not permitted", + current, + *mode)); + } + update.properties.storage_mode.op + = cluster::incremental_update_operation::set; + update.properties.storage_mode.value = *mode; + return true; +} } // namespace cluster_link::utils diff --git a/src/v/cluster_link/utils/topic_properties_utils.h b/src/v/cluster_link/utils/topic_properties_utils.h index f6e47ce36130d..b7693e2c956a3 100644 --- a/src/v/cluster_link/utils/topic_properties_utils.h +++ b/src/v/cluster_link/utils/topic_properties_utils.h @@ -11,6 +11,7 @@ #pragma once +#include "features/fwd.h" #include "kafka/server/handlers/configs/config_utils.h" namespace cluster_link::utils { @@ -21,4 +22,20 @@ bool maybe_append_update( const ss::sstring& config_value, const cluster::topic_configuration& topic_config); -} +/// Combine the source topic's redpanda.storage.mode and its read-only +/// redpanda.storage.mode.impl companion into a storage-mode update. The +/// impl value is exact and wins; the mode value alone is the fallback for +/// sources that predate the impl property (where 'tiered' meant the +/// classic tiered storage). Updates that are not permitted storage-mode +/// transitions are rejected with a validation error rather than applied, +/// and so is a move to the tiered_v2 impl while the tiered_cloud_topics +/// feature is not active (the cluster is not fully upgraded to v26.2) -- +/// this internal path must enforce the same gate as the kafka handlers. +bool maybe_append_storage_mode_update( + cluster::topic_properties_update& update, + const std::optional& mode_value, + const std::optional& impl_value, + const cluster::topic_configuration& topic_config, + const features::feature_table& feature_table); + +} // namespace cluster_link::utils diff --git a/src/v/config/configuration.cc b/src/v/config/configuration.cc index 78e74d8cc64ae..7fabbc8a6da31 100644 --- a/src/v/config/configuration.cc +++ b/src/v/config/configuration.cc @@ -2099,6 +2099,22 @@ configuration::configuration() model::redpanda_storage_mode::tiered, model::redpanda_storage_mode::cloud, model::redpanda_storage_mode::unset}) + , default_redpanda_storage_mode_tiered_impl( + *this, + "default_redpanda_storage_mode_tiered_impl", + "Default implementation of the `tiered` storage mode for " + "newly-created topics. When `redpanda.storage.mode` is set to " + "`tiered`, this property determines whether the topic uses the " + "classic tiered-storage architecture (`tiered_v1`) or the new " + "tiered-storage architecture (`tiered_v2`). The implementation of " + "each topic is reported by the read-only `redpanda.storage.mode.impl` " + "topic property.", + {.needs_restart = needs_restart::no, + .example = "tiered_v2", + .visibility = visibility::user}, + model::redpanda_storage_mode_tiered_impl::tiered_v1, + {model::redpanda_storage_mode_tiered_impl::tiered_v1, + model::redpanda_storage_mode_tiered_impl::tiered_v2}) , cloud_storage_disable_archiver_manager( *this, "cloud_storage_disable_archiver_manager", diff --git a/src/v/config/configuration.h b/src/v/config/configuration.h index a0be803c52df7..d885c1421d799 100644 --- a/src/v/config/configuration.h +++ b/src/v/config/configuration.h @@ -383,6 +383,8 @@ struct configuration final : public config_store { property cloud_storage_enable_remote_read; property cloud_storage_enable_remote_write; enum_property default_redpanda_storage_mode; + enum_property + default_redpanda_storage_mode_tiered_impl; property cloud_storage_disable_archiver_manager; property> cloud_storage_access_key; property> cloud_storage_secret_key; diff --git a/src/v/config/convert.h b/src/v/config/convert.h index dd492e0348552..7e9cb9a9ddff4 100644 --- a/src/v/config/convert.h +++ b/src/v/config/convert.h @@ -601,6 +601,23 @@ struct convert { } }; +template<> +struct convert { + using type = model::redpanda_storage_mode_tiered_impl; + + static Node encode(const type& rhs) { return Node(fmt::format("{}", rhs)); } + + static bool decode(const Node& node, type& rhs) { + auto value = node.as(); + auto mode = model::redpanda_storage_mode_tiered_impl_from_string(value); + if (!mode) { + return false; + } + rhs = mode.value(); + return true; + } +}; + template<> struct convert { using type = model::recovery_validation_mode; diff --git a/src/v/config/property.h b/src/v/config/property.h index 9ff9b27d6e32e..b861631d6f771 100644 --- a/src/v/config/property.h +++ b/src/v/config/property.h @@ -806,6 +806,9 @@ consteval std::string_view property_type_name() { return "string"; } else if constexpr (std::is_same_v) { return "string"; + } else if constexpr ( + std::is_same_v) { + return "string"; } else if constexpr (std::is_same_v) { return "string"; } else { diff --git a/src/v/config/rjson_serialization.cc b/src/v/config/rjson_serialization.cc index 6ecb4c81df394..96534e3808f33 100644 --- a/src/v/config/rjson_serialization.cc +++ b/src/v/config/rjson_serialization.cc @@ -342,4 +342,10 @@ void rjson_serialize( stringize(w, m); } +void rjson_serialize( + json::Writer& w, + const model::redpanda_storage_mode_tiered_impl& m) { + stringize(w, m); +} + } // namespace json diff --git a/src/v/config/rjson_serialization.h b/src/v/config/rjson_serialization.h index ec29ec33b0a32..def2504be8736 100644 --- a/src/v/config/rjson_serialization.h +++ b/src/v/config/rjson_serialization.h @@ -177,4 +177,8 @@ void rjson_serialize( void rjson_serialize( json::Writer&, const model::redpanda_storage_mode&); +void rjson_serialize( + json::Writer&, + const model::redpanda_storage_mode_tiered_impl&); + } // namespace json diff --git a/src/v/features/feature_table.h b/src/v/features/feature_table.h index 5a8a3c07a302a..1128d0f4aeeae 100644 --- a/src/v/features/feature_table.h +++ b/src/v/features/feature_table.h @@ -561,7 +561,7 @@ inline constexpr std::array feature_schema{ release_version::v26_2_1, "tiered_cloud_topics", feature::tiered_cloud_topics, - feature_spec::available_policy::explicit_only, + feature_spec::available_policy::always, feature_spec::prepare_policy::always}, feature_spec{ release_version::v26_2_1, diff --git a/src/v/kafka/protocol/topic_properties.h b/src/v/kafka/protocol/topic_properties.h index 6952abdd7ec92..4fa2a4b3857a1 100644 --- a/src/v/kafka/protocol/topic_properties.h +++ b/src/v/kafka/protocol/topic_properties.h @@ -79,4 +79,11 @@ inline constexpr std::string_view topic_property_leaders_preference inline constexpr std::string_view topic_property_redpanda_storage_mode = "redpanda.storage.mode"; +// Read-only companion of redpanda.storage.mode: the resolved variant +// (tiered_v1/tiered_v2) of a topic in the tiered storage mode. Accepted on +// topic creation (only together with redpanda.storage.mode=tiered) to pick a +// non-default variant; not alterable. +inline constexpr std::string_view topic_property_redpanda_storage_mode_impl + = "redpanda.storage.mode.impl"; + } // namespace kafka diff --git a/src/v/kafka/server/handlers/alter_configs.cc b/src/v/kafka/server/handlers/alter_configs.cc index 397199b657770..9fe8acfae3e36 100644 --- a/src/v/kafka/server/handlers/alter_configs.cc +++ b/src/v/kafka/server/handlers/alter_configs.cc @@ -542,6 +542,23 @@ create_topic_properties_update( /*clamp_to_duration_max=*/true); continue; } + if (cfg.name == topic_property_redpanda_storage_mode_impl) { + // Read-only, but tolerate idempotent sets: describe-then- + // alter round trips (e.g. kafka-configs.sh) replay every + // config, including this one. + if ( + cfg.value.has_value() && current_storage_mode.has_value() + && cfg.value.value() + == model::redpanda_storage_mode_impl_name( + *current_storage_mode)) { + continue; + } + throw validation_error( + "redpanda.storage.mode.impl is read-only and can only be " + "set on topic creation; to change a topic's storage mode, " + "alter redpanda.storage.mode instead"); + } + if (cfg.name == topic_property_redpanda_storage_mode) { auto validator = [current_storage_mode, &feature_table = ctx.feature_table().local()]( @@ -557,16 +574,28 @@ create_topic_properties_update( value == model::redpanda_storage_mode::tiered_cloud && !feature_table.is_active( features::feature::tiered_cloud_topics)) { - return "tiered_cloud storage mode requires the " - "tiered_cloud_topics feature to be enabled"; + return "Cannot use the tiered_v2 storage mode " + "until the cluster is fully upgraded to at " + "least v26.2.1"; } return std::nullopt; }; + auto parse = [](const ss::sstring& raw) { + auto mode = model::redpanda_storage_mode_from_user_string( + raw, + config::shard_local_cfg() + .default_redpanda_storage_mode_tiered_impl()); + if (!mode) { + throw boost::bad_lexical_cast(); + } + return *mode; + }; parse_and_set_optional( update.properties.storage_mode, cfg.value, kafka::config_resource_operation::set, - validator); + validator, + parse); continue; } diff --git a/src/v/kafka/server/handlers/configs/config_response_utils.cc b/src/v/kafka/server/handlers/configs/config_response_utils.cc index 972fc77a01e05..2ebfff6ce17d6 100644 --- a/src/v/kafka/server/handlers/configs/config_response_utils.cc +++ b/src/v/kafka/server/handlers/configs/config_response_utils.cc @@ -1218,7 +1218,41 @@ config_response_container_t make_topic_configs( maybe_make_documentation( include_documentation, config::shard_local_cfg().default_redpanda_storage_mode.desc()), - &describe_as_string); + [](const model::redpanda_storage_mode& mode) { + return ss::sstring(model::redpanda_storage_mode_user_name(mode)); + }); + + // Read-only companion of redpanda.storage.mode: the exact + // implementation of the topic's storage mode, always present and never + // ambiguous (the tiered variants report tiered_v1/tiered_v2 while the + // mode itself displays both as 'tiered'). + if ( + config_property_requested( + config_keys, topic_property_redpanda_storage_mode_impl)) { + result.push_back( + config_response{ + .name = ss::sstring(topic_property_redpanda_storage_mode_impl), + .value = ss::sstring( + model::redpanda_storage_mode_impl_name( + topic_properties.storage_mode)), + .read_only = true, + // The value is derived from the storage mode rather than being + // an explicit override: report it as a default so config + // backup/replay tooling does not treat it as user-set (and so + // upgrade config comparisons tolerate its appearance). + .is_default = true, + .config_source = describe_configs_source::default_config, + .config_type = describe_configs_type::string, + .documentation = maybe_make_documentation( + include_documentation, + "Exact implementation of the topic's storage mode. Tiered " + "topics report tiered_v1 (classic tiered-storage " + "architecture) or tiered_v2 (new tiered-storage " + "architecture); other modes mirror redpanda.storage.mode. " + "Read-only after creation: supply it at topic creation to " + "select the implementation explicitly."), + }); + } return result; } diff --git a/src/v/kafka/server/handlers/configs/config_utils.h b/src/v/kafka/server/handlers/configs/config_utils.h index a10f92b4458e8..37a8e085b17d1 100644 --- a/src/v/kafka/server/handlers/configs/config_utils.h +++ b/src/v/kafka/server/handlers/configs/config_utils.h @@ -447,8 +447,8 @@ struct storage_mode_validator { return fmt::format( "Cannot alter redpanda.storage.mode from {} to {} - this " "transition is not permitted", - *current_mode, - value); + model::redpanda_storage_mode_impl_name(*current_mode), + model::redpanda_storage_mode_impl_name(value)); } return std::nullopt; } diff --git a/src/v/kafka/server/handlers/create_topics.cc b/src/v/kafka/server/handlers/create_topics.cc index cde31cbc1ea3d..f1fa8ea80aba0 100644 --- a/src/v/kafka/server/handlers/create_topics.cc +++ b/src/v/kafka/server/handlers/create_topics.cc @@ -86,7 +86,8 @@ bool is_supported(std::string_view name) { topic_property_remote_allow_gaps, topic_property_message_timestamp_before_max_ms, topic_property_message_timestamp_after_max_ms, - topic_property_redpanda_storage_mode}); + topic_property_redpanda_storage_mode, + topic_property_redpanda_storage_mode_impl}); if ( std::any_of( diff --git a/src/v/kafka/server/handlers/incremental_alter_configs.cc b/src/v/kafka/server/handlers/incremental_alter_configs.cc index 16e8d25a72554..3faa6eae07649 100644 --- a/src/v/kafka/server/handlers/incremental_alter_configs.cc +++ b/src/v/kafka/server/handlers/incremental_alter_configs.cc @@ -512,6 +512,24 @@ create_topic_properties_update( continue; } + if (cfg.name == topic_property_redpanda_storage_mode_impl) { + // Read-only, but tolerate idempotent sets: describe-then- + // alter round trips (e.g. kafka-configs.sh) replay every + // config, including this one. + if ( + op == config_resource_operation::set && cfg.value.has_value() + && current_storage_mode.has_value() + && cfg.value.value() + == model::redpanda_storage_mode_impl_name( + *current_storage_mode)) { + continue; + } + throw validation_error( + "redpanda.storage.mode.impl is read-only and can only be " + "set on topic creation; to change a topic's storage mode, " + "alter redpanda.storage.mode instead"); + } + if (cfg.name == topic_property_redpanda_storage_mode) { auto validator = [current_storage_mode, &feature_table = ctx.feature_table().local()]( @@ -527,13 +545,28 @@ create_topic_properties_update( value == model::redpanda_storage_mode::tiered_cloud && !feature_table.is_active( features::feature::tiered_cloud_topics)) { - return "tiered_cloud storage mode requires the " - "tiered_cloud_topics feature to be enabled"; + return "Cannot use the tiered_v2 storage mode " + "until the cluster is fully upgraded to at " + "least v26.2.1"; } return std::nullopt; }; + auto parse = [](const ss::sstring& raw) { + auto mode = model::redpanda_storage_mode_from_user_string( + raw, + config::shard_local_cfg() + .default_redpanda_storage_mode_tiered_impl()); + if (!mode) { + throw boost::bad_lexical_cast(); + } + return *mode; + }; parse_and_set_optional( - update.properties.storage_mode, cfg.value, op, validator); + update.properties.storage_mode, + cfg.value, + op, + validator, + parse); continue; } } catch (const validation_error& e) { diff --git a/src/v/kafka/server/handlers/topics/types.cc b/src/v/kafka/server/handlers/topics/types.cc index ec7960efde772..4fa511e23e8b0 100644 --- a/src/v/kafka/server/handlers/topics/types.cc +++ b/src/v/kafka/server/handlers/topics/types.cc @@ -353,9 +353,25 @@ cluster::topic_configuration to_topic_config( topic_property_message_timestamp_after_max_ms, /*clamp_to_duration_max=*/true); + // The exact implementation (redpanda.storage.mode.impl) wins over the + // alias-resolved mode. Malformed combinations are rejected by + // storage_mode_config_validator before conversion. cfg.properties.storage_mode - = get_enum_value( - config_entries, topic_property_redpanda_storage_mode) + = get_string_value( + config_entries, topic_property_redpanda_storage_mode_impl) + .and_then([](const ss::sstring& raw) { + return model::redpanda_storage_mode_from_impl_string(raw); + }) + .or_else([&config_entries]() { + return get_string_value( + config_entries, topic_property_redpanda_storage_mode) + .and_then([](const ss::sstring& raw) { + return model::redpanda_storage_mode_from_user_string( + raw, + config::shard_local_cfg() + .default_redpanda_storage_mode_tiered_impl()); + }); + }) .value_or(config::shard_local_cfg().default_redpanda_storage_mode()); schema_id_validation_config_parser schema_id_validation_config_parser{ diff --git a/src/v/kafka/server/handlers/topics/validators.h b/src/v/kafka/server/handlers/topics/validators.h index f22e14353875f..147a96ad32aca 100644 --- a/src/v/kafka/server/handlers/topics/validators.h +++ b/src/v/kafka/server/handlers/topics/validators.h @@ -549,30 +549,69 @@ struct min_max_compaction_lag_ms_validator { * - 'cloud' mode requires cloud_storage_enabled() * - 'tiered' mode requires cloud_storage_enabled() * - 'local' mode is always allowed + * The optional redpanda.storage.mode.impl property picks the tiered + * variant and is only valid together with redpanda.storage.mode=tiered. */ struct storage_mode_config_validator { static constexpr const char* error_message - = "Invalid storage mode: 'cloud' requires cloud storage to be enabled " - "and the cluster to be fully upgraded to at least v26.1.1, " - "'tiered_cloud' additionally requires the tiered_cloud_topics feature " - "to be explicitly enabled, " - "'tiered' requires cloud storage to be enabled."; + = "Invalid storage mode: redpanda.storage.mode accepts local, tiered, " + "cloud or unset; redpanda.storage.mode.impl accepts local, " + "tiered_v1, tiered_v2, cloud or unset and must agree with the mode " + "when both are set. Tiered and cloud modes require cloud storage, " + "'cloud' requires at least v26.1.1 and 'tiered_v2' at least " + "v26.2.1."; static constexpr error_code ec = error_code::invalid_config; static bool is_valid(const creatable_topic& c, features::feature_table* ft) { - auto it = std::find_if( - c.configs.begin(), - c.configs.end(), - [](const createable_topic_config& cfg) { - return cfg.name == topic_property_redpanda_storage_mode; - }); - if (it == c.configs.end() || !it->value.has_value()) { - return true; + auto find_cfg = [&c](std::string_view name) { + auto it = std::find_if( + c.configs.begin(), + c.configs.end(), + [name](const createable_topic_config& cfg) { + return cfg.name == name; + }); + return it == c.configs.end() ? nullptr : &*it; + }; + const auto* mode_cfg = find_cfg(topic_property_redpanda_storage_mode); + const auto* impl_cfg = find_cfg( + topic_property_redpanda_storage_mode_impl); + + std::optional impl; + if (impl_cfg != nullptr && impl_cfg->value.has_value()) { + impl = model::redpanda_storage_mode_from_impl_string( + impl_cfg->value.value()); + if (!impl) { + return false; + } } - auto mode = model::redpanda_storage_mode_from_string(it->value.value()); - if (!mode) { - return false; + std::optional mode; + if (mode_cfg != nullptr && mode_cfg->value.has_value()) { + mode = model::redpanda_storage_mode_from_user_string( + mode_cfg->value.value(), + config::shard_local_cfg() + .default_redpanda_storage_mode_tiered_impl()); + if (!mode) { + return false; + } + } + if (mode.has_value() && impl.has_value()) { + // The mode is the user-facing name of the implementation: + // when both are given they must agree ('tiered' matching both + // tiered variants). + if ( + mode_cfg->value.value() + != model::redpanda_storage_mode_user_name(*impl)) { + return false; + } + } + // The implementation is exact, so it wins over the alias-resolved + // mode. + if (impl.has_value()) { + mode = impl; + } + if (!mode.has_value()) { + return true; } switch (*mode) { case model::redpanda_storage_mode::local: diff --git a/src/v/kafka/server/tests/alter_config_test.cc b/src/v/kafka/server/tests/alter_config_test.cc index da6284c3cd170..d1656037bf8ae 100644 --- a/src/v/kafka/server/tests/alter_config_test.cc +++ b/src/v/kafka/server/tests/alter_config_test.cc @@ -794,7 +794,8 @@ FIXTURE_TEST( "max.compaction.lag.ms", "message.timestamp.before.max.ms", "message.timestamp.after.max.ms", - "redpanda.storage.mode"}; + "redpanda.storage.mode", + "redpanda.storage.mode.impl"}; // All properties_request auto all_describe_resp = describe_configs(test_tp); diff --git a/src/v/model/metadata.h b/src/v/model/metadata.h index 171f80fa0c18a..d5f23405259a7 100644 --- a/src/v/model/metadata.h +++ b/src/v/model/metadata.h @@ -646,6 +646,74 @@ std::optional fmt::iterator format_to(redpanda_storage_mode m, fmt::iterator out); std::istream& operator>>(std::istream&, redpanda_storage_mode&); +// Selects which storage mode the plain 'tiered' user-facing alias refers to: +// 'tiered_v1' is redpanda_storage_mode::tiered, 'tiered_v2' is +// redpanda_storage_mode::tiered_cloud. Value of the +// redpanda_storage_mode_tiered_impl cluster config. +enum class redpanda_storage_mode_tiered_impl : uint8_t { + tiered_v1 = 0, + tiered_v2 = 1, +}; + +constexpr const char* redpanda_storage_mode_tiered_impl_to_string( + redpanda_storage_mode_tiered_impl m) { + switch (m) { + case redpanda_storage_mode_tiered_impl::tiered_v1: + return "tiered_v1"; + case redpanda_storage_mode_tiered_impl::tiered_v2: + return "tiered_v2"; + } + throw std::invalid_argument("unknown redpanda_storage_mode_tiered_impl"); +} + +std::optional + redpanda_storage_mode_tiered_impl_from_string(std::string_view); + +fmt::iterator format_to(redpanda_storage_mode_tiered_impl m, fmt::iterator out); +std::istream& operator>>(std::istream&, redpanda_storage_mode_tiered_impl&); + +/// Parse a user-supplied storage mode string (the redpanda.storage.mode topic +/// property). The user vocabulary is local/tiered/cloud/unset: 'tiered' +/// resolves against the redpanda_storage_mode_tiered_impl cluster config, and +/// the variant names (tiered_v1/tiered_v2) as well as the internal +/// 'tiered_cloud' spelling are rejected -- a specific variant is selected with +/// the separate redpanda.storage.mode.impl property instead. +std::optional redpanda_storage_mode_from_user_string( + std::string_view, redpanda_storage_mode_tiered_impl); + +/// The user-facing name of a storage mode: both tiered variants display as +/// 'tiered' (the variant is exposed via redpanda.storage.mode.impl); +/// other modes display as their enum name. +const char* redpanda_storage_mode_user_name(redpanda_storage_mode); + +/// The tiered variant of a storage mode: tiered -> tiered_v1, +/// tiered_cloud -> tiered_v2, nullopt for the other modes. +std::optional + storage_mode_tiered_impl(redpanda_storage_mode); + +/// The unambiguous implementation name of a storage mode, the value of the +/// read-only redpanda.storage.mode.impl topic property: +/// unset/local/cloud display as their enum name, the tiered variants as +/// tiered_v1/tiered_v2. +const char* redpanda_storage_mode_impl_name(redpanda_storage_mode); + +/// Parse the redpanda.storage.mode.impl vocabulary +/// (unset|local|tiered_v1|tiered_v2|cloud): the exact inverse of +/// redpanda_storage_mode_impl_name. The ambiguous 'tiered' alias and the +/// internal 'tiered_cloud' spelling are rejected. +std::optional + redpanda_storage_mode_from_impl_string(std::string_view); + +/// Combine the redpanda.storage.mode value with an explicit +/// redpanda.storage.mode.impl into the storage mode enum: the version +/// picks the tiered variant. +constexpr redpanda_storage_mode +storage_mode_with_tiered_impl(redpanda_storage_mode_tiered_impl version) { + return version == redpanda_storage_mode_tiered_impl::tiered_v2 + ? redpanda_storage_mode::tiered_cloud + : redpanda_storage_mode::tiered; +} + enum class recovery_validation_mode : std::uint16_t { // ensure that either the manifest is in TS or that no manifest is present. // download issues will fail the validation diff --git a/src/v/model/model.cc b/src/v/model/model.cc index 3555ead5b7f63..32877a06793b3 100644 --- a/src/v/model/model.cc +++ b/src/v/model/model.cc @@ -585,9 +585,136 @@ redpanda_storage_mode_from_string(std::string_view s) { model::redpanda_storage_mode_to_string( model::redpanda_storage_mode::unset), model::redpanda_storage_mode::unset) + .match( + model::redpanda_storage_mode_tiered_impl_to_string( + model::redpanda_storage_mode_tiered_impl::tiered_v1), + model::redpanda_storage_mode::tiered) + .match( + model::redpanda_storage_mode_tiered_impl_to_string( + model::redpanda_storage_mode_tiered_impl::tiered_v2), + model::redpanda_storage_mode::tiered_cloud) + .default_match(std::nullopt); +} + +fmt::iterator +format_to(redpanda_storage_mode_tiered_impl mode, fmt::iterator out) { + return fmt::format_to( + out, "{}", redpanda_storage_mode_tiered_impl_to_string(mode)); +} + +std::istream& +operator>>(std::istream& i, redpanda_storage_mode_tiered_impl& mode) { + ss::sstring s; + i >> s; + auto value = redpanda_storage_mode_tiered_impl_from_string(s); + if (!value) { + i.setstate(std::ios::failbit); + return i; + } + mode = *value; + return i; +} + +std::optional +redpanda_storage_mode_tiered_impl_from_string(std::string_view s) { + return string_switch>(s) + .match( + redpanda_storage_mode_tiered_impl_to_string( + redpanda_storage_mode_tiered_impl::tiered_v1), + redpanda_storage_mode_tiered_impl::tiered_v1) + .match( + redpanda_storage_mode_tiered_impl_to_string( + redpanda_storage_mode_tiered_impl::tiered_v2), + redpanda_storage_mode_tiered_impl::tiered_v2) .default_match(std::nullopt); } +std::optional redpanda_storage_mode_from_user_string( + std::string_view s, redpanda_storage_mode_tiered_impl default_mode) { + if (s == redpanda_storage_mode_to_string(redpanda_storage_mode::tiered)) { + return storage_mode_with_tiered_impl(default_mode); + } + return string_switch>(s) + .match( + redpanda_storage_mode_to_string(redpanda_storage_mode::local), + redpanda_storage_mode::local) + .match( + redpanda_storage_mode_to_string(redpanda_storage_mode::cloud), + redpanda_storage_mode::cloud) + .match( + redpanda_storage_mode_to_string(redpanda_storage_mode::unset), + redpanda_storage_mode::unset) + // The variant names and the internal 'tiered_cloud' spelling are not + // valid mode values: variants are selected with the separate + // redpanda.storage.mode.impl property. + .default_match(std::nullopt); +} + +const char* redpanda_storage_mode_user_name(redpanda_storage_mode mode) { + switch (mode) { + case redpanda_storage_mode::tiered: + case redpanda_storage_mode::tiered_cloud: + return redpanda_storage_mode_to_string(redpanda_storage_mode::tiered); + case redpanda_storage_mode::local: + case redpanda_storage_mode::cloud: + case redpanda_storage_mode::unset: + return redpanda_storage_mode_to_string(mode); + } + throw std::invalid_argument("unknown redpanda_storage_mode"); +} + +const char* redpanda_storage_mode_impl_name(redpanda_storage_mode mode) { + switch (mode) { + case redpanda_storage_mode::tiered: + return redpanda_storage_mode_tiered_impl_to_string( + redpanda_storage_mode_tiered_impl::tiered_v1); + case redpanda_storage_mode::tiered_cloud: + return redpanda_storage_mode_tiered_impl_to_string( + redpanda_storage_mode_tiered_impl::tiered_v2); + case redpanda_storage_mode::local: + case redpanda_storage_mode::cloud: + case redpanda_storage_mode::unset: + return redpanda_storage_mode_to_string(mode); + } + throw std::invalid_argument("unknown redpanda_storage_mode"); +} + +std::optional +redpanda_storage_mode_from_impl_string(std::string_view s) { + return string_switch>(s) + .match( + redpanda_storage_mode_impl_name(redpanda_storage_mode::local), + redpanda_storage_mode::local) + .match( + redpanda_storage_mode_impl_name(redpanda_storage_mode::tiered), + redpanda_storage_mode::tiered) + .match( + redpanda_storage_mode_impl_name(redpanda_storage_mode::tiered_cloud), + redpanda_storage_mode::tiered_cloud) + .match( + redpanda_storage_mode_impl_name(redpanda_storage_mode::cloud), + redpanda_storage_mode::cloud) + .match( + redpanda_storage_mode_impl_name(redpanda_storage_mode::unset), + redpanda_storage_mode::unset) + .default_match(std::nullopt); +} + +std::optional +storage_mode_tiered_impl(redpanda_storage_mode mode) { + switch (mode) { + case redpanda_storage_mode::tiered: + return redpanda_storage_mode_tiered_impl::tiered_v1; + case redpanda_storage_mode::tiered_cloud: + return redpanda_storage_mode_tiered_impl::tiered_v2; + case redpanda_storage_mode::local: + case redpanda_storage_mode::cloud: + case redpanda_storage_mode::unset: + return std::nullopt; + } + throw std::invalid_argument("unknown redpanda_storage_mode"); +} + fmt::iterator format_to(recovery_validation_mode vm, fmt::iterator out) { using enum recovery_validation_mode; switch (vm) { diff --git a/src/v/model/tests/BUILD b/src/v/model/tests/BUILD index 76720a77ae18e..66cd0fbdc51c1 100644 --- a/src/v/model/tests/BUILD +++ b/src/v/model/tests/BUILD @@ -258,6 +258,20 @@ redpanda_cc_gtest( ], ) +redpanda_cc_gtest( + name = "storage_mode_alias_test", + timeout = "short", + srcs = [ + "storage_mode_alias_test.cc", + ], + cpu = 1, + deps = [ + "//src/v/model", + "//src/v/test_utils:gtest", + "@googletest//:gtest", + ], +) + redpanda_cc_gtest( name = "record_batch_reader_gtest", timeout = "short", diff --git a/src/v/model/tests/storage_mode_alias_test.cc b/src/v/model/tests/storage_mode_alias_test.cc new file mode 100644 index 0000000000000..a18fee610121d --- /dev/null +++ b/src/v/model/tests/storage_mode_alias_test.cc @@ -0,0 +1,161 @@ +/* + * Copyright 2026 Redpanda Data, Inc. + * + * Use of this software is governed by the Business Source License + * included in the file licenses/BSL.md + * + * As of the Change Date specified in that file, in accordance with + * the Business Source License, use of this software will be governed + * by the Apache License, Version 2.0 + */ + +#include "model/metadata.h" + +#include + +using model::redpanda_storage_mode; +using model::redpanda_storage_mode_tiered_impl; + +// Full matrix: user input string x redpanda_storage_mode_tiered_impl -> parsed +// enum (nullopt = rejected). Only local/tiered/cloud/unset are valid mode +// values; the variant names and the internal 'tiered_cloud' spelling are +// rejected -- a variant is selected with redpanda.storage.mode.impl instead. +TEST(storage_mode_alias, from_user_string_matrix) { + struct { + std::string_view input; + std::optional under_v1; + std::optional under_v2; + } cases[] = { + {"local", redpanda_storage_mode::local, redpanda_storage_mode::local}, + {"tiered", + redpanda_storage_mode::tiered, + redpanda_storage_mode::tiered_cloud}, + {"cloud", redpanda_storage_mode::cloud, redpanda_storage_mode::cloud}, + {"unset", redpanda_storage_mode::unset, redpanda_storage_mode::unset}, + {"tiered_v1", std::nullopt, std::nullopt}, + {"tiered_v2", std::nullopt, std::nullopt}, + {"tiered_cloud", std::nullopt, std::nullopt}, + {"bogus", std::nullopt, std::nullopt}, + }; + for (const auto& c : cases) { + EXPECT_EQ( + model::redpanda_storage_mode_from_user_string( + c.input, redpanda_storage_mode_tiered_impl::tiered_v1), + c.under_v1) + << c.input << " under tiered_v1"; + EXPECT_EQ( + model::redpanda_storage_mode_from_user_string( + c.input, redpanda_storage_mode_tiered_impl::tiered_v2), + c.under_v2) + << c.input << " under tiered_v2"; + } +} + +// Both tiered variants display as 'tiered'; the variant is exposed through +// redpanda.storage.mode.impl (storage_mode_tiered_impl). +TEST(storage_mode_alias, user_name) { + EXPECT_STREQ( + model::redpanda_storage_mode_user_name(redpanda_storage_mode::tiered), + "tiered"); + EXPECT_STREQ( + model::redpanda_storage_mode_user_name( + redpanda_storage_mode::tiered_cloud), + "tiered"); + EXPECT_STREQ( + model::redpanda_storage_mode_user_name(redpanda_storage_mode::local), + "local"); + EXPECT_STREQ( + model::redpanda_storage_mode_user_name(redpanda_storage_mode::cloud), + "cloud"); + EXPECT_STREQ( + model::redpanda_storage_mode_user_name(redpanda_storage_mode::unset), + "unset"); +} + +TEST(storage_mode_alias, storage_mode_tiered_impl) { + EXPECT_EQ( + model::storage_mode_tiered_impl(redpanda_storage_mode::tiered), + redpanda_storage_mode_tiered_impl::tiered_v1); + EXPECT_EQ( + model::storage_mode_tiered_impl(redpanda_storage_mode::tiered_cloud), + redpanda_storage_mode_tiered_impl::tiered_v2); + EXPECT_EQ( + model::storage_mode_tiered_impl(redpanda_storage_mode::local), + std::nullopt); + EXPECT_EQ( + model::storage_mode_tiered_impl(redpanda_storage_mode::cloud), + std::nullopt); + EXPECT_EQ( + model::storage_mode_tiered_impl(redpanda_storage_mode::unset), + std::nullopt); +} + +TEST(storage_mode_alias, storage_mode_with_tiered_impl) { + EXPECT_EQ( + model::storage_mode_with_tiered_impl( + redpanda_storage_mode_tiered_impl::tiered_v1), + redpanda_storage_mode::tiered); + EXPECT_EQ( + model::storage_mode_with_tiered_impl( + redpanda_storage_mode_tiered_impl::tiered_v2), + redpanda_storage_mode::tiered_cloud); +} + +// The context-free parser keeps the static aliases and the internal spelling +// (used for cluster-config round-trips and the shadow-link sync fallback). +TEST(storage_mode_alias, from_string_static_aliases) { + EXPECT_EQ( + model::redpanda_storage_mode_from_string("tiered_v1"), + redpanda_storage_mode::tiered); + EXPECT_EQ( + model::redpanda_storage_mode_from_string("tiered_v2"), + redpanda_storage_mode::tiered_cloud); + EXPECT_EQ( + model::redpanda_storage_mode_from_string("tiered_cloud"), + redpanda_storage_mode::tiered_cloud); + EXPECT_EQ( + model::redpanda_storage_mode_from_string("tiered"), + redpanda_storage_mode::tiered); +} + +// The impl name is the unambiguous spelling of every storage mode, used by +// the read-only redpanda.storage.mode.impl property. +TEST(storage_mode_alias, impl_name_round_trip) { + struct { + redpanda_storage_mode mode; + std::string_view name; + } cases[] = { + {redpanda_storage_mode::local, "local"}, + {redpanda_storage_mode::tiered, "tiered_v1"}, + {redpanda_storage_mode::tiered_cloud, "tiered_v2"}, + {redpanda_storage_mode::cloud, "cloud"}, + {redpanda_storage_mode::unset, "unset"}, + }; + for (const auto& c : cases) { + EXPECT_EQ(model::redpanda_storage_mode_impl_name(c.mode), c.name); + EXPECT_EQ( + model::redpanda_storage_mode_from_impl_string(c.name), c.mode); + } + // Ambiguous / internal spellings are not part of the impl vocabulary. + EXPECT_EQ( + model::redpanda_storage_mode_from_impl_string("tiered"), std::nullopt); + EXPECT_EQ( + model::redpanda_storage_mode_from_impl_string("tiered_cloud"), + std::nullopt); + EXPECT_EQ( + model::redpanda_storage_mode_from_impl_string("bogus"), std::nullopt); +} + +TEST(storage_mode_alias, redpanda_storage_mode_tiered_impl_round_trip) { + for (auto m : + {redpanda_storage_mode_tiered_impl::tiered_v1, + redpanda_storage_mode_tiered_impl::tiered_v2}) { + EXPECT_EQ( + model::redpanda_storage_mode_tiered_impl_from_string( + model::redpanda_storage_mode_tiered_impl_to_string(m)), + m); + } + EXPECT_EQ( + model::redpanda_storage_mode_tiered_impl_from_string("tiered"), + std::nullopt); +} diff --git a/tests/rptest/clients/types.py b/tests/rptest/clients/types.py index 63955b2f19033..ba1d6c48274d8 100644 --- a/tests/rptest/clients/types.py +++ b/tests/rptest/clients/types.py @@ -27,9 +27,16 @@ class TopicSpec: STORAGE_MODE_LOCAL = "local" STORAGE_MODE_TIERED = "tiered" STORAGE_MODE_CLOUD = "cloud" - STORAGE_MODE_TIERED_CLOUD = "tiered_cloud" STORAGE_MODE_UNSET = "unset" + # Tiered values of the read-only redpanda.storage.mode.impl property, + # whose vocabulary is the unambiguous + # [unset|local|tiered_v1|tiered_v2|cloud]. On creation the property (set + # alone or together with a matching redpanda.storage.mode) selects the + # storage mode exactly. Not valid as redpanda.storage.mode values. + STORAGE_MODE_IMPL_TIERED_V1 = "tiered_v1" + STORAGE_MODE_IMPL_TIERED_V2 = "tiered_v2" + PROPERTY_COMPRESSSION = "compression.type" PROPERTY_CLEANUP_POLICY = "cleanup.policy" PROPERTY_COMPACTION_STRATEGY = "compaction.strategy" @@ -57,6 +64,24 @@ class TopicSpec: PROPERTY_REMOTE_READ = "redpanda.remote.read" PROPERTY_REMOTE_WRITE = "redpanda.remote.write" PROPERTY_STORAGE_MODE = "redpanda.storage.mode" + PROPERTY_STORAGE_MODE_IMPL = "redpanda.storage.mode.impl" + + @staticmethod + def storage_mode_config(mode: str) -> dict[str, str]: + """Topic config dict selecting a storage mode. Tiered variants + (tiered_v1/tiered_v2) are spelled as redpanda.storage.mode=tiered plus + the redpanda.storage.mode.impl property; sending both keeps v26.1 + brokers (which ignore the unknown impl property) on a tiered mode + rather than silently falling back to the cluster default.""" + if mode in ( + TopicSpec.STORAGE_MODE_IMPL_TIERED_V1, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, + ): + return { + TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_TIERED, + TopicSpec.PROPERTY_STORAGE_MODE_IMPL: mode, + } + return {TopicSpec.PROPERTY_STORAGE_MODE: mode} class CompressionTypes(str, Enum): """ @@ -153,6 +178,7 @@ def __init__( min_compaction_lag_ms: int | None = None, max_compaction_lag_ms: int | None = None, redpanda_storage_mode: str | None = None, + redpanda_storage_mode_impl: str | None = None, ): self.name = name or f"topic-{self._random_topic_suffix()}" self.partition_count = partition_count @@ -192,6 +218,10 @@ def __init__( self.min_compaction_lag_ms = min_compaction_lag_ms self.max_compaction_lag_ms = max_compaction_lag_ms self.redpanda_storage_mode = redpanda_storage_mode + # Read-only on the broker (reported by DescribeConfigs for every + # topic); accepted here so describe output can round-trip through + # TopicSpec. + self.redpanda_storage_mode_impl = redpanda_storage_mode_impl def __str__(self): return self.name diff --git a/tests/rptest/services/redpanda.py b/tests/rptest/services/redpanda.py index d7fd9dd4894ab..26a476eae84ed 100644 --- a/tests/rptest/services/redpanda.py +++ b/tests/rptest/services/redpanda.py @@ -4368,8 +4368,12 @@ def set_feature_active( cur_state = self.get_feature_state(feature_name) if active and cur_state == "unavailable": # If we have just restarted after an upgrade, wait for cluster version - # to progress and for the feature to become available. - self.await_feature(feature_name, "available", timeout_sec=timeout_sec) + # to progress and for the feature to become available. Features with + # available_policy::always auto-activate as soon as they become + # available, so accept "active" too. + self.await_feature( + feature_name, {"available", "active"}, timeout_sec=timeout_sec + ) self._admin.put_feature(feature_name, {"state": target_state}) self.await_feature(feature_name, target_state, timeout_sec=timeout_sec) @@ -4385,7 +4389,7 @@ def get_feature_state(self, feature_name: str, node: ClusterNode | None = None): def await_feature( self, feature_name: str, - await_state: str, + await_state: str | set[str], *, timeout_sec: int, nodes: list[ClusterNode] | None = None, @@ -4397,16 +4401,18 @@ def await_feature( if nodes is None: nodes = self.started_nodes() + await_states = {await_state} if isinstance(await_state, str) else await_state + def is_awaited_state(): for n in nodes: state = self.get_feature_state(feature_name, node=n) - if state != await_state: + if state not in await_states: self.logger.info( - f"Feature {feature_name} not yet {await_state} on {n.name} (state {state})" + f"Feature {feature_name} not yet in {await_states} on {n.name} (state {state})" ) return False - self.logger.info(f"Feature {feature_name} is now {await_state}") + self.logger.info(f"Feature {feature_name} is now in {await_states}") return True wait_until(is_awaited_state, timeout_sec=timeout_sec, backoff_sec=1) diff --git a/tests/rptest/tests/cloud_topics/compaction_stress_test.py b/tests/rptest/tests/cloud_topics/compaction_stress_test.py index b4af797178110..4d04b2e848564 100644 --- a/tests/rptest/tests/cloud_topics/compaction_stress_test.py +++ b/tests/rptest/tests/cloud_topics/compaction_stress_test.py @@ -185,7 +185,7 @@ def setUp(self): storage_mode = (self.test_context.injected_args or {}).get( "storage_mode", TopicSpec.STORAGE_MODE_CLOUD ) - if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD: + if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2: self.redpanda.set_feature_active( "tiered_cloud_topics", True, timeout_sec=30 ) @@ -194,7 +194,7 @@ def setUp(self): partitions=1, replicas=3, config={ - TopicSpec.PROPERTY_STORAGE_MODE: storage_mode, + **TopicSpec.storage_mode_config(storage_mode), "cleanup.policy": TopicSpec.CLEANUP_COMPACT, "min.cleanable.dirty.ratio": "0.0", }, @@ -204,7 +204,7 @@ def setUp(self): @matrix( storage_mode=[ TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ], ) def test_key_cardinality_overflow(self, storage_mode: str): diff --git a/tests/rptest/tests/cloud_topics/debug_rows_test.py b/tests/rptest/tests/cloud_topics/debug_rows_test.py index 2d8b0248633c5..4374a856a743c 100644 --- a/tests/rptest/tests/cloud_topics/debug_rows_test.py +++ b/tests/rptest/tests/cloud_topics/debug_rows_test.py @@ -169,7 +169,7 @@ def read_objects( @matrix( storage_mode=[ TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ], ) def test_read_and_write_rows(self, storage_mode: str) -> None: diff --git a/tests/rptest/tests/cloud_topics/e2e_test.py b/tests/rptest/tests/cloud_topics/e2e_test.py index ca0c8550fac49..9fc7b24a01ee8 100644 --- a/tests/rptest/tests/cloud_topics/e2e_test.py +++ b/tests/rptest/tests/cloud_topics/e2e_test.py @@ -100,13 +100,13 @@ def setUp(self): storage_mode = (self.test_context.injected_args or {}).get( "storage_mode", TopicSpec.STORAGE_MODE_CLOUD ) - if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD: + if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2: self.redpanda.set_feature_active( "tiered_cloud_topics", True, timeout_sec=30 ) for topic in self.topics: config = { - TopicSpec.PROPERTY_STORAGE_MODE: storage_mode, + **TopicSpec.storage_mode_config(storage_mode), "cleanup.policy": topic.cleanup_policy, } if topic.min_cleanable_dirty_ratio is not None: @@ -383,7 +383,7 @@ def await_num_produced(self, min_records, timeout_sec=120): @matrix( storage_mode=[ TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ], ) def test_write(self, storage_mode: str): @@ -400,7 +400,7 @@ def test_write(self, storage_mode: str): @matrix( storage_mode=[ TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ], ) def test_delete_records(self, storage_mode: str): @@ -430,7 +430,7 @@ def test_delete_records(self, storage_mode: str): @matrix( storage_mode=[ TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ], ) def test_get_size(self, storage_mode: str): @@ -472,9 +472,10 @@ def get_partition_size() -> int | None: class EndToEndCloudTopicsStorageModeToggleTest(EndToEndCloudTopicsBase): - """Exercise toggling a topic between 'cloud' and 'tiered_cloud' storage - modes while a rate-limited producer is running, then validate the - resulting log with a sequential consumer.""" + """Exercise toggling a topic between 'cloud' and 'tiered' storage modes + (the latter resolves to tiered_v2 via the cluster default) while a + rate-limited producer is running, then validate the resulting log with a + sequential consumer.""" topics = ( TopicSpec( @@ -522,6 +523,13 @@ def test_toggle_storage_mode(self): loop=False, producer=producer, ) + # The toggle flips through the 'tiered' alias; point it at the + # cloud-architecture variant (the default is tiered_v1, under which + # cloud -> tiered is a forbidden transition). + self.rpk.cluster_config_set( + "default_redpanda_storage_mode_tiered_impl", "tiered_v2" + ) + try: producer.start() @@ -530,7 +538,7 @@ def test_toggle_storage_mode(self): while time.time() - start < self.toggle_duration_sec: time.sleep(self.toggle_interval_sec) mode = ( - TopicSpec.STORAGE_MODE_TIERED_CLOUD + TopicSpec.STORAGE_MODE_TIERED if mode == TopicSpec.STORAGE_MODE_CLOUD else TopicSpec.STORAGE_MODE_CLOUD ) @@ -624,7 +632,7 @@ def start_consumer_with_tx(self): @matrix( storage_mode=[ TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ], ) def test_write(self, storage_mode: str): diff --git a/tests/rptest/tests/cloud_topics/iceberg_test.py b/tests/rptest/tests/cloud_topics/iceberg_test.py index dd94754711512..5a6b94311f396 100644 --- a/tests/rptest/tests/cloud_topics/iceberg_test.py +++ b/tests/rptest/tests/cloud_topics/iceberg_test.py @@ -124,7 +124,7 @@ def __init__(self, test_ctx: TestContext): cloud_storage_type=supported_storage_types(), storage_mode=[ TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ], ) def test_compaction_preserves_all_offsets_in_iceberg( @@ -142,7 +142,7 @@ def test_compaction_preserves_all_offsets_in_iceberg( include_query_engines=[QueryEngineType.SPARK], catalog_type=CatalogType.REST_JDBC, ) as dl: - if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD: + if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2: self.redpanda.set_feature_active( "tiered_cloud_topics", True, timeout_sec=30 ) @@ -150,7 +150,7 @@ def test_compaction_preserves_all_offsets_in_iceberg( self.topic_name, iceberg_mode="key_value", config={ - TopicSpec.PROPERTY_STORAGE_MODE: storage_mode, + **TopicSpec.storage_mode_config(storage_mode), "cleanup.policy": TopicSpec.CLEANUP_COMPACT, }, ) @@ -230,7 +230,7 @@ def __init__(self, test_ctx: TestContext) -> None: cloud_storage_type=supported_storage_types(), storage_mode=[ TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ], ) def test_deletion_blocked_until_translated( @@ -248,7 +248,7 @@ def test_deletion_blocked_until_translated( include_query_engines=[QueryEngineType.SPARK], catalog_type=CatalogType.REST_JDBC, ) as dl: - if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD: + if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2: self.redpanda.set_feature_active( "tiered_cloud_topics", True, timeout_sec=30 ) @@ -256,7 +256,7 @@ def test_deletion_blocked_until_translated( self.topic_name, iceberg_mode="key_value", config={ - TopicSpec.PROPERTY_STORAGE_MODE: storage_mode, + **TopicSpec.storage_mode_config(storage_mode), "retention.ms": 500, "retention.bytes": 1024, }, diff --git a/tests/rptest/tests/cloud_topics/retention_test.py b/tests/rptest/tests/cloud_topics/retention_test.py index 09c8c0e60f859..7bfaf2010bfab 100644 --- a/tests/rptest/tests/cloud_topics/retention_test.py +++ b/tests/rptest/tests/cloud_topics/retention_test.py @@ -128,7 +128,7 @@ def _verify_offset_consumable(self, topic: str, offset: int): cloud_storage_type=get_cloud_storage_type(), storage_mode=[ TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ], ) def test_size_based_retention( @@ -152,7 +152,7 @@ def test_size_based_retention( partitions=1, replicas=3, config={ - TopicSpec.PROPERTY_STORAGE_MODE: storage_mode, + **TopicSpec.storage_mode_config(storage_mode), "cleanup.policy": TopicSpec.CLEANUP_DELETE, "retention.bytes": str(initial_retention), }, @@ -223,7 +223,7 @@ def test_size_based_retention( cloud_storage_type=get_cloud_storage_type(), storage_mode=[ TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ], ) def test_time_based_retention( @@ -244,7 +244,7 @@ def test_time_based_retention( partitions=1, replicas=3, config={ - TopicSpec.PROPERTY_STORAGE_MODE: storage_mode, + **TopicSpec.storage_mode_config(storage_mode), "cleanup.policy": TopicSpec.CLEANUP_DELETE, "retention.ms": str(3600000), # 1 hour - data won't be deleted yet }, diff --git a/tests/rptest/tests/cloud_topics_test.py b/tests/rptest/tests/cloud_topics_test.py index 6716d2a896663..54d3e58411154 100644 --- a/tests/rptest/tests/cloud_topics_test.py +++ b/tests/rptest/tests/cloud_topics_test.py @@ -33,7 +33,7 @@ def __create_initial_topics(self, storage_mode): Create initial test topics. Cloud topics are gated by cloud storage, which is already configured via SISettings. """ - if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD: + if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2: self.redpanda.set_feature_active( "tiered_cloud_topics", True, timeout_sec=30 ) @@ -43,7 +43,7 @@ def __create_initial_topics(self, storage_mode): spec.name, spec.partition_count, spec.replication_factor, - config={TopicSpec.PROPERTY_STORAGE_MODE: storage_mode}, + config=TopicSpec.storage_mode_config(storage_mode), ) # Ignored because it's flaky but the test is still useful locally. @@ -53,7 +53,7 @@ def __create_initial_topics(self, storage_mode): cloud_storage_type=get_cloud_storage_type(), storage_mode=[ TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ], ) def test_reconciler_uploads(self, cloud_storage_type, storage_mode): @@ -76,7 +76,7 @@ def count_objects(prefix): err_msg=lambda: f"failed to find at least 1 l1 object(s), instead got {count_objects('l1_')}", ) - if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD: + if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2: # In tiered_cloud mode, data is replicated through raft (no L0 # uploads). Verify no L0 objects were created. l0_count = count_objects("l0_") diff --git a/tests/rptest/tests/cluster_features_test.py b/tests/rptest/tests/cluster_features_test.py index 6bc5b4fde3839..c942742e04fbb 100644 --- a/tests/rptest/tests/cluster_features_test.py +++ b/tests/rptest/tests/cluster_features_test.py @@ -1461,8 +1461,9 @@ def _perturb_v26_1_to_v26_2(self, phase): # (batch_mirror_topic_status additionally needs a second cluster). def _exercise_tiered_cloud_topics(self): - """tiered_cloud_topics gate: creating a topic with storage mode - `tiered_cloud` is refused until the feature is active. Drive that + """tiered_cloud_topics gate: creating a topic with the tiered_v2 + (cloud-architecture) storage mode is refused until the feature is + active. Drive that validator path while unfinalized and confirm the feature is unavailable and the create is rejected. The topic is never created, so exercising the gate cannot leave state that would block a downgrade.""" @@ -1473,9 +1474,9 @@ def _exercise_tiered_cloud_topics(self): RpkTool(self.redpanda).create_topic( "perturb-tiered-cloud", partitions=1, - config={ - TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_TIERED_CLOUD - }, + config=TopicSpec.storage_mode_config( + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 + ), ) except RpkException as e: # Confirm the create failed via the storage-mode gate, not an @@ -1550,23 +1551,15 @@ def _exercise_shadow_link_role_sync(self): def _verify_tiered_cloud_topics_working(self): """After finalize the active version has advanced past the feature's - require_version, so the gate opens: the feature moves from unavailable to - available (it is explicit_only, so it does not auto-activate) and can - then be enabled to active -- the simple signal that it now works. + require_version, so the gate opens: the feature auto-activates (it is + available_policy::always) -- the simple signal that it now works. (Creating an actual tiered_cloud topic additionally needs cloud storage, which this test does not configure.)""" - wait_until( - lambda: self._feature_state("tiered_cloud_topics") == "available", - timeout_sec=30, - backoff_sec=1, - err_msg="tiered_cloud_topics did not become available after finalize", - ) - self.admin.put_feature("tiered_cloud_topics", {"state": "active"}) wait_until( lambda: self._feature_state("tiered_cloud_topics") == "active", timeout_sec=30, backoff_sec=1, - err_msg="tiered_cloud_topics did not activate after being enabled", + err_msg="tiered_cloud_topics did not auto-activate after finalize", ) def _verify_shadow_link_role_sync_working(self): diff --git a/tests/rptest/tests/cluster_linking_e2e_test.py b/tests/rptest/tests/cluster_linking_e2e_test.py index 8642ddc42b2a9..0b918f7269b13 100644 --- a/tests/rptest/tests/cluster_linking_e2e_test.py +++ b/tests/rptest/tests/cluster_linking_e2e_test.py @@ -1929,7 +1929,7 @@ def test_replication_with_failures(self, storage_mode): cloud_backed = storage_mode in ( TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ) progress_timeout = 120 if cloud_backed else 60 @@ -2054,7 +2054,7 @@ def test_replication_with_transactions(self, storage_mode): # cloud-topics linger. if storage_mode in ( TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ): self.source_cluster_service.set_cluster_config( {"cloud_topics_produce_upload_interval": 25} @@ -2551,7 +2551,7 @@ def test_replication_with_compaction(self, storage_mode): if storage_mode not in ( TopicSpec.STORAGE_MODE_LOCAL, TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ): # Compaction on shadow topics is not yet supported with # tiered storage mode. @@ -4378,7 +4378,7 @@ def test_starting_offset( if ( starting_offset == self.timequery_offset - and storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD + and storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 ): # Timestamp queries on tiered_cloud shadow topics are not # yet supported. @@ -4705,7 +4705,8 @@ def target_has_new_data(): class ShadowLinkingCloudTopicReplicationTests(ShadowLinkPreAllocTestBase): """ Tests cluster linking replication with cloud topics - (redpanda.storage.mode=cloud and tiered_cloud) on the source cluster. + (redpanda.storage.mode=cloud and tiered with version tiered_v2) on the + source cluster. """ def __init__(self, test_context: TestContext, *args: Any, **kwargs: Any): @@ -4738,15 +4739,15 @@ def __init__(self, test_context: TestContext, *args: Any, **kwargs: Any): @matrix( storage_mode=[ TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ], ) def test_cloud_topic_replication(self, storage_mode): """ - Verify that data produced to a cloud/tiered_cloud topic on the source + Verify that data produced to a cloud/tiered_v2 topic on the source cluster is replicated to the target cluster via cluster linking. """ - if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD: + if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2: self.source_cluster_service.set_feature_active( "tiered_cloud_topics", True, timeout_sec=30 ) @@ -4768,9 +4769,7 @@ def create_source_topic(): topic=topic.name, partitions=topic.partition_count, replicas=topic.replication_factor, - config={ - TopicSpec.PROPERTY_STORAGE_MODE: storage_mode, - }, + config=TopicSpec.storage_mode_config(storage_mode), ) return True except Exception as e: @@ -4787,11 +4786,23 @@ def create_source_topic(): err_msg=f"Failed to create source topic with storage_mode={storage_mode}", ) - source_configs = source_rpk.describe_topic_configs(topic.name) - assert source_configs[TopicSpec.PROPERTY_STORAGE_MODE][0] == storage_mode, ( - f"Source topic storage mode: {source_configs[TopicSpec.PROPERTY_STORAGE_MODE]}" + expected_mode = ( + TopicSpec.STORAGE_MODE_TIERED + if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 + else storage_mode ) + source_configs = source_rpk.describe_topic_configs(topic.name) + assert source_configs[TopicSpec.PROPERTY_STORAGE_MODE][0] == expected_mode, ( + f"Source topic storage mode: {source_configs[TopicSpec.PROPERTY_STORAGE_MODE]}, " + f"expected: {expected_mode}" + ) + if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2: + source_version = source_configs[TopicSpec.PROPERTY_STORAGE_MODE_IMPL][0] + assert source_version == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ( + f"Source topic storage mode version: {source_version}" + ) + self.create_link("test-link") self.target_cluster.service.wait_until( @@ -4804,10 +4815,15 @@ def create_source_topic(): # Verify target topic has the same storage mode target_rpk = RpkTool(self.target_cluster.service) target_configs = target_rpk.describe_topic_configs(topic.name) - assert target_configs[TopicSpec.PROPERTY_STORAGE_MODE][0] == storage_mode, ( + assert target_configs[TopicSpec.PROPERTY_STORAGE_MODE][0] == expected_mode, ( f"Target topic storage mode: {target_configs[TopicSpec.PROPERTY_STORAGE_MODE]}, " - f"expected: {storage_mode}" + f"expected: {expected_mode}" ) + if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2: + target_version = target_configs[TopicSpec.PROPERTY_STORAGE_MODE_IMPL][0] + assert target_version == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ( + f"Target topic storage mode version: {target_version}" + ) with self.producer_consumer(topic=topic.name, msg_size=128, msg_cnt=10000): self.verify() diff --git a/tests/rptest/tests/cluster_linking_storage_mode_sync_test.py b/tests/rptest/tests/cluster_linking_storage_mode_sync_test.py new file mode 100644 index 0000000000000..541e9779e4696 --- /dev/null +++ b/tests/rptest/tests/cluster_linking_storage_mode_sync_test.py @@ -0,0 +1,211 @@ +# Copyright 2026 Redpanda Data, Inc. +# +# Use of this software is governed by the Business Source License +# included in the file licenses/BSL.md +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0 + +from typing import Any + +from ducktape.mark import matrix +from ducktape.tests.test import TestContext +from ducktape.utils.util import wait_until + +from rptest.clients.rpk import RpkException, RpkTool +from rptest.clients.types import TopicSpec +from rptest.services.cluster import cluster +from rptest.services.redpanda import SISettings +from rptest.tests.cluster_linking_test_base import ( + SecondaryClusterArgs, + ShadowLinkTestBase, +) + + +class ShadowLinkStorageModeSyncTest(ShadowLinkTestBase): + """ + Verify that the storage mode of a source topic survives the shadow-link + property sync even when the two clusters disagree on what the 'tiered' + alias means (different default_redpanda_storage_mode_tiered_impl values). + + The sync transports the describe output, where both tiered variants + display as 'tiered'; the read-only redpanda.storage.mode.impl + property carries the variant, and the target must reconstruct the + source topic's exact variant from the pair rather than re-interpreting + 'tiered' through its own cluster config. + """ + + def __init__(self, test_context: TestContext, *args: Any, **kwargs: Any): + si_settings = SISettings( + test_context, + cloud_storage_max_connections=10, + cloud_storage_enable_remote_read=False, + cloud_storage_enable_remote_write=False, + fast_uploads=True, + ) + + injected = test_context.injected_args or {} + source_default = injected.get("source_default", "tiered_v2") + target_default = injected.get("target_default", "tiered_v1") + + super().__init__( + test_context, + si_settings=si_settings, + extra_rp_conf={ + "enable_cluster_metadata_upload_loop": False, + "default_redpanda_storage_mode_tiered_impl": target_default, + }, + secondary_cluster_args=SecondaryClusterArgs( + si_settings=si_settings, + extra_rp_conf={ + "enable_shadow_linking": True, + "enable_cluster_metadata_upload_loop": False, + "default_redpanda_storage_mode_tiered_impl": source_default, + }, + ), + *args, + **kwargs, + ) + + def _create_source_topic(self, name: str, config: dict[str, str]): + source_rpk = RpkTool(self.source_cluster.service) + + def try_create(): + try: + source_rpk.create_topic( + topic=name, partitions=1, replicas=1, config=config + ) + return True + except Exception as e: + if "INVALID_CONFIG" in str(e): + return False + raise + + # Retry topic creation: feature flag propagation may lag behind + # the admin API response on some nodes. + wait_until( + try_create, + timeout_sec=30, + backoff_sec=2, + err_msg=f"Failed to create source topic {name}", + ) + + def _target_storage_mode(self, topic: str) -> tuple[str | None, str | None]: + try: + configs = RpkTool(self.target_cluster.service).describe_topic_configs(topic) + except RpkException as e: + # The mirror topic may not exist (yet) or its metadata may not + # have propagated to the broker rpk picked. + self.logger.debug(f"describe {topic} failed: {e}") + return (None, None) + mode = configs.get(TopicSpec.PROPERTY_STORAGE_MODE) + version = configs.get(TopicSpec.PROPERTY_STORAGE_MODE_IMPL) + return ( + mode[0] if mode is not None else None, + version[0] if version is not None else None, + ) + + @cluster(num_nodes=6) + @matrix( + source_default=["tiered_v1", "tiered_v2"], + target_default=["tiered_v1", "tiered_v2"], + ) + def test_storage_mode_sync_across_defaults( + self, source_default: str, target_default: str + ): + self.source_cluster_service.set_feature_active( + "tiered_cloud_topics", True, timeout_sec=30 + ) + self.target_cluster.service.set_feature_active( + "tiered_cloud_topics", True, timeout_sec=30 + ) + + # topic -> (create config, expected impl on the target); the impl + # property is always present and unambiguous + cases: dict[str, tuple[dict[str, str], str]] = { + "sync-cloud": ( + {TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_CLOUD}, + TopicSpec.STORAGE_MODE_CLOUD, + ), + # The alias resolves on the SOURCE at creation; the synced topic + # must land on the source's variant, not the target's. + "sync-tiered-alias": ( + {TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_TIERED}, + source_default, + ), + "sync-tiered-v1": ( + TopicSpec.storage_mode_config(TopicSpec.STORAGE_MODE_IMPL_TIERED_V1), + TopicSpec.STORAGE_MODE_IMPL_TIERED_V1, + ), + "sync-tiered-v2": ( + TopicSpec.storage_mode_config(TopicSpec.STORAGE_MODE_IMPL_TIERED_V2), + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, + ), + } + + for name, (config, _) in cases.items(): + self._create_source_topic(name, config) + + self.create_link("storage-mode-sync-link") + + for name, (_, expected_impl) in cases.items(): + expected_mode = ( + TopicSpec.STORAGE_MODE_TIERED + if expected_impl + in ( + TopicSpec.STORAGE_MODE_IMPL_TIERED_V1, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, + ) + else expected_impl + ) + + def synced( + name: str = name, + expected_mode: str = expected_mode, + expected_version: str | None = expected_impl, + ) -> bool: + if not self.topic_exists_in_target(topic=name, partition_count=1): + return False + mode, version = self._target_storage_mode(name) + return mode == expected_mode and version == expected_version + + self.target_cluster.service.wait_until( + synced, + timeout_sec=60, + backoff_sec=2, + err_msg=( + f"{name}: target did not reach mode={expected_mode} " + f"impl={expected_impl}" + ), + ) + + # Update path: altering the source topic's mode must propagate the + # variant through the (mode, version) pair as well. The conversion + # goes through the 'tiered' alias, which on the source only reaches + # the tiered_v2 variant (cloud -> tiered_v1 is not a permitted + # transition), so this phase only applies when the source alias + # points at tiered_v2. + if source_default == "tiered_v2": + RpkTool(self.source_cluster.service).alter_topic_config( + "sync-cloud", + TopicSpec.PROPERTY_STORAGE_MODE, + TopicSpec.STORAGE_MODE_TIERED, + ) + + def converted() -> bool: + mode, version = self._target_storage_mode("sync-cloud") + return ( + mode == TopicSpec.STORAGE_MODE_TIERED + and version == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 + ) + + self.target_cluster.service.wait_until( + converted, + timeout_sec=60, + backoff_sec=2, + err_msg=( + "sync-cloud: cloud -> tiered_v2 conversion did not " + "propagate to the target" + ), + ) diff --git a/tests/rptest/tests/cluster_linking_test_base.py b/tests/rptest/tests/cluster_linking_test_base.py index d0246da868662..91edf3e17b3bc 100644 --- a/tests/rptest/tests/cluster_linking_test_base.py +++ b/tests/rptest/tests/cluster_linking_test_base.py @@ -86,6 +86,7 @@ "max.compaction.lag.ms", "min.compaction.lag.ms", "redpanda.storage.mode", + "redpanda.storage.mode.impl", ] DISALLOWED_SYNCED_TOPIC_PROPERTIES = [ @@ -104,9 +105,9 @@ ALL_STORAGE_MODES = [ TopicSpec.STORAGE_MODE_LOCAL, - TopicSpec.STORAGE_MODE_TIERED, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V1, TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ] # Log messages that are expected when running shadow link tests with @@ -601,12 +602,13 @@ def __init__( storage_mode = (test_context.injected_args or {}).get("storage_mode") needs_si = storage_mode in ( TopicSpec.STORAGE_MODE_TIERED, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V1, TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ) needs_cloud_topics = storage_mode in ( TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ) if needs_si and "si_settings" not in kwargs: @@ -1027,7 +1029,7 @@ def create_source_topic(self, topic: TopicSpec, storage_mode: str | None = None) self.source_default_client().create_topic(topic) return - if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD: + if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2: self.source_cluster_service.set_feature_active( "tiered_cloud_topics", True, timeout_sec=30 ) @@ -1036,7 +1038,7 @@ def create_source_topic(self, topic: TopicSpec, storage_mode: str | None = None) ) config = self._topic_config_from_spec(topic) - config[TopicSpec.PROPERTY_STORAGE_MODE] = storage_mode + config.update(TopicSpec.storage_mode_config(storage_mode)) source_rpk = RpkTool(self.source_cluster.service) diff --git a/tests/rptest/tests/describe_topics_test.py b/tests/rptest/tests/describe_topics_test.py index 25d1981792322..ca308b018ebfb 100644 --- a/tests/rptest/tests/describe_topics_test.py +++ b/tests/rptest/tests/describe_topics_test.py @@ -343,6 +343,16 @@ def test_describe_topics_with_documentation_and_types(self): "the Cloud Topics architecture, or `unset` to use legacy " "remote.read/write configs for backwards compatibility.", ), + "redpanda.storage.mode.impl": ConfigProperty( + config_type="STRING", + value="unset", + doc_string="Exact implementation of the topic's storage mode. Tiered " + "topics report tiered_v1 (classic tiered-storage " + "architecture) or tiered_v2 (new tiered-storage " + "architecture); other modes mirror redpanda.storage.mode. " + "Read-only after creation: supply it at topic creation to " + "select the implementation explicitly.", + ), } tp_spec = TopicSpec() @@ -364,8 +374,10 @@ def test_describe_topics_with_documentation_and_types(self): # 2. Property documentation string # 3. An empty line + # kcl marks read-only properties with a trailing asterisk + # (e.g. redpanda.storage.mode.impl*) property_re = re.compile( - r"^(?P[a-z.]+?)\s+(?P[a-z]+?)\s+(?P\S+?)\s+(?P[a-z_]+?)$" + r"^(?P[a-z.]+?)\*?\s+(?P[a-z]+?)\s+(?P\S+?)\s+(?P[a-z_]+?)$" ) last_pos = None @@ -405,7 +417,7 @@ def test_describe_topics_with_documentation_and_types(self): assert len(doc_lines) % 3 == 0 # The property name in the doc section has a colon - name_re = re.compile(r"^(?P[a-z.]+?):$") + name_re = re.compile(r"^(?P[a-z.]+?)\*?:$") # Finally, check the doc section for i in range(0, len(doc_lines) // 3): diff --git a/tests/rptest/tests/follower_fetching_test.py b/tests/rptest/tests/follower_fetching_test.py index a0fccdc9eed66..a0bdd4df2f5c3 100644 --- a/tests/rptest/tests/follower_fetching_test.py +++ b/tests/rptest/tests/follower_fetching_test.py @@ -40,9 +40,7 @@ def make_topic_config(fetch_from): TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_CLOUD, } elif fetch_from == FetchFrom.TIERED_CLOUD_TOPIC: - config = { - TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_TIERED_CLOUD, - } + config = TopicSpec.storage_mode_config(TopicSpec.STORAGE_MODE_IMPL_TIERED_V2) elif fetch_from == FetchFrom.TIERED_STORAGE: config = { TopicSpec.PROPERTY_REMOTE_READ: "true", diff --git a/tests/rptest/tests/random_node_operations_smoke_test.py b/tests/rptest/tests/random_node_operations_smoke_test.py index 189e87d7fdcdf..16bf059febf62 100644 --- a/tests/rptest/tests/random_node_operations_smoke_test.py +++ b/tests/rptest/tests/random_node_operations_smoke_test.py @@ -768,7 +768,9 @@ def schema_registry_topic_created(): "cleanup.policy": "delete", "redpanda.remote.read": "false", "redpanda.remote.write": "false", - TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_TIERED_CLOUD, + **TopicSpec.storage_mode_config( + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 + ), }, ) self.maybe_enable_iceberg_for_topic( @@ -785,7 +787,9 @@ def schema_registry_topic_created(): "cleanup.policy": "compact", "redpanda.remote.read": "false", "redpanda.remote.write": "false", - TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_TIERED_CLOUD, + **TopicSpec.storage_mode_config( + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 + ), }, ) self.maybe_enable_iceberg_for_topic( diff --git a/tests/rptest/tests/shadow_linking_rnot_test.py b/tests/rptest/tests/shadow_linking_rnot_test.py index e0e8e80fbf0cf..6542ff4c7db63 100644 --- a/tests/rptest/tests/shadow_linking_rnot_test.py +++ b/tests/rptest/tests/shadow_linking_rnot_test.py @@ -277,6 +277,11 @@ def __init__(self, test_ctx: TestContext): ), extra_rp_conf={ "group_new_member_join_timeout": 3000, + # The flipping workload alters redpanda.storage.mode + # between 'cloud' and 'tiered'; the alter path resolves + # 'tiered' through default_redpanda_storage_mode_tiered_impl, so pin + # tiered_v2 to flip to the cloud-architecture variant. + "default_redpanda_storage_mode_tiered_impl": "tiered_v2", }, ), extra_rp_conf={ @@ -388,7 +393,9 @@ def _basic_workload_specs(self) -> list[ClusterLinkingWorkloadSpec]: ClusterLinkingWorkloadSpec( topic="tiered-cloud-topic", topic_properties={ - TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_TIERED_CLOUD, + **TopicSpec.storage_mode_config( + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 + ), "segment.bytes": f"{1024 * 1024}", }, partition_count=self.partition_count, @@ -410,7 +417,7 @@ def _flipping_workload_specs(self) -> list[ClusterLinkingWorkloadSpec]: msg_size=self.msg_size, flip_storage_modes=[ TopicSpec.STORAGE_MODE_CLOUD, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_TIERED, ], flip_interval_seconds=3.0, ), @@ -420,13 +427,13 @@ def _cloud_combo_workload_specs(self) -> list[ClusterLinkingWorkloadSpec]: specs: list[ClusterLinkingWorkloadSpec] = [] for mode, mode_label in ( (TopicSpec.STORAGE_MODE_CLOUD, "cloud"), - (TopicSpec.STORAGE_MODE_TIERED_CLOUD, "tiered-cloud"), + (TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, "tiered-cloud"), ): specs.append( ClusterLinkingWorkloadSpec( topic=f"{mode_label}-compacted-topic", topic_properties={ - TopicSpec.PROPERTY_STORAGE_MODE: mode, + **TopicSpec.storage_mode_config(mode), "cleanup.policy": "compact", "segment.bytes": f"{1024 * 1024}", }, @@ -447,7 +454,7 @@ def _cloud_combo_workload_specs(self) -> list[ClusterLinkingWorkloadSpec]: ClusterLinkingWorkloadSpec( topic=f"{mode_label}-topic-txns", topic_properties={ - TopicSpec.PROPERTY_STORAGE_MODE: mode, + **TopicSpec.storage_mode_config(mode), }, msg_count=math.floor(self.msg_count / 10), msg_size=self.msg_size, @@ -475,9 +482,9 @@ def _cloud_combo_workload_specs(self) -> list[ClusterLinkingWorkloadSpec]: def test_node_operations(self, failures: bool, workload_set: str): self.setup_scale() - # tiered_cloud_topics is an explicit-only feature and must be - # enabled on both clusters before any tiered_cloud topic can be - # created. + # tiered_cloud_topics must be active on both clusters before any + # tiered_cloud topic can be created (it auto-activates once the + # cluster is fully on v26.2; this is a no-op then). self.source_cluster.service.set_feature_active( "tiered_cloud_topics", True, timeout_sec=30 ) diff --git a/tests/rptest/tests/storage_mode_test.py b/tests/rptest/tests/storage_mode_test.py index dfd821de990cb..ec5ba8a075b16 100644 --- a/tests/rptest/tests/storage_mode_test.py +++ b/tests/rptest/tests/storage_mode_test.py @@ -7,13 +7,16 @@ # the Business Source License, use of this software will be governed # by the Apache License, Version 2.0 +from collections.abc import Callable + from ducktape.tests.test import TestContext -from rptest.clients.rpk import RpkTool +from rptest.clients.rpk import RpkException, RpkTool from rptest.clients.types import TopicSpec from rptest.services.admin import Admin from rptest.services.cluster import cluster from rptest.services.redpanda import ( + PREV_VERSION_LOG_ALLOW_LIST, RESTART_LOG_ALLOW_LIST, SISettings, ) @@ -45,6 +48,13 @@ def _get_topic_storage_mode(self, rpk: RpkTool, topic_name: str) -> str | None: """Get the storage mode property value for a topic.""" return self._get_topic_config(rpk, topic_name, TopicSpec.PROPERTY_STORAGE_MODE) + def _get_topic_storage_mode_impl(self, rpk: RpkTool, topic_name: str) -> str | None: + """Get the read-only storage mode impl property, or None if the + broker does not report it (pre-impl binaries).""" + return self._get_topic_config( + rpk, topic_name, TopicSpec.PROPERTY_STORAGE_MODE_IMPL + ) + def _get_topic_remote_read(self, rpk: RpkTool, topic_name: str) -> str | None: """Get the remote read property value for a topic.""" return self._get_topic_config(rpk, topic_name, TopicSpec.PROPERTY_REMOTE_READ) @@ -566,7 +576,9 @@ def test_storage_mode_transitions(self): == TopicSpec.STORAGE_MODE_UNSET ), "Storage mode should still be unset after rejected transition" - # Test blocked transition: unset -> tiered_cloud + # Test rejected input: the tiered variant names are not valid + # redpanda.storage.mode values (a variant is selected at creation + # via redpanda.storage.mode.impl) self._create_topic(rpk, "topic-unset-to-tiered-cloud") assert ( self._get_topic_storage_mode(rpk, "topic-unset-to-tiered-cloud") @@ -576,19 +588,20 @@ def test_storage_mode_transitions(self): rpk.alter_topic_config( "topic-unset-to-tiered-cloud", TopicSpec.PROPERTY_STORAGE_MODE, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ) assert False, ( - "Transition from unset to tiered_cloud should have been rejected" + "'tiered_v2' should be rejected as a redpanda.storage.mode value" ) except Exception: - pass # Expected - transition should be rejected + pass # Expected - invalid mode value assert ( self._get_topic_storage_mode(rpk, "topic-unset-to-tiered-cloud") == TopicSpec.STORAGE_MODE_UNSET ), "Storage mode should still be unset after rejected transition" - # Test blocked transition: local -> tiered_cloud + # Test read-only property: redpanda.storage.mode.impl cannot be + # altered self._create_topic( rpk, "topic-local-to-tiered-cloud", @@ -597,14 +610,12 @@ def test_storage_mode_transitions(self): try: rpk.alter_topic_config( "topic-local-to-tiered-cloud", - TopicSpec.PROPERTY_STORAGE_MODE, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, - ) - assert False, ( - "Transition from local to tiered_cloud should have been rejected" + TopicSpec.PROPERTY_STORAGE_MODE_IMPL, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ) + assert False, "redpanda.storage.mode.impl should be read-only for alter" except Exception: - pass # Expected - transition should be rejected + pass # Expected - read-only property assert ( self._get_topic_storage_mode(rpk, "topic-local-to-tiered-cloud") == TopicSpec.STORAGE_MODE_LOCAL @@ -671,10 +682,15 @@ def test_default_storage_mode_validation(self): ) -class StorageModeCloudTransitionTest(StorageModeTestBase): +class TieredCloudUpgradeTest(StorageModeTestBase): """ - Test storage mode transitions between cloud and tiered_cloud. - Requires cloud topics to be enabled. + The tiered_v2 storage mode is gated by the tiered_cloud_topics feature + flag (v26.2, available_policy::always). While the cluster is only + partially upgraded the flag is not active, so creating a tiered_v2 topic + or converting a cloud topic to it must be rejected. Once every node runs + v26.2 the flag auto-activates and both operations succeed, with the + upgraded cluster keeping the classic meaning of 'tiered' + (default_redpanda_storage_mode_tiered_impl=tiered_v1 via legacy default). """ def __init__(self, test_context: TestContext): @@ -684,25 +700,203 @@ def __init__(self, test_context: TestContext): cloud_storage_enable_remote_write=False, ) - super(StorageModeCloudTransitionTest, self).__init__( + super(TieredCloudUpgradeTest, self).__init__( test_context=test_context, num_brokers=3, si_settings=si_settings, + # The v26.1 binary additionally gates cloud-mode topics on the + # cloud_topics_enabled cluster property; deprecated and ignored + # from v26.2 on. + extra_rp_conf={"cloud_topics_enabled": True}, ) + self.installer = self.redpanda._installer def setUp(self): - super().setUp() - self.redpanda.set_feature_active("tiered_cloud_topics", True, timeout_sec=30) + # Start the whole cluster on the latest release of the prior feature + # line (26.1.x), which supports the cloud storage mode but predates + # the tiered_v2 variant and its feature flag. + old_version = self.installer.highest_from_prior_feature_version( + RedpandaInstaller.HEAD + ) + self.installer.install(self.redpanda.nodes, old_version) + super(TieredCloudUpgradeTest, self).setUp() + + def _expect_rejected(self, what: str, fn: Callable[[], object]): + try: + fn() + except RpkException as e: + self.logger.info(f"{what} rejected as expected: {e}") + else: + raise AssertionError( + f"{what} should have been rejected in a partially upgraded cluster" + ) + + @cluster( + num_nodes=3, + log_allow_list=RESTART_LOG_ALLOW_LIST + PREV_VERSION_LOG_ALLOW_LIST, + ) + def test_tiered_cloud_gated_in_mixed_cluster(self): + rpk = RpkTool(self.redpanda) + _ = wait_for_num_versions(self.redpanda, 1) + + # The cloud storage mode is available since v26.1, so this topic can + # be created before the upgrade begins. + self._create_topic( + rpk, + "topic-cloud", + config={TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_CLOUD}, + ) + assert ( + self._get_topic_storage_mode(rpk, "topic-cloud") + == TopicSpec.STORAGE_MODE_CLOUD + ) + + # Upgrade a single node to HEAD to put the cluster in a mixed state. + first = self.redpanda.nodes[0] + self.installer.install([first], RedpandaInstaller.HEAD) + self.redpanda.restart_nodes([first]) + _ = wait_for_num_versions(self.redpanda, 2) + + # The upgraded node knows the feature but must not report it active + # while old nodes are still in the cluster. + state = self.redpanda.get_feature_state("tiered_cloud_topics", node=first) + assert state == "unavailable", ( + f"tiered_cloud_topics should be unavailable in a mixed cluster, got {state}" + ) + + # CreateTopics is routed to the controller broker. A HEAD controller + # rejects the request via the feature gate. A v26.1 controller does + # not know the redpanda.storage.mode.impl property: unsupported + # topic configs are ignored on create, so the request silently + # degrades to a classic tiered topic. Either way the gating + # invariant holds: no tiered_v2 topic can exist in a partially + # upgraded cluster. + created_in_mixed = True + try: + self._create_topic( + rpk, + "topic-tiered-v2-mixed", + config=TopicSpec.storage_mode_config( + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 + ), + ) + except RpkException as e: + created_in_mixed = False + self.logger.info(f"tiered_v2 creation rejected in mixed cluster: {e}") + if created_in_mixed: + version = self._get_topic_storage_mode_impl(rpk, "topic-tiered-v2-mixed") + assert version != TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, ( + "a tiered_v2 topic must not be creatable in a partially " + "upgraded cluster" + ) + + # Conversion is expressed as an alter to 'tiered'; both binaries + # resolve it to a variant the cloud -> X transition rules forbid + # (classic tiered) while the cluster is not fully upgraded. + self._expect_rejected( + "cloud to tiered_v2 conversion", + lambda: rpk.alter_topic_config( + "topic-cloud", + TopicSpec.PROPERTY_STORAGE_MODE, + TopicSpec.STORAGE_MODE_TIERED, + ), + ) + assert ( + self._get_topic_storage_mode(rpk, "topic-cloud") + == TopicSpec.STORAGE_MODE_CLOUD + ), "Storage mode should still be cloud after the rejected conversion" + + # Finish the upgrade: the feature auto-activates once every node + # runs v26.2. + self.installer.install(self.redpanda.nodes, RedpandaInstaller.HEAD) + self.redpanda.restart_nodes(self.redpanda.nodes[1:]) + _ = wait_for_num_versions(self.redpanda, 1) + self.redpanda.await_feature("tiered_cloud_topics", "active", timeout_sec=60) + + # tiered_v1 is the default for all clusters, so an upgrade never + # changes what 'tiered' means. + default_impl = Admin(self.redpanda).get_cluster_config()[ + "default_redpanda_storage_mode_tiered_impl" + ] + assert default_impl == "tiered_v1", ( + f"clusters should default to tiered_v1, got {default_impl}" + ) + + # If the mixed-cluster create went through a v26.1 controller, the + # resulting topic must have degraded to classic tiered - never the + # v2 variant. Now that every broker runs HEAD, the version property + # is authoritative. + if created_in_mixed: + assert ( + self._get_topic_storage_mode_impl(rpk, "topic-tiered-v2-mixed") + == TopicSpec.STORAGE_MODE_IMPL_TIERED_V1 + ), "the mixed-cluster create must not have produced a tiered_v2 topic" + + # Creation with an explicit version works under any default. + self._create_topic( + rpk, + "topic-tiered-v2", + config=TopicSpec.storage_mode_config(TopicSpec.STORAGE_MODE_IMPL_TIERED_V2), + ) + assert ( + self._get_topic_storage_mode(rpk, "topic-tiered-v2") + == TopicSpec.STORAGE_MODE_TIERED + ), "tiered_v2 topic creation should succeed after the upgrade" + assert ( + self._get_topic_storage_mode_impl(rpk, "topic-tiered-v2") + == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 + ) + + # Conversion goes through the 'tiered' alias, so the admin first + # points it at the tiered_v2 variant (the version property is + # read-only for alter). + rpk.cluster_config_set("default_redpanda_storage_mode_tiered_impl", "tiered_v2") + rpk.alter_topic_config( + "topic-cloud", + TopicSpec.PROPERTY_STORAGE_MODE, + TopicSpec.STORAGE_MODE_TIERED, + ) + assert ( + self._get_topic_storage_mode(rpk, "topic-cloud") + == TopicSpec.STORAGE_MODE_TIERED + ), "cloud to tiered_v2 conversion should succeed after the upgrade" + assert ( + self._get_topic_storage_mode_impl(rpk, "topic-cloud") + == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 + ) + + +class StorageModeCloudTransitionTest(StorageModeTestBase): + """ + Test storage mode transitions between the cloud and tiered_v2 modes. + Transitions are expressed through the 'tiered' alias (the version + property is read-only), so the tests flip default_redpanda_storage_mode_tiered_impl at + runtime to select the variant the alias refers to. + """ + + def __init__(self, test_context: TestContext): + si_settings = SISettings( + test_context, + cloud_storage_enable_remote_read=False, + cloud_storage_enable_remote_write=False, + ) + + super(StorageModeCloudTransitionTest, self).__init__( + test_context=test_context, + num_brokers=3, + si_settings=si_settings, + ) @cluster(num_nodes=3) def test_cloud_to_tiered_cloud_transition(self): """ - Test that cloud -> tiered_cloud and tiered_cloud -> cloud transitions - are permitted. + cloud -> tiered_v2 and tiered_v2 -> cloud transitions are permitted; + with default_redpanda_storage_mode_tiered_impl=tiered_v2 both are + reachable through the 'tiered' alias. """ rpk = RpkTool(self.redpanda) + rpk.cluster_config_set("default_redpanda_storage_mode_tiered_impl", "tiered_v2") - # Create a cloud topic self._create_topic( rpk, "topic-cloud-to-tc", @@ -713,18 +907,20 @@ def test_cloud_to_tiered_cloud_transition(self): == TopicSpec.STORAGE_MODE_CLOUD ) - # Transition cloud -> tiered_cloud (permitted) rpk.alter_topic_config( "topic-cloud-to-tc", TopicSpec.PROPERTY_STORAGE_MODE, - TopicSpec.STORAGE_MODE_TIERED_CLOUD, + TopicSpec.STORAGE_MODE_TIERED, ) assert ( self._get_topic_storage_mode(rpk, "topic-cloud-to-tc") - == TopicSpec.STORAGE_MODE_TIERED_CLOUD - ), "Transition from cloud to tiered_cloud should succeed" + == TopicSpec.STORAGE_MODE_TIERED + ), "Transition from cloud to tiered_v2 should succeed" + assert ( + self._get_topic_storage_mode_impl(rpk, "topic-cloud-to-tc") + == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 + ) - # Transition tiered_cloud -> cloud (permitted) rpk.alter_topic_config( "topic-cloud-to-tc", TopicSpec.PROPERTY_STORAGE_MODE, @@ -733,94 +929,238 @@ def test_cloud_to_tiered_cloud_transition(self): assert ( self._get_topic_storage_mode(rpk, "topic-cloud-to-tc") == TopicSpec.STORAGE_MODE_CLOUD - ), "Transition from tiered_cloud to cloud should succeed" + ), "Transition from tiered_v2 to cloud should succeed" + assert ( + self._get_topic_storage_mode_impl(rpk, "topic-cloud-to-tc") + == TopicSpec.STORAGE_MODE_CLOUD + ), "the impl property must mirror the cloud storage mode" @cluster(num_nodes=3) def test_tiered_cloud_blocked_transitions(self): """ - Test that tiered_cloud cannot transition to local, tiered, or unset. + A tiered_v2 topic cannot transition to local, classic tiered, or + unset. """ rpk = RpkTool(self.redpanda) - # Create a tiered_cloud topic self._create_topic( rpk, "topic-tc-blocked", - config={ - TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_TIERED_CLOUD - }, + config=TopicSpec.storage_mode_config(TopicSpec.STORAGE_MODE_IMPL_TIERED_V2), ) assert ( - self._get_topic_storage_mode(rpk, "topic-tc-blocked") - == TopicSpec.STORAGE_MODE_TIERED_CLOUD + self._get_topic_storage_mode_impl(rpk, "topic-tc-blocked") + == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 ) - # tiered_cloud -> local (blocked) - try: - rpk.alter_topic_config( - "topic-tc-blocked", - TopicSpec.PROPERTY_STORAGE_MODE, - TopicSpec.STORAGE_MODE_LOCAL, - ) - assert False, ( - "Transition from tiered_cloud to local should have been rejected" - ) - except Exception: - pass - assert ( - self._get_topic_storage_mode(rpk, "topic-tc-blocked") - == TopicSpec.STORAGE_MODE_TIERED_CLOUD - ), "Storage mode should still be tiered_cloud after rejected transition" + # With the alias pointing at the classic variant, altering to + # 'tiered' attempts tiered_v2 -> tiered_v1: blocked. + rpk.cluster_config_set("default_redpanda_storage_mode_tiered_impl", "tiered_v1") + for target in ( + TopicSpec.STORAGE_MODE_TIERED, + TopicSpec.STORAGE_MODE_LOCAL, + TopicSpec.STORAGE_MODE_UNSET, + ): + try: + rpk.alter_topic_config( + "topic-tc-blocked", + TopicSpec.PROPERTY_STORAGE_MODE, + target, + ) + assert False, ( + f"Transition from tiered_v2 to {target} should have been rejected" + ) + except RpkException: + pass + assert ( + self._get_topic_storage_mode_impl(rpk, "topic-tc-blocked") + == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 + ), "Topic should still be tiered_v2 after rejected transition" - # tiered_cloud -> tiered (blocked) - try: - rpk.alter_topic_config( - "topic-tc-blocked", - TopicSpec.PROPERTY_STORAGE_MODE, - TopicSpec.STORAGE_MODE_TIERED, - ) - assert False, ( - "Transition from tiered_cloud to tiered should have been rejected" - ) - except Exception: - pass - assert ( - self._get_topic_storage_mode(rpk, "topic-tc-blocked") - == TopicSpec.STORAGE_MODE_TIERED_CLOUD - ), "Storage mode should still be tiered_cloud after rejected transition" + @cluster(num_nodes=3) + def test_tiered_cloud_topic_creation(self): + """ + Topics can be created directly in either tiered variant with an + explicit version, regardless of default_redpanda_storage_mode_tiered_impl. + """ + rpk = RpkTool(self.redpanda) - # tiered_cloud -> unset (blocked) - try: - rpk.alter_topic_config( - "topic-tc-blocked", - TopicSpec.PROPERTY_STORAGE_MODE, - TopicSpec.STORAGE_MODE_UNSET, - ) - assert False, ( - "Transition from tiered_cloud to unset should have been rejected" - ) - except Exception: - pass - assert ( - self._get_topic_storage_mode(rpk, "topic-tc-blocked") - == TopicSpec.STORAGE_MODE_TIERED_CLOUD - ), "Storage mode should still be tiered_cloud after rejected transition" + for version in ( + TopicSpec.STORAGE_MODE_IMPL_TIERED_V1, + TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, + ): + name = f"topic-created-as-{version.replace('_', '-')}" + self._create_topic(rpk, name, config=TopicSpec.storage_mode_config(version)) + assert ( + self._get_topic_storage_mode(rpk, name) == TopicSpec.STORAGE_MODE_TIERED + ), f"{name} should display storage mode 'tiered'" + assert self._get_topic_storage_mode_impl(rpk, name) == version + + +class StorageModeAliasMatrixTest(StorageModeTestBase): + """ + Full matrix of default_redpanda_storage_mode_tiered_impl values over all + redpanda.storage.mode / redpanda.storage.mode.impl inputs. + + The mode vocabulary is local/tiered/cloud/unset: 'tiered' resolves via + the default_redpanda_storage_mode_tiered_impl cluster config. The impl + property (create-only, unambiguous local/tiered_v1/tiered_v2/cloud/unset + vocabulary) selects the storage mode exactly, alone or together with a + matching mode. On describe, both tiered variants display as 'tiered' and + the read-only impl property, always present, carries the exact + implementation. + """ + + CLUSTER_CONFIG_CLOUD_STORAGE_DEFAULT_MODE = ( + "default_redpanda_storage_mode_tiered_impl" + ) + + def __init__(self, test_context: TestContext): + si_settings = SISettings( + test_context, + cloud_storage_enable_remote_read=False, + cloud_storage_enable_remote_write=False, + ) + + super(StorageModeAliasMatrixTest, self).__init__( + test_context=test_context, + num_brokers=3, + si_settings=si_settings, + ) + + def _set_default_mode(self, rpk: RpkTool, value: str): + rpk.cluster_config_set(self.CLUSTER_CONFIG_CLOUD_STORAGE_DEFAULT_MODE, value) @cluster(num_nodes=3) - def test_tiered_cloud_topic_creation(self): + def test_storage_mode_matrix(self): + rpk = RpkTool(self.redpanda) + admin = Admin(self.redpanda) + + # tiered_v1 is the default for all clusters. + default_impl = admin.get_cluster_config()[ + self.CLUSTER_CONFIG_CLOUD_STORAGE_DEFAULT_MODE + ] + assert default_impl == "tiered_v1", ( + f"clusters should default to tiered_v1, got {default_impl}" + ) + + # (mode, impl) input -> (displayed mode, displayed impl) where a None + # input means the property is not set, and DEFAULT stands for the + # current default_redpanda_storage_mode_tiered_impl. The impl is + # always present in describe output and never ambiguous. + DEFAULT = "" + accepted: list[tuple[str | None, str | None, str, str]] = [ + # mode only: 'tiered' resolves through the cluster config + ("local", None, "local", "local"), + ("tiered", None, "tiered", DEFAULT), + ("cloud", None, "cloud", "cloud"), + ("unset", None, "unset", "unset"), + # impl only: selects the storage mode exactly + (None, "tiered_v1", "tiered", "tiered_v1"), + (None, "tiered_v2", "tiered", "tiered_v2"), + (None, "local", "local", "local"), + (None, "cloud", "cloud", "cloud"), + (None, "unset", "unset", "unset"), + # both: they must agree; the impl picks the tiered variant + ("tiered", "tiered_v1", "tiered", "tiered_v1"), + ("tiered", "tiered_v2", "tiered", "tiered_v2"), + ("cloud", "cloud", "cloud", "cloud"), + ("local", "local", "local", "local"), + ] + rejected: list[tuple[str | None, str | None]] = [ + # impl names and the internal spelling are not mode values + ("tiered_v1", None), + ("tiered_v2", None), + ("tiered_cloud", None), + # the ambiguous alias and the internal spelling are not impl + # values + (None, "tiered"), + (None, "tiered_cloud"), + ("tiered", "bogus"), + # mode and impl must agree + ("local", "tiered_v1"), + ("cloud", "tiered_v2"), + ("unset", "cloud"), + ] + + for cluster_default in ("tiered_v1", "tiered_v2"): + self._set_default_mode(rpk, cluster_default) + for i, (mode, impl, want_mode, want_impl) in enumerate(accepted): + topic = f"topic-{cluster_default}-ok-{i}" + config: dict[str, str] = {} + if mode is not None: + config[TopicSpec.PROPERTY_STORAGE_MODE] = mode + if impl is not None: + config[TopicSpec.PROPERTY_STORAGE_MODE_IMPL] = impl + self._create_topic(rpk, topic, config=config) + actual_mode = self._get_topic_storage_mode(rpk, topic) + assert actual_mode == want_mode, ( + f"mode={mode} impl={impl} under {cluster_default}: " + f"expected displayed mode {want_mode}, got {actual_mode}" + ) + expected_impl = cluster_default if want_impl == DEFAULT else want_impl + actual_impl = self._get_topic_storage_mode_impl(rpk, topic) + assert actual_impl == expected_impl, ( + f"mode={mode} impl={impl} under {cluster_default}: " + f"expected impl {expected_impl}, got {actual_impl}" + ) + + for i, (mode, impl) in enumerate(rejected): + topic = f"topic-{cluster_default}-rejected-{i}" + config = {} + if mode is not None: + config[TopicSpec.PROPERTY_STORAGE_MODE] = mode + if impl is not None: + config[TopicSpec.PROPERTY_STORAGE_MODE_IMPL] = impl + try: + self._create_topic(rpk, topic, config=config) + except RpkException as e: + assert "Invalid storage mode" in str(e), ( + f"mode={mode} impl={impl} under {cluster_default}: " + f"rejected, but not by the storage-mode validator: {e}" + ) + else: + raise AssertionError( + f"mode={mode} impl={impl} should be rejected " + f"under {cluster_default}" + ) + + @cluster(num_nodes=3) + def test_variant_is_stable_across_default_flips(self): """ - Test that a topic can be created directly with tiered_cloud mode. + The variant is fixed at creation: flipping default_redpanda_storage_mode_tiered_impl + afterwards changes neither the displayed mode ('tiered' for both + variants) nor the version property. """ rpk = RpkTool(self.redpanda) + self._set_default_mode(rpk, "tiered_v1") self._create_topic( rpk, - "topic-created-as-tc", - config={ - TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_TIERED_CLOUD - }, + "topic-alias-v1", + config={TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_TIERED}, + ) + assert ( + self._get_topic_storage_mode_impl(rpk, "topic-alias-v1") + == TopicSpec.STORAGE_MODE_IMPL_TIERED_V1 + ) + + self._set_default_mode(rpk, "tiered_v2") + assert ( + self._get_topic_storage_mode(rpk, "topic-alias-v1") + == TopicSpec.STORAGE_MODE_TIERED + ) + assert ( + self._get_topic_storage_mode_impl(rpk, "topic-alias-v1") + == TopicSpec.STORAGE_MODE_IMPL_TIERED_V1 + ), "flipping the cluster default must not re-label existing topics" + + self._create_topic( + rpk, + "topic-alias-v2", + config={TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_TIERED}, ) assert ( - self._get_topic_storage_mode(rpk, "topic-created-as-tc") - == TopicSpec.STORAGE_MODE_TIERED_CLOUD - ), "Topic should be created with storage_mode=tiered_cloud" + self._get_topic_storage_mode_impl(rpk, "topic-alias-v2") + == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2 + ) diff --git a/tests/rptest/tests/tiered_cloud_local_retention_test.py b/tests/rptest/tests/tiered_cloud_local_retention_test.py index 4c499e7b8b6dd..90095d1188d25 100644 --- a/tests/rptest/tests/tiered_cloud_local_retention_test.py +++ b/tests/rptest/tests/tiered_cloud_local_retention_test.py @@ -119,7 +119,7 @@ def _create_topic( ): rpk = RpkTool(self.redpanda) config = { - TopicSpec.PROPERTY_STORAGE_MODE: storage_mode, + **TopicSpec.storage_mode_config(storage_mode), "cleanup.policy": cleanup_policy, "retention.bytes": str(self.retention_bytes), "retention.local.target.bytes": str(self.local_target_bytes), @@ -264,7 +264,7 @@ def test_local_data_grows_in_tiered_cloud(self): Replication factor is 3, so the cluster-wide sum of partition_size is roughly 3 * per-replica retention. """ - self._create_topic(storage_mode=TopicSpec.STORAGE_MODE_TIERED_CLOUD) + self._create_topic(storage_mode=TopicSpec.STORAGE_MODE_IMPL_TIERED_V2) self._wait_for_partition_info() self._produce(self.bytes_to_produce) @@ -338,7 +338,7 @@ def test_capacity_pressure_reclaims_tiered_cloud(self): partitions=1, replicas=3, config={ - TopicSpec.PROPERTY_STORAGE_MODE: (TopicSpec.STORAGE_MODE_TIERED_CLOUD), + **TopicSpec.storage_mode_config(TopicSpec.STORAGE_MODE_IMPL_TIERED_V2), "cleanup.policy": TopicSpec.CLEANUP_DELETE, "retention.bytes": str(self.retention_bytes), "segment.bytes": str(self.segment_size), @@ -399,7 +399,7 @@ def test_idle_partition_reclaims_on_pin(self): partitions=1, replicas=3, config={ - TopicSpec.PROPERTY_STORAGE_MODE: (TopicSpec.STORAGE_MODE_TIERED_CLOUD), + **TopicSpec.storage_mode_config(TopicSpec.STORAGE_MODE_IMPL_TIERED_V2), "cleanup.policy": TopicSpec.CLEANUP_DELETE, "retention.bytes": str(self.retention_bytes), "segment.bytes": str(self.segment_size), @@ -453,7 +453,7 @@ def test_compact_topic_evicts_via_l1_floor(self): # times, and L1 compaction can reduce the log meaningfully. key_set_cardinality = 64 self._create_topic( - storage_mode=TopicSpec.STORAGE_MODE_TIERED_CLOUD, + storage_mode=TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, cleanup_policy=TopicSpec.CLEANUP_COMPACT, ) self._wait_for_partition_info() @@ -511,7 +511,7 @@ def test_compact_leadership_move_advances_mash(self): """ key_set_cardinality = 64 self._create_topic( - storage_mode=TopicSpec.STORAGE_MODE_TIERED_CLOUD, + storage_mode=TopicSpec.STORAGE_MODE_IMPL_TIERED_V2, cleanup_policy=TopicSpec.CLEANUP_COMPACT, ) self._wait_for_partition_info() @@ -615,7 +615,7 @@ def test_read_below_local_start_pivots_to_l1(self): of the trimmed prefix would hit the empty local log and the consumer would fail to read the full range. """ - self._create_topic(storage_mode=TopicSpec.STORAGE_MODE_TIERED_CLOUD) + self._create_topic(storage_mode=TopicSpec.STORAGE_MODE_IMPL_TIERED_V2) self._wait_for_partition_info() assert self.redpanda is not None diff --git a/tests/rptest/tests/topic_creation_test.py b/tests/rptest/tests/topic_creation_test.py index 5ba3124fc3746..91886de2d6194 100644 --- a/tests/rptest/tests/topic_creation_test.py +++ b/tests/rptest/tests/topic_creation_test.py @@ -732,6 +732,8 @@ def test_create_topic_response_configs(self): describe_configs = [line.split() for line in res] for key, value, source in describe_configs: + # kcl appends '*' to read-only properties + key = key.rstrip("*") topic_config = self.get_config_by_name(topic_response, key) assert topic_config, (