r/utils: fixed setting log start delta in snapshot

When topic partition is recovered from cloud data Raft protocol is
seeded with special snapshot created during recovery. The snapshot may
then be used to seed out of date followers. The snapshot content must
be valid in terms of offsets and offset translator deltas.

Previously the delta set in the snapshot created during recovery was set
to 0. When this snapshot is delivered to the follower it will reset the
follower offset translation state accordingly. This may lead to
inconsistent deltas across replicas.

Signed-off-by: Michał Maślanka <[email protected]>

src/v/raft/consensus_utils.cc

@@ -377,7 +377,8 @@ ss::future<> create_raft_state_for_pre_existing_partition(
   model::offset min_rp_offset,
   model::offset max_rp_offset,
   model::term_id last_included_term,
-  std::vector<raft::vnode> initial_nodes) {
+  std::vector<raft::vnode> initial_nodes,
+  model::offset_delta log_start_delta) {
     // Prepare Raft state in kvstore
     vlog(
       raftlog.debug,
@@ -403,7 +404,7 @@ ss::future<> create_raft_state_for_pre_existing_partition(
       .version = raft::snapshot_metadata::current_version,
       .latest_configuration = std::move(group_config),
       .cluster_time = ss::lowres_clock::now(),
-      .log_start_delta = raft::offset_translator_delta{0},
+      .log_start_delta = offset_translator_delta{log_start_delta},
     };
 
     vlog(
@@ -458,7 +459,8 @@ ss::future<> bootstrap_pre_existing_partition(
       min_rp_offset,
       max_rp_offset,
       last_included_term,
-      initial_nodes);
+      initial_nodes,
+      model::offset_delta{ot_state->delta(min_rp_offset)});
 }
 
 } // namespace raft::details