Skip to content

Commit

Permalink
cloud_storage: Store model::ntp inside remote_partition
Browse files Browse the repository at this point in the history
Previously, the remote_partition used async_manifest_view instance to
acquire the ntp. This is not safe in some cases because the
async_manifest_view could be disposed. Normally this isn't happening but
this could happen in tests and it might happen in Redpanda when
something is stuck during shutdown process.

(cherry picked from commit 5b0017f)
  • Loading branch information
Lazin authored and vbotbuildovich committed Feb 5, 2024
1 parent 9aca62f commit 257ab80
Show file tree
Hide file tree
Showing 2 changed files with 8 additions and 8 deletions.
15 changes: 7 additions & 8 deletions src/v/cloud_storage/remote_partition.cc
Original file line number Diff line number Diff line change
Expand Up @@ -859,7 +859,8 @@ remote_partition::remote_partition(
cache& c,
cloud_storage_clients::bucket_name bucket,
partition_probe& probe)
: _rtc(_as)
: _ntp(m->get_ntp())
, _rtc(_as)
, _ctxlog(cst_log, _rtc, m->get_ntp().path())
, _api(api)
, _cache(c)
Expand Down Expand Up @@ -897,7 +898,7 @@ ss::future<> remote_partition::run_eviction_loop() {
// got stuck. The deadline is set to 5 minutes to avoid false positives.
// The callback is self sufficient and can outlive the remote_partition
// instance.
watchdog wd(300s, [ntp = get_ntp()] {
watchdog wd(300s, [ntp = _ntp] {
vlog(cst_log.error, "Eviction loop for partition {} stuck", ntp);
});
auto eviction_in_flight = std::exchange(_eviction_pending, {});
Expand Down Expand Up @@ -925,7 +926,7 @@ kafka::offset remote_partition::first_uploaded_offset() {
vassert(
_manifest_view->stm_manifest().size() > 0,
"The manifest for {} is not expected to be empty",
_manifest_view->stm_manifest().get_ntp());
_ntp);
auto so
= _manifest_view->stm_manifest().full_log_start_kafka_offset().value();
vlog(_ctxlog.trace, "remote partition first_uploaded_offset: {}", so);
Expand All @@ -936,15 +937,13 @@ kafka::offset remote_partition::next_kafka_offset() {
vassert(
_manifest_view->stm_manifest().size() > 0,
"The manifest for {} is not expected to be empty",
_manifest_view->get_ntp());
_ntp);
auto next = _manifest_view->stm_manifest().get_next_kafka_offset().value();
vlog(_ctxlog.debug, "remote partition next_kafka_offset: {}", next);
return next;
}

const model::ntp& remote_partition::get_ntp() const {
return _manifest_view->get_ntp();
}
const model::ntp& remote_partition::get_ntp() const { return _ntp; }

bool remote_partition::is_data_available() const {
const auto& stmm = _manifest_view->stm_manifest();
Expand Down Expand Up @@ -1330,7 +1329,7 @@ void remote_partition::finalize() {
auto serialized_manifest = stm_manifest.to_iobuf();

finalize_data data{
.ntp = stm_manifest.get_ntp(),
.ntp = get_ntp(),
.revision = stm_manifest.get_revision_id(),
.bucket = _bucket,
.key
Expand Down
1 change: 1 addition & 0 deletions src/v/cloud_storage/remote_partition.h
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,7 @@ class remote_partition
iterator get_or_materialize_segment(
const remote_segment_path& path, const segment_meta&, segment_units);

model::ntp _ntp;
retry_chain_node _rtc;
retry_chain_logger _ctxlog;
ss::gate _gate;
Expand Down

0 comments on commit 257ab80

Please sign in to comment.