Merge pull request #10389 from vbotbuildovich/backport-5763-v23.1.x-528

[v23.1.x] k/client: auto-retry on consumer offsets commit and fetch

src/v/kafka/client/client.cc

@@ -481,8 +481,10 @@ ss::future<offset_fetch_response> client::consumer_offset_fetch(
   const member_id& name,
   std::vector<offset_fetch_request_topic> topics) {
     return get_consumer(g_id, name)
-      .then([topics{std::move(topics)}](shared_consumer_t c) mutable {
-          return c->offset_fetch(std::move(topics));
+      .then([this, topics{std::move(topics)}](shared_consumer_t c) mutable {
+          return gated_retry_with_mitigation([c, topics{std::move(topics)}]() {
+              return c->offset_fetch(topics);
+          });
       });
 }
 
@@ -491,8 +493,10 @@ ss::future<offset_commit_response> client::consumer_offset_commit(
   const member_id& name,
   std::vector<offset_commit_request_topic> topics) {
     return get_consumer(g_id, name)
-      .then([topics{std::move(topics)}](shared_consumer_t c) mutable {
-          return c->offset_commit(std::move(topics));
+      .then([this, topics{std::move(topics)}](shared_consumer_t c) mutable {
+          return gated_retry_with_mitigation([c, topics{std::move(topics)}]() {
+              return c->offset_commit(topics);
+          });
       });
 }