storage: add test_concurrent_segment_roll_and_close

To test race conditions between `segment::close()` and a segment roll,
particularly one which goes through `release_appender_in_background()`.

Author: WillemKauf

src/v/storage/tests/storage_e2e_fixture_test.cc

@@ -156,3 +156,31 @@ FIXTURE_TEST(test_concurrent_log_eviction_and_append, storage_e2e_fixture) {
     // final round of eviction.
     BOOST_REQUIRE_LE(log->segment_count(), 1);
 }
+
+FIXTURE_TEST(test_concurrent_segment_roll_and_close, storage_e2e_fixture) {
+    const auto topic_name = model::topic("tapioca");
+    const auto ntp = model::ntp(model::kafka_namespace, topic_name, 0);
+
+    cluster::topic_properties props;
+    add_topic({model::kafka_namespace, topic_name}, 1, props).get();
+    wait_for_leader(ntp).get();
+
+    auto partition = app.partition_manager.local().get(ntp);
+    auto* log = dynamic_cast<storage::disk_log_impl*>(partition->log().get());
+    auto seg = log->segments().back();
+
+    // Hold a read lock, which will force release_appender() to go through
+    // release_appender_in_background()
+    auto read_lock_holder = seg->read_lock().get();
+
+    auto roll_fut = log->force_roll(ss::default_priority_class());
+    auto release_holder_fut = ss::sleep(100ms).then(
+      [read_locker_holder = std::move(read_lock_holder)] {});
+    auto remove_segment_fut = remove_segment_permanently(log, seg);
+
+    ss::when_all(
+      std::move(roll_fut),
+      std::move(remove_segment_fut),
+      std::move(release_holder_fut))
+      .get();
+}