sr/service: Add topic metadata polling loop after creation

Signed-off-by: Oren Leiman <oren.leiman@redpanda.com>

Author: oleiman

src/v/pandaproxy/schema_registry/service.cc

@@ -21,6 +21,7 @@
 #include "kafka/client/exceptions.h"
 #include "kafka/data/rpc/deps.h"
 #include "kafka/protocol/errors.h"
+#include "kafka/protocol/exceptions.h"
 #include "kafka/protocol/list_offset.h"
 #include "kafka/server/handlers/topics/types.h"
 #include "model/fundamental.h"
@@ -29,6 +30,7 @@
 #include "pandaproxy/logger.h"
 #include "pandaproxy/schema_registry/auth.h"
 #include "pandaproxy/schema_registry/configuration.h"
+#include "pandaproxy/schema_registry/exceptions.h"
 #include "pandaproxy/schema_registry/handlers.h"
 #include "pandaproxy/schema_registry/storage.h"
 #include "pandaproxy/schema_registry/types.h"
@@ -43,6 +45,8 @@
 
 #include <seastar/core/coroutine.hh>
 #include <seastar/core/future-util.hh>
+#include <seastar/core/sleep.hh>
+#include <seastar/coroutine/parallel_for_each.hh>
 #include <seastar/http/api_docs.hh>
 #include <seastar/util/log.hh>
 #include <seastar/util/noncopyable_function.hh>
@@ -381,12 +385,44 @@ ss::future<> service::do_start() {
           std::current_exception());
         throw;
     }
-    co_await container().invoke_on_all(_ctx.smp_sg, [](service& s) {
-        s._is_started = true;
-        return ss::this_shard_id() == seq_writer::reader_shard
-                 ? s.fetch_internal_topic()
-                 : ss::now();
-    });
+    co_await container().invoke_on_all(
+      _ctx.smp_sg, [](this auto, service& s) -> ss::future<> {
+          s._is_started = true;
+          if (ss::this_shard_id() != seq_writer::reader_shard) {
+              co_return;
+          }
+
+          // create_internal_topic returns once the controller commits
+          // the topic, but the metadata cache is updated asynchronously.
+          // Transient errors during this window — topic not yet visible,
+          // leader not yet elected — are retried here. The kafka client
+          // transport surfaces these as SR exceptions, while the RPC
+          // transport surfaces them as kafka::exceptions. Both use
+          // unknown_server_error.
+          for (int attempts = 0;; ++attempts) {
+              auto fut = co_await ss::coroutine::as_future(
+                s.fetch_internal_topic());
+              if (!fut.failed()) {
+                  co_return;
+              }
+              auto eptr = fut.get_exception();
+              bool retriable = false;
+              try {
+                  std::rethrow_exception(eptr);
+              } catch (const kafka::exception_base& e) {
+                  retriable = e.error == kafka::error_code::unknown_server_error
+                              && attempts < 10;
+              } catch (const exception& e) {
+                  retriable = e.code()
+                                == kafka::error_code::unknown_server_error
+                              && attempts < 10;
+              }
+              if (!retriable) {
+                  std::rethrow_exception(eptr);
+              }
+              co_await ss::sleep(std::chrono::milliseconds(100));
+          }
+      });
 }
 
 ss::future<> create_acls(cluster::security_frontend& security_fe) {