Skip to content

Commit 20abbaa

Browse files
LazinLazin
committed
tests/rptest: cover storage mode gating and the impl property
TieredCloudUpgradeTest checks the mixed-version behavior: creating a tiered_v2 topic or converting a cloud topic to it is rejected (or, via a v26.1 controller that drops the unknown impl property, degrades to a classic tiered topic) while the cluster is partially upgraded, and both operations work once every node runs v26.2. StorageModeAliasMatrixTest covers every mode/impl input combination under both tiered impls, including impl-only creation, mode/impl consistency rejections, read-only enforcement and the always-present describe entry. ShadowLinkStorageModeSyncTest syncs cloud and tiered topics across all four combinations of source/target tiered impls, including a mid-flight cloud to tiered_v2 conversion. TopicSpec spells tiered variants through the new storage_mode_config helper (mode=tiered plus the impl property, which keeps v26.1 brokers on a tiered mode instead of silently falling back to the cluster default), kafka-cli describe output round-trips through the new impl kwarg, and the describe-topics parser accepts the asterisk kcl appends to read-only properties. The storage-mode toggle and rnot flipping tests point the 'tiered' alias at tiered_v2 explicitly, since the alter path resolves it through the cluster config. Signed-off-by: Evgeny Lazin <4lazin@gmail.com>
1 parent ba9d9a2 commit 20abbaa

18 files changed

Lines changed: 800 additions & 169 deletions

‎tests/rptest/clients/types.py‎

Lines changed: 31 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,9 +27,16 @@ class TopicSpec:
2727
STORAGE_MODE_LOCAL = "local"
2828
STORAGE_MODE_TIERED = "tiered"
2929
STORAGE_MODE_CLOUD = "cloud"
30-
STORAGE_MODE_TIERED_CLOUD = "tiered_cloud"
3130
STORAGE_MODE_UNSET = "unset"
3231

32+
# Tiered values of the read-only redpanda.storage.mode.impl property,
33+
# whose vocabulary is the unambiguous
34+
# [unset|local|tiered_v1|tiered_v2|cloud]. On creation the property (set
35+
# alone or together with a matching redpanda.storage.mode) selects the
36+
# storage mode exactly. Not valid as redpanda.storage.mode values.
37+
STORAGE_MODE_IMPL_TIERED_V1 = "tiered_v1"
38+
STORAGE_MODE_IMPL_TIERED_V2 = "tiered_v2"
39+
3340
PROPERTY_COMPRESSSION = "compression.type"
3441
PROPERTY_CLEANUP_POLICY = "cleanup.policy"
3542
PROPERTY_COMPACTION_STRATEGY = "compaction.strategy"
@@ -57,6 +64,24 @@ class TopicSpec:
5764
PROPERTY_REMOTE_READ = "redpanda.remote.read"
5865
PROPERTY_REMOTE_WRITE = "redpanda.remote.write"
5966
PROPERTY_STORAGE_MODE = "redpanda.storage.mode"
67+
PROPERTY_STORAGE_MODE_IMPL = "redpanda.storage.mode.impl"
68+
69+
@staticmethod
70+
def storage_mode_config(mode: str) -> dict[str, str]:
71+
"""Topic config dict selecting a storage mode. Tiered variants
72+
(tiered_v1/tiered_v2) are spelled as redpanda.storage.mode=tiered plus
73+
the redpanda.storage.mode.impl property; sending both keeps v26.1
74+
brokers (which ignore the unknown impl property) on a tiered mode
75+
rather than silently falling back to the cluster default."""
76+
if mode in (
77+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V1,
78+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V2,
79+
):
80+
return {
81+
TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_TIERED,
82+
TopicSpec.PROPERTY_STORAGE_MODE_IMPL: mode,
83+
}
84+
return {TopicSpec.PROPERTY_STORAGE_MODE: mode}
6085

6186
class CompressionTypes(str, Enum):
6287
"""
@@ -153,6 +178,7 @@ def __init__(
153178
min_compaction_lag_ms: int | None = None,
154179
max_compaction_lag_ms: int | None = None,
155180
redpanda_storage_mode: str | None = None,
181+
redpanda_storage_mode_impl: str | None = None,
156182
):
157183
self.name = name or f"topic-{self._random_topic_suffix()}"
158184
self.partition_count = partition_count
@@ -192,6 +218,10 @@ def __init__(
192218
self.min_compaction_lag_ms = min_compaction_lag_ms
193219
self.max_compaction_lag_ms = max_compaction_lag_ms
194220
self.redpanda_storage_mode = redpanda_storage_mode
221+
# Read-only on the broker (reported by DescribeConfigs for every
222+
# topic); accepted here so describe output can round-trip through
223+
# TopicSpec.
224+
self.redpanda_storage_mode_impl = redpanda_storage_mode_impl
195225

196226
def __str__(self):
197227
return self.name

‎tests/rptest/tests/cloud_topics/compaction_stress_test.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -185,7 +185,7 @@ def setUp(self):
185185
storage_mode = (self.test_context.injected_args or {}).get(
186186
"storage_mode", TopicSpec.STORAGE_MODE_CLOUD
187187
)
188-
if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD:
188+
if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2:
189189
self.redpanda.set_feature_active(
190190
"tiered_cloud_topics", True, timeout_sec=30
191191
)
@@ -194,7 +194,7 @@ def setUp(self):
194194
partitions=1,
195195
replicas=3,
196196
config={
197-
TopicSpec.PROPERTY_STORAGE_MODE: storage_mode,
197+
**TopicSpec.storage_mode_config(storage_mode),
198198
"cleanup.policy": TopicSpec.CLEANUP_COMPACT,
199199
"min.cleanable.dirty.ratio": "0.0",
200200
},
@@ -204,7 +204,7 @@ def setUp(self):
204204
@matrix(
205205
storage_mode=[
206206
TopicSpec.STORAGE_MODE_CLOUD,
207-
TopicSpec.STORAGE_MODE_TIERED_CLOUD,
207+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V2,
208208
],
209209
)
210210
def test_key_cardinality_overflow(self, storage_mode: str):

‎tests/rptest/tests/cloud_topics/debug_rows_test.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -169,7 +169,7 @@ def read_objects(
169169
@matrix(
170170
storage_mode=[
171171
TopicSpec.STORAGE_MODE_CLOUD,
172-
TopicSpec.STORAGE_MODE_TIERED_CLOUD,
172+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V2,
173173
],
174174
)
175175
def test_read_and_write_rows(self, storage_mode: str) -> None:

‎tests/rptest/tests/cloud_topics/e2e_test.py‎

Lines changed: 18 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -100,13 +100,13 @@ def setUp(self):
100100
storage_mode = (self.test_context.injected_args or {}).get(
101101
"storage_mode", TopicSpec.STORAGE_MODE_CLOUD
102102
)
103-
if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD:
103+
if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2:
104104
self.redpanda.set_feature_active(
105105
"tiered_cloud_topics", True, timeout_sec=30
106106
)
107107
for topic in self.topics:
108108
config = {
109-
TopicSpec.PROPERTY_STORAGE_MODE: storage_mode,
109+
**TopicSpec.storage_mode_config(storage_mode),
110110
"cleanup.policy": topic.cleanup_policy,
111111
}
112112
if topic.min_cleanable_dirty_ratio is not None:
@@ -383,7 +383,7 @@ def await_num_produced(self, min_records, timeout_sec=120):
383383
@matrix(
384384
storage_mode=[
385385
TopicSpec.STORAGE_MODE_CLOUD,
386-
TopicSpec.STORAGE_MODE_TIERED_CLOUD,
386+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V2,
387387
],
388388
)
389389
def test_write(self, storage_mode: str):
@@ -400,7 +400,7 @@ def test_write(self, storage_mode: str):
400400
@matrix(
401401
storage_mode=[
402402
TopicSpec.STORAGE_MODE_CLOUD,
403-
TopicSpec.STORAGE_MODE_TIERED_CLOUD,
403+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V2,
404404
],
405405
)
406406
def test_delete_records(self, storage_mode: str):
@@ -430,7 +430,7 @@ def test_delete_records(self, storage_mode: str):
430430
@matrix(
431431
storage_mode=[
432432
TopicSpec.STORAGE_MODE_CLOUD,
433-
TopicSpec.STORAGE_MODE_TIERED_CLOUD,
433+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V2,
434434
],
435435
)
436436
def test_get_size(self, storage_mode: str):
@@ -472,9 +472,10 @@ def get_partition_size() -> int | None:
472472

473473

474474
class EndToEndCloudTopicsStorageModeToggleTest(EndToEndCloudTopicsBase):
475-
"""Exercise toggling a topic between 'cloud' and 'tiered_cloud' storage
476-
modes while a rate-limited producer is running, then validate the
477-
resulting log with a sequential consumer."""
475+
"""Exercise toggling a topic between 'cloud' and 'tiered' storage modes
476+
(the latter resolves to tiered_v2 via the cluster default) while a
477+
rate-limited producer is running, then validate the resulting log with a
478+
sequential consumer."""
478479

479480
topics = (
480481
TopicSpec(
@@ -522,6 +523,13 @@ def test_toggle_storage_mode(self):
522523
loop=False,
523524
producer=producer,
524525
)
526+
# The toggle flips through the 'tiered' alias; point it at the
527+
# cloud-architecture variant (the default is tiered_v1, under which
528+
# cloud -> tiered is a forbidden transition).
529+
self.rpk.cluster_config_set(
530+
"default_redpanda_storage_mode_tiered_impl", "tiered_v2"
531+
)
532+
525533
try:
526534
producer.start()
527535

@@ -530,7 +538,7 @@ def test_toggle_storage_mode(self):
530538
while time.time() - start < self.toggle_duration_sec:
531539
time.sleep(self.toggle_interval_sec)
532540
mode = (
533-
TopicSpec.STORAGE_MODE_TIERED_CLOUD
541+
TopicSpec.STORAGE_MODE_TIERED
534542
if mode == TopicSpec.STORAGE_MODE_CLOUD
535543
else TopicSpec.STORAGE_MODE_CLOUD
536544
)
@@ -624,7 +632,7 @@ def start_consumer_with_tx(self):
624632
@matrix(
625633
storage_mode=[
626634
TopicSpec.STORAGE_MODE_CLOUD,
627-
TopicSpec.STORAGE_MODE_TIERED_CLOUD,
635+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V2,
628636
],
629637
)
630638
def test_write(self, storage_mode: str):

‎tests/rptest/tests/cloud_topics/iceberg_test.py‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -124,7 +124,7 @@ def __init__(self, test_ctx: TestContext):
124124
cloud_storage_type=supported_storage_types(),
125125
storage_mode=[
126126
TopicSpec.STORAGE_MODE_CLOUD,
127-
TopicSpec.STORAGE_MODE_TIERED_CLOUD,
127+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V2,
128128
],
129129
)
130130
def test_compaction_preserves_all_offsets_in_iceberg(
@@ -142,15 +142,15 @@ def test_compaction_preserves_all_offsets_in_iceberg(
142142
include_query_engines=[QueryEngineType.SPARK],
143143
catalog_type=CatalogType.REST_JDBC,
144144
) as dl:
145-
if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD:
145+
if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2:
146146
self.redpanda.set_feature_active(
147147
"tiered_cloud_topics", True, timeout_sec=30
148148
)
149149
dl.create_iceberg_enabled_topic(
150150
self.topic_name,
151151
iceberg_mode="key_value",
152152
config={
153-
TopicSpec.PROPERTY_STORAGE_MODE: storage_mode,
153+
**TopicSpec.storage_mode_config(storage_mode),
154154
"cleanup.policy": TopicSpec.CLEANUP_COMPACT,
155155
},
156156
)
@@ -230,7 +230,7 @@ def __init__(self, test_ctx: TestContext) -> None:
230230
cloud_storage_type=supported_storage_types(),
231231
storage_mode=[
232232
TopicSpec.STORAGE_MODE_CLOUD,
233-
TopicSpec.STORAGE_MODE_TIERED_CLOUD,
233+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V2,
234234
],
235235
)
236236
def test_deletion_blocked_until_translated(
@@ -248,15 +248,15 @@ def test_deletion_blocked_until_translated(
248248
include_query_engines=[QueryEngineType.SPARK],
249249
catalog_type=CatalogType.REST_JDBC,
250250
) as dl:
251-
if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD:
251+
if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2:
252252
self.redpanda.set_feature_active(
253253
"tiered_cloud_topics", True, timeout_sec=30
254254
)
255255
dl.create_iceberg_enabled_topic(
256256
self.topic_name,
257257
iceberg_mode="key_value",
258258
config={
259-
TopicSpec.PROPERTY_STORAGE_MODE: storage_mode,
259+
**TopicSpec.storage_mode_config(storage_mode),
260260
"retention.ms": 500,
261261
"retention.bytes": 1024,
262262
},

‎tests/rptest/tests/cloud_topics/retention_test.py‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,7 @@ def _verify_offset_consumable(self, topic: str, offset: int):
128128
cloud_storage_type=get_cloud_storage_type(),
129129
storage_mode=[
130130
TopicSpec.STORAGE_MODE_CLOUD,
131-
TopicSpec.STORAGE_MODE_TIERED_CLOUD,
131+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V2,
132132
],
133133
)
134134
def test_size_based_retention(
@@ -152,7 +152,7 @@ def test_size_based_retention(
152152
partitions=1,
153153
replicas=3,
154154
config={
155-
TopicSpec.PROPERTY_STORAGE_MODE: storage_mode,
155+
**TopicSpec.storage_mode_config(storage_mode),
156156
"cleanup.policy": TopicSpec.CLEANUP_DELETE,
157157
"retention.bytes": str(initial_retention),
158158
},
@@ -223,7 +223,7 @@ def test_size_based_retention(
223223
cloud_storage_type=get_cloud_storage_type(),
224224
storage_mode=[
225225
TopicSpec.STORAGE_MODE_CLOUD,
226-
TopicSpec.STORAGE_MODE_TIERED_CLOUD,
226+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V2,
227227
],
228228
)
229229
def test_time_based_retention(
@@ -244,7 +244,7 @@ def test_time_based_retention(
244244
partitions=1,
245245
replicas=3,
246246
config={
247-
TopicSpec.PROPERTY_STORAGE_MODE: storage_mode,
247+
**TopicSpec.storage_mode_config(storage_mode),
248248
"cleanup.policy": TopicSpec.CLEANUP_DELETE,
249249
"retention.ms": str(3600000), # 1 hour - data won't be deleted yet
250250
},

‎tests/rptest/tests/cloud_topics_test.py‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ def __create_initial_topics(self, storage_mode):
3333
Create initial test topics. Cloud topics are gated by cloud storage,
3434
which is already configured via SISettings.
3535
"""
36-
if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD:
36+
if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2:
3737
self.redpanda.set_feature_active(
3838
"tiered_cloud_topics", True, timeout_sec=30
3939
)
@@ -43,7 +43,7 @@ def __create_initial_topics(self, storage_mode):
4343
spec.name,
4444
spec.partition_count,
4545
spec.replication_factor,
46-
config={TopicSpec.PROPERTY_STORAGE_MODE: storage_mode},
46+
config=TopicSpec.storage_mode_config(storage_mode),
4747
)
4848

4949
# Ignored because it's flaky but the test is still useful locally.
@@ -53,7 +53,7 @@ def __create_initial_topics(self, storage_mode):
5353
cloud_storage_type=get_cloud_storage_type(),
5454
storage_mode=[
5555
TopicSpec.STORAGE_MODE_CLOUD,
56-
TopicSpec.STORAGE_MODE_TIERED_CLOUD,
56+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V2,
5757
],
5858
)
5959
def test_reconciler_uploads(self, cloud_storage_type, storage_mode):
@@ -76,7 +76,7 @@ def count_objects(prefix):
7676
err_msg=lambda: f"failed to find at least 1 l1 object(s), instead got {count_objects('l1_')}",
7777
)
7878

79-
if storage_mode == TopicSpec.STORAGE_MODE_TIERED_CLOUD:
79+
if storage_mode == TopicSpec.STORAGE_MODE_IMPL_TIERED_V2:
8080
# In tiered_cloud mode, data is replicated through raft (no L0
8181
# uploads). Verify no L0 objects were created.
8282
l0_count = count_objects("l0_")

‎tests/rptest/tests/cluster_features_test.py‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1461,8 +1461,9 @@ def _perturb_v26_1_to_v26_2(self, phase):
14611461
# (batch_mirror_topic_status additionally needs a second cluster).
14621462

14631463
def _exercise_tiered_cloud_topics(self):
1464-
"""tiered_cloud_topics gate: creating a topic with storage mode
1465-
`tiered_cloud` is refused until the feature is active. Drive that
1464+
"""tiered_cloud_topics gate: creating a topic with the tiered_v2
1465+
(cloud-architecture) storage mode is refused until the feature is
1466+
active. Drive that
14661467
validator path while unfinalized and confirm the feature is unavailable
14671468
and the create is rejected. The topic is never created, so exercising
14681469
the gate cannot leave state that would block a downgrade."""
@@ -1473,9 +1474,9 @@ def _exercise_tiered_cloud_topics(self):
14731474
RpkTool(self.redpanda).create_topic(
14741475
"perturb-tiered-cloud",
14751476
partitions=1,
1476-
config={
1477-
TopicSpec.PROPERTY_STORAGE_MODE: TopicSpec.STORAGE_MODE_TIERED_CLOUD
1478-
},
1477+
config=TopicSpec.storage_mode_config(
1478+
TopicSpec.STORAGE_MODE_IMPL_TIERED_V2
1479+
),
14791480
)
14801481
except RpkException as e:
14811482
# Confirm the create failed via the storage-mode gate, not an

0 commit comments

Comments
 (0)