Merge pull request #27724 from wdberkeley/rc-add-objects-retries

cloud_topics/reconciler: add retry mechanism for transport errors

Author: dotnwat

src/v/cloud_topics/reconciler/BUILD

@@ -52,6 +52,7 @@ redpanda_cc_library(
         "//src/v/model",
         "//src/v/random:generators",
         "//src/v/ssx:future_util",
+        "//src/v/utils:retry_chain_node",
         "@abseil-cpp//absl/container:btree",
         "@seastar",
     ],

src/v/cloud_topics/reconciler/reconciler.cc

@@ -24,10 +24,15 @@
 #include "model/namespace.h"
 #include "random/generators.h"
 #include "ssx/future-util.h"
+#include "utils/retry_chain_node.h"
 
 #include <seastar/core/shared_ptr.hh>
 #include <seastar/util/log.hh>
 
+#include <chrono>
+
+using namespace std::chrono_literals;
+
 namespace {
 ss::logger lg("reconciler");
 
@@ -535,6 +540,37 @@ reconciler::add_object_metadata(
     co_return std::expected<void, reconcile_error>{};
 }
 
+ss::future<std::expected<l1::metastore::add_response, l1::metastore::errc>>
+reconciler::add_objects_with_retry(
+  std::unique_ptr<l1::metastore::object_metadata_builder> meta_builder,
+  l1::metastore::term_offset_map_t terms) {
+    static constexpr auto timeout = 5s;
+    static constexpr auto backoff = 100ms;
+
+    retry_chain_node rtc(_as, ss::lowres_clock::now() + timeout, backoff);
+    retry_chain_logger ctxlog(lg, rtc, "add_objects");
+    for (auto permit = rtc.retry(); permit.is_allowed; permit = rtc.retry()) {
+        auto add_result = co_await _metastore->add_objects(
+          *meta_builder, terms);
+
+        if (add_result.has_value()) {
+            co_return std::move(add_result).value();
+        }
+
+        if (add_result.error() != l1::metastore::errc::transport_error) {
+            vlog(
+              lg.error,
+              "Non-retryable error adding objects to the L1 metastore: {}",
+              add_result.error());
+            co_return std::unexpected(add_result.error());
+        }
+
+        co_await ss::sleep_abortable(permit.delay, rtc.root_abort_source());
+    }
+
+    co_return std::unexpected(l1::metastore::errc::transport_error);
+}
+
 ss::future<std::expected<void, reconcile_error>> reconciler::commit_objects(
   const chunked_vector<built_object_metadata>& objects,
   std::unique_ptr<l1::metastore::object_metadata_builder> meta_builder) {
@@ -552,11 +588,11 @@ ss::future<std::expected<void, reconcile_error>> reconciler::commit_objects(
         }
     }
 
-    auto add_objects_result = co_await _metastore->add_objects(
-      *meta_builder, terms);
+    auto add_objects_result = co_await add_objects_with_retry(
+      std::move(meta_builder), std::move(terms));
     if (!add_objects_result.has_value()) {
         vlog(
-          lg.error,
+          lg.warn,
           "Failed to add objects to the L1 metastore: {}",
           add_objects_result.error());
         // TODO: The objects have been uploaded. The reconciler could

src/v/cloud_topics/reconciler/reconciler.h

@@ -254,6 +254,15 @@ class reconciler {
     make_reader(frontend*, kafka::offset start_offset, size_t);
 
 private:
+    /*
+     * Retry metastore add_objects calls on transport errors.
+     * Other metastore errors are not retried.
+     */
+    ss::future<std::expected<l1::metastore::add_response, l1::metastore::errc>>
+    add_objects_with_retry(
+      std::unique_ptr<l1::metastore::object_metadata_builder> meta_builder,
+      l1::metastore::term_offset_map_t terms);
+
     data_plane_api* _data_plane;
     l1::io* _l1_io;
     l1::metastore* _metastore;