Skip to content

Commit 77eedf2

Browse files
WillemKaufWillemKauf
committed
storage: use validated max_ts for retention_ms
The motivating case for `broker_time_based_retention` was the fact that records with bad timestamps produced in the future could lead to time-based retention being stuck indefinitely [1]. However, using the `broker_ts` can lead to unexpected behavior when e.g. replicating data from an existing cluster using MM2, as the timestamps of the Kafka records themselves are correctly preserved, but internally, `redpanda` data structures are not. To avoid the potentially curious behavior of a divergence in retention enforcement, go back to using the `max_timestamp` for batches whose timestamps have been validated and unconditionally set in the produce path (see: `v/kafka/server/handlers/produce_validation.cc`) as of `v25.3.1`. [1]: * #9820 * #12991
1 parent 977bf57 commit 77eedf2

10 files changed

Lines changed: 194 additions & 93 deletions

‎src/v/cluster/archival/tests/ntp_archiver_reupload_test.cc‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -129,7 +129,8 @@ struct reupload_fixture : public archiver_fixture {
129129
10,
130130
1_MiB)
131131
.get();
132-
write_random_batches(segment, seg.num_records.value(), 2);
132+
write_random_batches(
133+
segment, seg.num_records.value(), 2, model::timestamp::min());
133134
disk_log_impl()->segments().add(segment);
134135
}
135136

@@ -540,7 +541,7 @@ FIXTURE_TEST(test_upload_both_compacted_and_non_compacted, reupload_fixture) {
540541
// Close current segment and add a new open segment, so both compacted and
541542
// non-compacted uploads run together.
542543
auto& last_segment = disk_log_impl()->segments().back();
543-
write_random_batches(last_segment, 20, 2);
544+
write_random_batches(last_segment, 20, 2, model::timestamp::min());
544545
last_segment->appender().close().get();
545546
last_segment->release_appender();
546547
add_segment_bytes(last_segment, last_segment->size_bytes());
@@ -610,7 +611,7 @@ FIXTURE_TEST(test_both_uploads_with_one_failing, reupload_fixture) {
610611
// Close current segment and add a new open segment, so both compacted and
611612
// non-compacted uploads run together.
612613
auto& last_segment = disk_log_impl()->segments().back();
613-
write_random_batches(last_segment, 20, 2);
614+
write_random_batches(last_segment, 20, 2, model::timestamp::min());
614615
last_segment->appender().close().get();
615616
last_segment->release_appender();
616617
add_segment_bytes(last_segment, last_segment->size_bytes());
@@ -798,7 +799,7 @@ FIXTURE_TEST(test_upload_limit, reupload_fixture) {
798799
// NOTE: uploaded 4 segments, so offset is 54 at the start
799800
for (auto i = 0; i < 3; ++i) {
800801
auto& last_segment = disk_log_impl()->segments().back();
801-
write_random_batches(last_segment, 10, 2);
802+
write_random_batches(last_segment, 10, 2, model::timestamp::min());
802803
last_segment->appender().close().get();
803804
last_segment->release_appender();
804805
add_segment_bytes(last_segment, last_segment->size_bytes());

‎src/v/cluster/archival/tests/ntp_archiver_test.cc‎

Lines changed: 15 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -416,18 +416,16 @@ FIXTURE_TEST(test_retention, archiver_fixture) {
416416
.term = model::term_id(2),
417417
.num_records = 1000,
418418
.timestamp = old_stamp},
419-
{
420-
.ntp = manifest_ntp,
421-
.base_offset = model::offset(2000),
422-
.term = model::term_id(3),
423-
.num_records = 1000,
424-
},
425-
{
426-
.ntp = manifest_ntp,
427-
.base_offset = model::offset(3000),
428-
.term = model::term_id(4),
429-
.num_records = 1000,
430-
}};
419+
{.ntp = manifest_ntp,
420+
.base_offset = model::offset(2000),
421+
.term = model::term_id(3),
422+
.num_records = 1000,
423+
.timestamp = model::timestamp::now()},
424+
{.ntp = manifest_ntp,
425+
.base_offset = model::offset(3000),
426+
.term = model::term_id(4),
427+
.num_records = 1000,
428+
.timestamp = model::timestamp::now()}};
431429

432430
init_storage_api_local(segments);
433431
vlog(test_log.info, "Initialized, start waiting for partition leadership");
@@ -732,12 +730,11 @@ FIXTURE_TEST(test_segments_pending_deletion_limit, archiver_fixture) {
732730
.term = model::term_id(3),
733731
.num_records = 1000,
734732
.timestamp = old_stamp},
735-
{
736-
.ntp = manifest_ntp,
737-
.base_offset = model::offset(3000),
738-
.term = model::term_id(4),
739-
.num_records = 1000,
740-
}};
733+
{.ntp = manifest_ntp,
734+
.base_offset = model::offset(3000),
735+
.term = model::term_id(4),
736+
.num_records = 1000,
737+
.timestamp = model::timestamp::now()}};
741738

742739
init_storage_api_local(segments);
743740

‎src/v/cluster/archival/tests/service_fixture.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ struct segment_desc {
3636
model::term_id term;
3737
std::optional<size_t> num_records;
3838
std::optional<size_t> records_per_batch;
39-
std::optional<model::timestamp> timestamp;
39+
std::optional<model::timestamp> timestamp = model::timestamp::min();
4040
};
4141

4242
struct offset_range {

‎src/v/storage/segment_index.h‎

Lines changed: 52 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -36,39 +36,63 @@ using broker_timestamp_t = ss::lowres_system_clock::time_point;
3636

3737
// clang-format off
3838
// this truth table shows which timestamps gets used as retention_timestamp
39+
// `validated_batch_timestamps` ensures that the `max_timestamp` of a batch
40+
// produced to `redpanda` has been validated by properties
41+
// `message.timestamp.{before/after}.max.ms` and are valid for retention enforcement.
42+
// Prior to this, the broker timestamp would be used for retention enforcement,
43+
// and even prior to this, the unvalidated `max_timestamp` would be used.
3944
//
40-
// use_broker_ts has_broker_ts ignore_future_ts has_alternative_to_future_ts retention_ts
45+
// use_broker_ts has_broker_ts ignore_future_ts has_alternative_to_future_ts validated_batch_timestamps retention_ts
4146
// (likely false)
42-
// TRUE TRUE TRUE TRUE broker_timestamp
43-
// TRUE TRUE TRUE FALSE broker_timestamp
44-
// TRUE TRUE FALSE TRUE broker_timestamp
45-
// TRUE TRUE FALSE FALSE broker_timestamp // new segment, new cluster
46-
// TRUE FALSE TRUE TRUE segment_index::_retention_ms // buggy old segment, new cluster
47-
// TRUE FALSE TRUE FALSE max_timestamp
48-
// TRUE FALSE FALSE TRUE max_timestamp
49-
// TRUE FALSE FALSE FALSE max_timestamp // old segment, new cluster
50-
// FALSE TRUE TRUE TRUE segment_index::_retention_ms
51-
// FALSE TRUE TRUE FALSE max_timestamp
52-
// FALSE TRUE FALSE TRUE max_timestamp
53-
// FALSE TRUE FALSE FALSE max_timestamp // new segment, upgraded cluster
54-
// FALSE FALSE TRUE TRUE segment_index::_retention_ms // buggy old segments
55-
// FALSE FALSE TRUE FALSE max_timestamp
56-
// FALSE FALSE FALSE TRUE max_timestamp
57-
// FALSE FALSE FALSE FALSE max_timestamp // old segment, upgraded cluster
47+
// TRUE TRUE TRUE TRUE TRUE max_timestamp
48+
// TRUE TRUE TRUE FALSE TRUE max_timestamp
49+
// TRUE TRUE FALSE TRUE TRUE max_timestamp
50+
// TRUE TRUE FALSE FALSE TRUE max_timestamp
51+
// TRUE FALSE TRUE TRUE TRUE max_timestamp
52+
// TRUE FALSE TRUE FALSE TRUE max_timestamp
53+
// TRUE FALSE FALSE TRUE TRUE max_timestamp
54+
// TRUE FALSE FALSE FALSE TRUE max_timestamp
55+
// TRUE TRUE TRUE TRUE FALSE broker_timestamp
56+
// TRUE TRUE TRUE FALSE FALSE broker_timestamp
57+
// TRUE TRUE FALSE TRUE FALSE broker_timestamp
58+
// TRUE TRUE FALSE FALSE FALSE broker_timestamp // new segment, new cluster
59+
// TRUE FALSE TRUE TRUE FALSE segment_index::_retention_ms // buggy old segment, new cluster
60+
// TRUE FALSE TRUE FALSE FALSE max_timestamp
61+
// TRUE FALSE FALSE TRUE FALSE max_timestamp
62+
// TRUE FALSE FALSE FALSE FALSE max_timestamp // old segment, new cluster
63+
// FALSE TRUE TRUE TRUE FALSE segment_index::_retention_ms
64+
// FALSE TRUE TRUE FALSE FALSE max_timestamp
65+
// FALSE TRUE FALSE TRUE FALSE max_timestamp
66+
// FALSE TRUE FALSE FALSE FALSE max_timestamp // new segment, upgraded cluster
67+
// FALSE FALSE TRUE TRUE FALSE segment_index::_retention_ms // buggy old segments
68+
// FALSE FALSE TRUE FALSE FALSE max_timestamp
69+
// FALSE FALSE FALSE TRUE FALSE max_timestamp
70+
// FALSE FALSE FALSE FALSE FALSE max_timestamp // old segment, upgraded cluster
71+
// FALSE TRUE TRUE TRUE TRUE max_timestamp
72+
// FALSE TRUE TRUE FALSE TRUE max_timestamp
73+
// FALSE TRUE FALSE TRUE TRUE max_timestamp
74+
// FALSE TRUE FALSE FALSE TRUE max_timestamp
75+
// FALSE FALSE TRUE TRUE TRUE max_timestamp
76+
// FALSE FALSE TRUE FALSE TRUE max_timestamp
77+
// FALSE FALSE FALSE TRUE TRUE max_timestamp
78+
// FALSE FALSE FALSE FALSE TRUE max_timestamp
5879
// clang-format on
5980

6081
// this struct is meant to be a local copy of the feature
6182
// broker_time_based_retention and configuration property
6283
// storage_ignore_timestamps_in_future_secs
6384
struct time_based_retention_cfg {
6485
bool use_broker_time;
86+
bool use_validated_batch_time;
6587
bool use_escape_hatch_for_timestamps_in_the_future;
6688

6789
static auto make(const features::feature_table& ft)
6890
-> time_based_retention_cfg {
6991
return {
7092
.use_broker_time = ft.is_active(
7193
features::feature::broker_time_based_retention),
94+
.use_validated_batch_time = ft.is_active(
95+
features::feature::validated_batch_timestamps),
7296
.use_escape_hatch_for_timestamps_in_the_future
7397
= config::shard_local_cfg()
7498
.storage_ignore_timestamps_in_future_sec()
@@ -81,7 +105,14 @@ struct time_based_retention_cfg {
81105
std::optional<model::timestamp> broker_ts,
82106
model::timestamp max_ts,
83107
std::optional<model::timestamp> alternative_retention_ts) const noexcept {
84-
// new clusters and new segments should hit this branch
108+
// For versions of `redpanda` that have `max_timestamp` in batches
109+
// validated by `message.timestamp.{before/after}.max.ms` and
110+
// unconditionally set in the produce path (see:
111+
// v/kafka/server/handlers/produce_validation.cc)
112+
if (use_validated_batch_time) {
113+
return max_ts;
114+
}
115+
85116
if (likely(use_broker_time && broker_ts.has_value())) {
86117
return *broker_ts;
87118
}
@@ -350,8 +381,10 @@ struct fmt::formatter<storage::time_based_retention_cfg>
350381
: public fmt::formatter<std::string_view> {
351382
auto format(const storage::time_based_retention_cfg& cfg, auto& ctx) const {
352383
auto str = ssx::sformat(
353-
"[.use_broker_time={}, "
384+
"[.use_validated_batch_time={}, "
385+
".use_broker_time={}, "
354386
".use_escape_hatch_for_timestamps_in_the_future={}]",
387+
cfg.use_validated_batch_time,
355388
cfg.use_broker_time,
356389
cfg.use_escape_hatch_for_timestamps_in_the_future);
357390
return formatter<std::string_view>::format(

‎src/v/storage/tests/log_truncate_test.cc‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -279,6 +279,7 @@ TEST_F(storage_test_fixture, test_truncate_last_single_record_batch) {
279279
log,
280280
15,
281281
model::term_id(0),
282+
std::nullopt,
282283
[](std::optional<model::timestamp> ts = std::nullopt) {
283284
chunked_circular_buffer<model::record_batch> ret;
284285
ret.push_back(
@@ -544,9 +545,10 @@ TEST_F(storage_test_fixture, test_concurrent_truncate_and_compaction) {
544545
storage::ntp_config(
545546
ntp, mgr.config().base_dir, std::move(overrides)))
546547
.get();
548+
auto ts = model::timestamp::now() - model::timestamp(1000);
547549
for (int seg = 0; seg < 2; seg++) {
548550
for (int i = 0; i < 5; i++) {
549-
append_random_batches(log, 1, model::term_id(0));
551+
append_random_batches(log, 1, model::term_id(0), ts);
550552
}
551553
log->flush().get();
552554
log->force_roll().get();
@@ -577,7 +579,6 @@ TEST_F(storage_test_fixture, test_concurrent_truncate_and_compaction) {
577579
}
578580

579581
// Now race windowed compaction and truncation.
580-
auto ts = now();
581582
auto sleep_ms1 = random_generators::get_int(0, 100);
582583
housekeeping_config housekeeping_cfg(
583584
ts,

‎src/v/storage/tests/segment_concatenation_test.cc‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
#include "bytes/iostream.h"
1212
#include "model/record.h"
1313
#include "model/record_batch_types.h"
14+
#include "model/timestamp.h"
1415
#include "random/generators.h"
1516
#include "storage/disk_log_impl.h"
1617
#include "storage/parser.h"
@@ -206,11 +207,18 @@ TEST_P(MakeConcatenatedSegmentFixture, ConcatenateSegments) {
206207

207208
auto [num_segments, maybe_compress] = GetParam();
208209
auto records_per_seg = random_generators::get_int(50, 200);
210+
model::timestamp base_ts = model::timestamp::min();
209211
for (size_t i = 0; i < num_segments; ++i) {
210212
auto offset = i * records_per_seg;
211213
b
212214
| add_random_batch(
213-
offset, records_per_seg, maybe_compress_batches(maybe_compress));
215+
offset,
216+
records_per_seg,
217+
maybe_compress_batches(maybe_compress),
218+
model::record_batch_type::raft_data,
219+
append_config(),
220+
disk_log_builder::should_flush_after::yes,
221+
base_ts);
214222
disk_log.force_roll().get();
215223
}
216224

‎src/v/storage/tests/segment_deduplication_test.cc‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,13 @@ void add_segments(
5353
auto offset = start_offset + i * records_per_seg;
5454
b | add_segment(offset)
5555
| add_random_batch(
56-
offset, records_per_seg, maybe_compress_batches::yes);
56+
offset,
57+
records_per_seg,
58+
maybe_compress_batches::yes,
59+
model::record_batch_type::raft_data,
60+
append_config(),
61+
disk_log_builder::should_flush_after::yes,
62+
model::timestamp::min());
5763
}
5864
for (auto& seg : disk_log.segments()) {
5965
if (mark_compacted) {

0 commit comments

Comments
 (0)