[upstream] PS-8385, Bug #108471 (change redo log consumer registrator…

… to allow user to specify a consumer name)

https://jira.percona.com/browse/PS-8385

Problem:
Currently implementation of redo log consumer UDF does not allow users
to specify a custom name for the consumer. This can be misleading when
ER_IB_MSG_LOG_WRITER_WAIT_ON_CONSUMER error is issued as it will always
report MEB as consumer name.

Fix:
Adjust the UDF to allow users to optionally specify a consumer name. In
case of no consumer name specified, assume MEB.
Also adjusted Log_consumer class to have a consumer_type ENUM. This is
used log_writer_wait_on_consumers to validate we are not waiting on
CONSUMER_TYPE_SERVER (log archive or checkpointer).

storage/innobase/arch/arch0log.cc

@@ -944,6 +944,10 @@ const std::string &Arch_log_consumer::get_name() const {
   return name;
 }
 
+Log_consumer::consumer_type Arch_log_consumer::get_consumer_type() const {
+  return Log_consumer::consumer_type::SERVER;
+}
+
 lsn_t Arch_log_consumer::get_consumed_lsn() const {
   ut_a(arch_log_sys != nullptr);
   ut_a(arch_log_sys->is_active());

storage/innobase/include/arch0arch.h

@@ -1308,6 +1308,8 @@ using Arch_Grp_List_Iter = Arch_Grp_List::iterator;
 
 class Arch_log_consumer : public Log_consumer {
  public:
+  Log_consumer::consumer_type get_consumer_type() const override;
+
   const std::string &get_name() const override;
 
   lsn_t get_consumed_lsn() const override;

storage/innobase/include/log0consumer.h

@@ -51,6 +51,11 @@ class Log_consumer {
   is the most lagging one and it is critical to consume
   the oldest redo log file. */
   virtual void consumption_requested() = 0;
+
+  enum class consumer_type { SERVER, USER };
+
+  /** @return Type of this consumer. */
+  virtual consumer_type get_consumer_type() const = 0;
 };
 
 class Log_user_consumer : public Log_consumer {
@@ -69,6 +74,8 @@ class Log_user_consumer : public Log_consumer {
 
   void consumption_requested() override;
 
+  Log_consumer::consumer_type get_consumer_type() const override;
+
  private:
   /** Name of this consumer (saved value from ctor). */
   const std::string m_name;
@@ -82,6 +89,8 @@ class Log_checkpoint_consumer : public Log_consumer {
  public:
   explicit Log_checkpoint_consumer(log_t &log);
 
+  Log_consumer::consumer_type get_consumer_type() const override;
+
   const std::string &get_name() const override;
 
   lsn_t get_consumed_lsn() const override;

storage/innobase/log/log0consumer.cc

@@ -54,6 +54,10 @@ lsn_t Log_user_consumer::get_consumed_lsn() const { return m_consumed_lsn; }
 
 void Log_user_consumer::consumption_requested() {}
 
+Log_consumer::consumer_type Log_user_consumer::get_consumer_type() const {
+  return Log_consumer::consumer_type::USER;
+}
+
 Log_checkpoint_consumer::Log_checkpoint_consumer(log_t &log) : m_log{log} {}
 
 const std::string &Log_checkpoint_consumer::get_name() const {
@@ -69,6 +73,10 @@ void Log_checkpoint_consumer::consumption_requested() {
   log_request_checkpoint_in_next_file(m_log);
 }
 
+Log_consumer::consumer_type Log_checkpoint_consumer::get_consumer_type() const {
+  return Log_consumer::consumer_type::SERVER;
+}
+
 void log_consumer_register(log_t &log, Log_consumer *log_consumer) {
   ut_ad(log_files_mutex_own(log) || srv_is_being_started);
 

storage/innobase/log/log0meb.cc

@@ -1669,7 +1669,8 @@ static bool redo_log_archive_flush(THD *thd) {
 static std::unique_ptr<Log_user_consumer> log_meb_consumer;
 static innodb_session_t *log_meb_consumer_session;
 
-static bool redo_log_consumer_register(innodb_session_t *session) {
+static bool redo_log_consumer_register(innodb_session_t *session,
+                                       std::string const &name) {
   log_t &log = *log_sys;
 
   IB_mutex_guard checkpointer_latch{&(log.checkpointer_mutex),
@@ -1683,7 +1684,7 @@ static bool redo_log_consumer_register(innodb_session_t *session) {
 
   ut_a(log_meb_consumer.get() == nullptr);
 
-  log_meb_consumer = std::make_unique<Log_user_consumer>("MEB");
+  log_meb_consumer = std::make_unique<Log_user_consumer>(name);
 
   log_meb_consumer->set_consumed_lsn(log_get_checkpoint_lsn(log));
 
@@ -2292,7 +2293,7 @@ long long innodb_redo_log_sharp_checkpoint(
 */
 bool innodb_redo_log_consumer_register_init([[maybe_unused]] UDF_INIT *initid,
                                             UDF_ARGS *args, char *message) {
-  if (args->arg_count != 0) {
+  if (args->arg_count > 1) {
     snprintf(message, MYSQL_ERRMSG_SIZE, "Invalid number of arguments.");
     return true;
   }
@@ -2321,12 +2322,18 @@ long long innodb_redo_log_consumer_register(
     [[maybe_unused]] UDF_INIT *initid, [[maybe_unused]] UDF_ARGS *args,
     [[maybe_unused]] unsigned char *null_value,
     [[maybe_unused]] unsigned char *error) {
+  std::string name = "MEB";
   if (current_thd == nullptr ||
       verify_privilege(current_thd, backup_admin_privilege)) {
     return 1;
   }
-  return static_cast<long long>(
-      meb::redo_log_consumer_register(thd_to_innodb_session(current_thd)));
+
+  if (args->arg_count >= 1) {
+    name.assign(args->args[0]);
+  }
+
+  return static_cast<long long>(meb::redo_log_consumer_register(
+      thd_to_innodb_session(current_thd), name));
 }
 
 /**

storage/innobase/log/log0write.cc

@@ -2085,11 +2085,12 @@ static void log_writer_wait_on_consumers(log_t &log, lsn_t next_write_lsn) {
       break;
     }
     const std::string name = consumer->get_name();
+
     log_files_mutex_exit(*log_sys);
-    /* This should not be a checkpointer nor archiver, as we've used dedicated
-    log_writer_wait_on_checkpoint() and log_writer_wait_on_archiver() to wait
-    for them already */
-    ut_ad(name == "MEB");
+    /* This should not be a checkpointer nor archiver (CONSUMER_TYPE_SERVER), as
+    we've used dedicated log_writer_wait_on_checkpoint() and
+    log_writer_wait_on_archiver() to wait for them already */
+    ut_ad(consumer->get_consumer_type() == Log_consumer::consumer_type::USER);
     log_writer_mutex_exit(log);
     if (attempt++ % ATTEMPTS_BETWEEN_WARNINGS == 0) {
       ib::log_warn(ER_IB_MSG_LOG_WRITER_WAIT_ON_CONSUMER, name.c_str(),