ct: Add batcher test

The test checks L0 object size distribution.

Signed-off-by: Evgeny Lazin <4lazin@gmail.com>

Author: Lazin

src/v/cloud_topics/level_zero/batcher/tests/BUILD

@@ -51,6 +51,7 @@ redpanda_cc_gtest(
         "//src/v/storage:record_batch_utils",
         "//src/v/test_utils:gtest",
         "//src/v/test_utils:random_bytes",
+        "//src/v/test_utils:scoped_config",
         "@googletest//:gtest",
         "@seastar",
     ],

src/v/cloud_topics/level_zero/batcher/tests/batcher_test.cc

@@ -27,6 +27,7 @@
 #include "remote_mock.h"
 #include "storage/record_batch_builder.h"
 #include "test_utils/random_bytes.h"
+#include "test_utils/scoped_config.h"
 #include "test_utils/test.h"
 
 #include <seastar/core/abort_source.hh>
@@ -352,6 +353,80 @@ TEST_CORO(batcher_test, expired_write_request) {
     }
 }
 
-// TODO: add more tests
-// - behaviour in case if pending write request sizes exceed L0 object size
-// limit
+TEST_CORO(batcher_test, chunk_splitting_balances_upload_sizes) {
+    scoped_config cfg;
+    // Use a small threshold so test data splits into multiple chunks.
+    cfg.get("cloud_topics_produce_batching_size_threshold")
+      .set_value(size_t{4096});
+
+    remote_mock mock;
+    mock.expect_upload_object_repeatedly();
+
+    cloud_storage_clients::bucket_name bucket("foo");
+    cloud_topics::l0::write_pipeline<ss::manual_clock> pipeline;
+    static_cluster_services cluster_services;
+    cloud_topics::l0::batcher<ss::manual_clock> batcher(
+      pipeline.register_write_pipeline_stage(),
+      bucket,
+      mock,
+      &cluster_services);
+    cloud_topics::l0::write_pipeline_accessor pipeline_accessor{
+      .pipeline = &pipeline,
+    };
+
+    // Push several write requests. Each has 1 batch with 10 records
+    // (~3KB serialized), so 6 requests total ~18KB. With threshold=4096
+    // this should produce multiple balanced chunks.
+    const int num_requests = 6;
+    std::vector<ss::future<std::expected<
+      chunked_vector<cloud_topics::extent_meta>,
+      std::error_code>>>
+      futures;
+
+    const auto timeout = 10s;
+    auto deadline = ss::manual_clock::now() + timeout;
+
+    for (int i = 0; i < num_requests; i++) {
+        auto [_, records, batches] = get_random_batches(1, 10);
+        futures.push_back(pipeline.write_and_debounce(
+          model::controller_ntp, min_epoch, std::move(batches), deadline));
+    }
+
+    // Wait for all write requests to be staged in the pipeline
+    // before starting the batcher. subscribe() checks pre-existing
+    // pending data, so bg_controller_loop's wait_next will return
+    // immediately seeing all requests at once.
+    co_await sleep_until(10ms, [&] {
+        return pipeline_accessor.write_requests_pending(num_requests);
+    });
+
+    // Start the batcher — bg_controller_loop will pull all 6 requests
+    // in one batch and split them into balanced chunks.
+    co_await batcher.start();
+
+    // Wait for all write request futures to resolve (the batcher
+    // loop processes chunks via spawn_with_gate, which sets the
+    // promises on each write request).
+    auto results = co_await ss::when_all_succeed(std::move(futures));
+    for (auto& res : results) {
+        ASSERT_TRUE_CORO(res.has_value());
+    }
+
+    co_await batcher.stop();
+
+    // Multiple uploads should happen since total data exceeds threshold.
+    ASSERT_GT_CORO(mock.payloads.size(), size_t{1});
+
+    // Verify uploads are balanced: no upload should be excessively small
+    // compared to the average.
+    size_t total_payload = 0;
+    for (const auto& p : mock.payloads) {
+        total_payload += p.size();
+    }
+    size_t avg_size = total_payload / mock.payloads.size();
+    for (size_t i = 0; i < mock.payloads.size(); i++) {
+        EXPECT_GE(mock.payloads[i].size(), avg_size / 3)
+          << "Upload " << i << " size " << mock.payloads[i].size()
+          << " is too small relative to average " << avg_size;
+    }
+}

src/v/cloud_topics/level_zero/batcher/tests/remote_mock.h

@@ -88,6 +88,19 @@ class remote_mock final : public cloud_io::remote_api<ss::manual_clock> {
               ss::make_ready_future<cloud_io::upload_result>(res)));
     }
 
+    /// Accept any number of upload_object calls, recording keys and payloads.
+    void expect_upload_object_repeatedly(
+      cloud_io::upload_result res = cloud_io::upload_result::success) {
+        EXPECT_CALL(*this, upload_object(::testing::_))
+          .WillRepeatedly(
+            [this,
+             res](const cloud_io::basic_upload_request<ss::manual_clock>& req) {
+                keys.push_back(req.transfer_details.key);
+                payloads.push_back(iobuf_to_bytes(req.payload));
+                return ss::make_ready_future<cloud_io::upload_result>(res);
+            });
+    }
+
     static std::deque<ss::sstring>
     convert_bytes_to_string(const chunked_vector<bytes>& expected) {
         std::deque<ss::sstring> result;