storage: fix compaction minting future record timestamps

copy_data_segment_reducer::filter() re-encodes a batch when some of its
records are compacted away. It advanced the new batch's first_timestamp
to the first surviving record's absolute timestamp, but did NOT rebase
the per-record timestamp deltas, which remain relative to the batch's
original first_timestamp. On read, every surviving record therefore
resolves to new_first_timestamp + delta, i.e. shifted forward by a
constant (the first surviving record's original delta).

This is only non-zero when the batch's base record (delta 0) is dropped
(e.g. a tombstone), and the shift is large and positive when the batch
is out of order (the dropped base record is much older than the first
surviving record) -- turning strictly-past records into records with
future timestamps. This breaks broker timestamp validation
(log.message.timestamp.after.max.ms) and freezes downstream Kafka
Streams monotonic-max KTable aggregates in the future. Replicas compact
independently, so the same offset can carry different timestamps on
different brokers. Offsets are unaffected because base_offset is
preserved on compaction; this restores the same symmetry for timestamps.

Fix: preserve the original first_timestamp (so the unchanged per-record
deltas stay valid) and compute max_timestamp from the greatest surviving
delta (out-of-order safe), instead of assuming the last record is the
max.

Reproduced on v26.1.6 and v26.1.12 (compact topic + out-of-order
timestamps + tombstones): records produced strictly in the past read
back with timestamps hundreds of days in the future after compaction.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Aangbaeck <3142272+Aangbaeck@users.noreply.github.com>

Author: Aangbaeck

src/v/storage/compaction_reducers.cc

@@ -230,16 +230,19 @@ copy_data_segment_reducer::filter(model::record_batch batch) {
     // 4. filter
     iobuf ret;
     int32_t rec_count = 0;
-    std::optional<int64_t> first_timestamp_delta;
-    int64_t last_timestamp_delta;
+    // Track the MAX surviving timestamp delta. Records may be out of order, so the last
+    // surviving record is not necessarily the one with the greatest timestamp. The per-record
+    // deltas are appended UNCHANGED (relative to the batch's original first_timestamp, which is
+    // preserved below) -- mirroring how base_offset is preserved so that offset deltas stay
+    // valid across compaction.
+    std::optional<int64_t> max_timestamp_delta;
     // We expect and enforce that offset_deltas is sorted.
     dassert(
       std::ranges::is_sorted(offset_deltas),
       "offset_deltas must be ascending in record-iteration order");
     size_t keep_idx = 0;
     batch.for_each_record([&rec_count,
-                           &first_timestamp_delta,
-                           &last_timestamp_delta,
+                           &max_timestamp_delta,
                            &ret,
                            &keep_idx,
                            &offset_deltas](model::record record) {
@@ -255,10 +258,11 @@ copy_data_segment_reducer::filter(model::record_batch batch) {
              * record batch with the uncompressed records so they were being
              * re-encoded.
              */
-            if (!first_timestamp_delta) {
-                first_timestamp_delta = record.timestamp_delta();
+            if (
+              !max_timestamp_delta
+              || record.timestamp_delta() > *max_timestamp_delta) {
+                max_timestamp_delta = record.timestamp_delta();
             }
-            last_timestamp_delta = record.timestamp_delta();
             model::append_record_to_buffer(ret, record);
             ++rec_count;
         }
@@ -303,11 +307,19 @@ copy_data_segment_reducer::filter(model::record_batch batch) {
     // previous value prior to becoming empty.
     //
     auto& hdr = batch.header();
-    const auto first_time = model::timestamp(
-      hdr.first_timestamp() + first_timestamp_delta.value());
+    // BUGFIX (future-timestamp mint via compaction): PRESERVE the original first_timestamp.
+    // The surviving records above kept their original timestamp deltas (relative to this
+    // first_timestamp), so their absolute timestamps remain correct. Previously first_timestamp
+    // was advanced to the first surviving record's absolute timestamp WITHOUT rebasing the
+    // per-record deltas -- which shifted every surviving record forward by first_timestamp_delta.
+    // When the batch's original base record was dropped (e.g. a tombstone) and the batch was out
+    // of order, that shift was large and positive, turning strictly-past records into records
+    // with FUTURE timestamps. Preserving first_timestamp mirrors how base_offset is preserved.
+    const auto first_time = hdr.first_timestamp;
     auto last_time = hdr.max_timestamp;
     if (hdr.attrs.timestamp_type() == model::timestamp_type::create_time) {
-        last_time = model::timestamp(first_time() + last_timestamp_delta);
+        last_time = model::timestamp(
+          hdr.first_timestamp() + max_timestamp_delta.value());
     }
     auto new_hdr = hdr;