sr: retry create ACLs for the ephemeral principal

It is possible that the call to create ACLs fails. For example, there
could not be a controller leader yet. This commit will auto-retry the
request after a small sleep.

Fixes redpanda-data#11141

src/v/pandaproxy/schema_registry/service.cc

@@ -37,6 +37,7 @@
 
 #include <seastar/core/coroutine.hh>
 #include <seastar/core/future-util.hh>
+#include <seastar/core/loop.hh>
 #include <seastar/core/memory.hh>
 #include <seastar/coroutine/parallel_for_each.hh>
 #include <seastar/http/api_docs.hh>
@@ -48,6 +49,7 @@ namespace pandaproxy::schema_registry {
 using server = ctx_server<service>;
 const security::acl_principal principal{
   security::principal_type::ephemeral_user, "__schema_registry"};
+static constexpr auto create_acls_backoff = 10ms;
 
 class wrap {
 public:
@@ -202,18 +204,44 @@ ss::future<> service::configure() {
       _ctx.smp_sg, [has_ephemeral_credentials](service& s) {
           s._has_ephemeral_credentials = has_ephemeral_credentials;
       });
-    co_await _controller->get_security_frontend().local().create_acls(
-      {security::acl_binding{
-        security::resource_pattern{
-          security::resource_type::topic,
-          model::schema_registry_internal_tp.topic,
-          security::pattern_type::literal},
-        security::acl_entry{
-          principal,
-          security::acl_host::wildcard_host(),
-          security::acl_operation::all,
-          security::acl_permission::allow}}},
-      5s);
+
+    std::vector<cluster::errc> err_vec;
+    // This could be any non-success errc
+    err_vec.push_back(cluster::errc::timeout);
+
+    co_await ss::do_until(
+      [&err_vec] { return err_vec[0] == cluster::errc::success; },
+      [this, &err_vec] {
+          return _controller->get_security_frontend()
+            .local()
+            .create_acls(
+              {security::acl_binding{
+                security::resource_pattern{
+                  security::resource_type::topic,
+                  model::schema_registry_internal_tp.topic,
+                  security::pattern_type::literal},
+                security::acl_entry{
+                  principal,
+                  security::acl_host::wildcard_host(),
+                  security::acl_operation::all,
+                  security::acl_permission::allow}}},
+              5s)
+            .then([&err_vec](std::vector<cluster::errc> ev) {
+                err_vec = std::move(ev);
+                if (err_vec[0] != cluster::errc::success) {
+                    vlog(
+                      plog.warn,
+                      "Failed creating ACLs, User {}, err {} - {}",
+                      principal,
+                      err_vec[0],
+                      cluster::make_error_code(err_vec[0]).message());
+                    return ss::sleep(create_acls_backoff);
+                }
+                return ss::make_ready_future<>();
+            });
+      });
+
+    vlog(plog.info, "Successfully created ACLs, User {}", principal);
 }
 
 ss::future<> service::mitigate_error(std::exception_ptr eptr) {