misc: use-after-move vlog fixes & remove NOLINTNEXTLINE

src/v/base/vlog.h

@@ -11,25 +11,19 @@
 #pragma once
 #include "base/source_location.h"
 
-// NOLINTNEXTLINE
 #define fmt_with_ctx(method, fmt, args...)                                     \
     method("{} - " fmt, vlog::file_line::current(), ##args)
 
-// NOLINTNEXTLINE
 #define vlog(method, fmt, args...) fmt_with_ctx(method, fmt, ##args)
 
-// NOLINTNEXTLINE
 #define fmt_with_ctx_level(logger, level, fmt, args...)                        \
     logger.log(level, "{} - " fmt, vlog::file_line::current(), ##args)
 
-// NOLINTNEXTLINE
 #define vlogl(logger, level, fmt, args...)                                     \
     fmt_with_ctx_level(logger, level, fmt, ##args)
 
-// NOLINTNEXTLINE
 #define fmt_with_ctx_level_and_rate(logger, level, rate, fmt, args...)         \
     logger.log(level, rate, "{} - " fmt, vlog::file_line::current(), ##args)
 
-// NOLINTNEXTLINE
 #define vloglr(logger, level, rate, fmt, args...)                              \
     fmt_with_ctx_level_and_rate(logger, level, rate, fmt, ##args)

src/v/cluster/log_eviction_stm.cc

@@ -224,8 +224,8 @@ ss::future<log_eviction_stm::offset_result> log_eviction_stm::truncate(
       "Replicating prefix_truncate command, redpanda start offset: {}, kafka "
       "start offset: {} "
       "current last snapshot offset: {}, current last visible offset: {}",
-      val.rp_start_offset,
-      val.kafka_start_offset,
+      rp_start_offset,
+      kafka_start_offset,
       _raft->last_snapshot_index(),
       _raft->last_visible_index());
 

src/v/cluster/rm_stm.cc

@@ -1755,11 +1755,12 @@ void rm_stm::apply_fence(model::producer_identity pid, model::record_batch b) {
     }
     auto producer = result.value().first;
     auto header = b.header();
+    auto base_offset = b.base_offset();
     auto batch_data = read_fence_batch(std::move(b));
     vlog(
       _ctx_log.trace,
       "applying fence batch, offset: {}, pid: {}",
-      b.base_offset(),
+      base_offset,
       batch_data.bid.pid);
     producer->apply_transaction_begin(header, batch_data);
     _highest_producer_id = std::max(

src/v/kafka/server/group_manager.cc

@@ -1055,10 +1055,11 @@ group_manager::do_bulk_write_offsets(group_offsets_snapshot snap, bool merge) {
             kafka_topics.emplace_back(std::move(kafka_t));
             co_await ss::maybe_yield();
         }
+        auto group_id = kafka_r.data.group_id;
         vlog(
           cg_klog.info,
           "Restoring group {} from snapshot on {}",
-          kafka_r.data.group_id,
+          group_id,
           offsets_ntp);
         auto stages = offset_commit(std::move(kafka_r));
         co_await std::move(stages.dispatched);
@@ -1072,7 +1073,7 @@ group_manager::do_bulk_write_offsets(group_offsets_snapshot snap, bool merge) {
                       "Error on {}/{} while restoring group {} on {}: {}",
                       kafka_t.name,
                       kafka_p.partition_index,
-                      kafka_r.data.group_id,
+                      group_id,
                       offsets_ntp,
                       kafka_p.error_code);
                     if (first_error != error_code::none) {

src/v/storage/mvlog/segment_appender.cc

@@ -19,6 +19,8 @@
 namespace storage::experimental::mvlog {
 
 ss::future<> segment_appender::append(model::record_batch batch) {
+    auto base_offset = batch.base_offset();
+    auto last_offset = batch.last_offset();
     // Build the body of the entry first so we can checksum its size.
     record_batch_entry_body entry_body;
     entry_body.term = batch.term();
@@ -42,8 +44,8 @@ ss::future<> segment_appender::append(model::record_batch batch) {
     vlog(
       log.trace,
       "Appended offsets [{}, {}], pos [{}, {})",
-      batch.base_offset(),
-      batch.last_offset(),
+      base_offset,
+      last_offset,
       orig_size,
       file_->size());
 }

src/v/transform/rpc/client.cc

@@ -1022,12 +1022,13 @@ ss::future<cluster::errc> client::do_remote_delete_committed_offsets(
   model::partition_id partition,
   absl::btree_set<model::transform_id> ids,
   model::timeout_clock::duration timeout) {
+    auto ids_size = ids.size();
     vlog(
       log.trace,
       "delete_committed_offsets(node={}): {} {}",
       node,
       partition,
-      ids.size());
+      ids_size);
     auto resp = co_await _connections->local()
                   .with_node_client<impl::transform_rpc_client_protocol>(
                     _self,
@@ -1047,7 +1048,7 @@ ss::future<cluster::errc> client::do_remote_delete_committed_offsets(
       "delete_committed_offsets(node={}): {} {}",
       node,
       resp,
-      ids.size());
+      ids_size);
     if (resp.has_error()) {
         co_return map_errc(resp.error());
     }