@@ -398,30 +398,42 @@ disk_log_impl::time_based_gc_max_offset(gc_config cfg) const {
398398 : _segs.front ()->index ().retention_timestamp (retention_cfg),
399399 retention_cfg);
400400
401+ auto validated_timestamps = _feature_table.local ().is_active (
402+ features::feature::validated_batch_timestamps);
401403 static constexpr auto const_threshold = 1min;
402404 auto bogus_threshold = model::timestamp (
403405 model::timestamp::now ().value () + const_threshold / 1ms);
404406
405407 auto it = std::find_if (
406408 std::cbegin (_segs),
407409 std::cend (_segs),
408- [this , time = cfg.eviction_time , bogus_threshold, retention_cfg](
409- const ss::lw_shared_ptr<segment>& s) {
410+ [this ,
411+ time = cfg.eviction_time ,
412+ bogus_threshold,
413+ retention_cfg,
414+ validated_timestamps](const ss::lw_shared_ptr<segment>& s) {
410415 auto retention_ts = s->index ().retention_timestamp (retention_cfg);
411416
412- if (retention_ts > bogus_threshold) {
413- // Warn on timestamps more than the "bogus" threshold in future
414- // this should not fire for segments created after v23.3
415- vlog (
416- gclog.warn ,
417- " [{}] found segment with bogus retention timestamp: {} (base "
418- " {}, max {}) - {}" ,
419- config ().ntp (),
420- retention_ts,
421- s->index ().base_timestamp (),
422- s->index ().max_timestamp (),
423- s->index ().broker_timestamp (),
424- s);
417+ if (!validated_timestamps) {
418+ // This legacy behavior of warning on timestamps above
419+ // `bogus_threshold` is only needed in older clusters that didn't
420+ // validate timestamps.
421+ if (retention_ts > bogus_threshold) {
422+ // Warn on timestamps more than the "bogus" threshold in
423+ // future this should not fire for segments created after
424+ // v23.3
425+ vlog (
426+ gclog.warn ,
427+ " [{}] found segment with bogus retention timestamp: {} "
428+ " (base "
429+ " {}, max {}) - {}" ,
430+ config ().ntp (),
431+ retention_ts,
432+ s->index ().base_timestamp (),
433+ s->index ().max_timestamp (),
434+ s->index ().broker_timestamp (),
435+ s);
436+ }
425437 }
426438
427439 // first that is not going to be collected
@@ -1776,6 +1788,13 @@ ss::future<std::optional<model::offset>> disk_log_impl::do_gc(gc_config cfg) {
17761788}
17771789
17781790ss::future<> disk_log_impl::maybe_adjust_retention_timestamps () {
1791+ auto validated_timestamps = _feature_table.local ().is_active (
1792+ features::feature::validated_batch_timestamps);
1793+ if (validated_timestamps) {
1794+ // Legacy behavior of adjusting retention timestamps is only needed in
1795+ // older clusters that didn't validate timestamps.
1796+ co_return ;
1797+ }
17791798 // Correct any timestamps too far in the future, meant to be called before
17801799 // calculating the retention offset for garbage collection.
17811800 // It's expected that this will be used only for segments pre v23.3,
0 commit comments