storage: fix test partition_size_while_cleanup

This test was racy. Two functional changes are made:
- flush the log after writes: this makes the readable
  range deterministic.
- modify the success condition to say that log byte count
  must equal sum of readable batch byte count (which is now
  all batches because of the flush).

Also add several extra assertions so that the test's expectations
are clearer & if something goes wrong it'll fail earlier.

Fixes #5179

src/v/storage/tests/storage_e2e_test.cc

@@ -1161,20 +1161,52 @@ FIXTURE_TEST(partition_size_while_cleanup, storage_test_fixture) {
       ntp, mgr.config().base_dir, std::make_unique<overrides_t>(ov));
     auto log = mgr.manage(std::move(ntp_cfg)).get0();
 
-    auto result = append_exactly(log, 100, 1_KiB, "key").get0(); // 100*1_KiB
+    auto sz_initial = get_disk_log(log)->get_probe().partition_size();
+    info("sz_initial={}", sz_initial);
+    BOOST_REQUIRE_EQUAL(sz_initial, 0);
+
+    // Add 100 batches with one event each, all events having the same key
+    static constexpr size_t batch_size = 1_KiB; // Size visible to Kafka API
+    static constexpr size_t batch_size_ondisk = 1033; // Size internally
+    static constexpr size_t input_batch_count = 100;
+
+    auto result = append_exactly(log, input_batch_count, batch_size, "key")
+                    .get0(); // 100*1_KiB
+
+    // Test becomes non-deterministic if we allow flush in background: flush
+    // explicitly instead.
+    log.flush().get();
+
+    // Read back and validate content of log pre-compaction.
+    BOOST_REQUIRE_EQUAL(
+      get_disk_log(log)->get_probe().partition_size(),
+      input_batch_count * batch_size_ondisk);
+    BOOST_REQUIRE_EQUAL(
+      read_and_validate_all_batches(log).size(), input_batch_count);
+    auto lstats_before = log.offsets();
+    BOOST_REQUIRE_EQUAL(
+      lstats_before.committed_offset, model::offset{input_batch_count - 1});
+    BOOST_REQUIRE_EQUAL(lstats_before.start_offset, model::offset{0});
 
-    auto sz_before = get_disk_log(log)->get_probe().partition_size();
     log.set_collectible_offset(log.offsets().dirty_offset);
     storage::compaction_config ccfg(
       model::timestamp::min(), 60_KiB, ss::default_priority_class(), as);
 
+    // Compact 10 times, with a configuration calling for 60kiB max log size.
+    // This results in approximately prefix truncating at offset 50, although
+    // this is nondeterministic due to max_segment_size jitter.
     for (int i = 0; i < 10; ++i) {
         log.compact(ccfg).get0();
     }
     auto sz_after = get_disk_log(log)->get_probe().partition_size();
 
     as.request_abort();
     auto lstats_after = log.offsets();
+    BOOST_REQUIRE_EQUAL(
+      lstats_after.committed_offset, lstats_before.committed_offset);
+
+    // Cannot assert on lstats_after.start_offset: its value depends on
+    // the jitter applied in disk_log_impl::_max_segment_size
 
     auto batches = read_and_validate_all_batches(log);
     auto total_batch_size = std::accumulate(
@@ -1185,13 +1217,16 @@ FIXTURE_TEST(partition_size_while_cleanup, storage_test_fixture) {
           return sum + b.size_bytes();
       });
 
-    // Expected size is equal to size of 4 batches (compacted one) plus batches
-    // being present in active segment that is not compacted
+    auto& segments = get_disk_log(log)->segments();
+    // One historic segment (all historic batches compacted down to 1), plus
+    // one active segment.
+    BOOST_REQUIRE_EQUAL(segments.size(), 2);
 
-    auto expected_size = total_batch_size
-                         + get_disk_log(log)->segments().back()->size_bytes();
-    BOOST_REQUIRE_EQUAL(
-      get_disk_log(log)->get_probe().partition_size(), expected_size);
+    // The log-scope reported size must be equal to the sum of the batch sizes
+    auto expected_size = total_batch_size;
+    auto actual_size = get_disk_log(log)->get_probe().partition_size();
+    info("expected_size={}, actual_size={}", expected_size, actual_size);
+    BOOST_REQUIRE_EQUAL(actual_size, expected_size);
 };
 
 FIXTURE_TEST(check_segment_size_jitter, storage_test_fixture) {