MDEV-34504 PURGE BINARY LOGS not working anymore

PURGE BINARY LOGS did not always purge binary logs. This commit fixes
some of the issues and adds notifications if a binary log cannot be
purged.

User visible changes:
- 'PURGE BINARY LOG TO log_name' and 'PURGE BINARY LOGS BEFORE date'
  worked differently. 'TO' ignored 'slave_connections_needed_for_purge'
  while 'BEFORE' did not. Now both versions ignores the
  'slave_connections_needed_for_purge variable'.
- 'PURGE BINARY LOG..' commands now returns 'note' if a binary log cannot
   be deleted like
   Note 1375 Binary log 'master-bin.000004' is not purged because it is
             the current active binlog
- Automatic binary log purges, based on date or size, will write a
  note to the error log if a binary log matching the size or date
  cannot yet be deleted.
- If 'slave_connections_needed_for_purge' is set from a config or
  command line, it is set to 0 if Galera is enabled and 1 otherwise
  (old default). This ensures that automatic binary log purge works
  with Galera as before the addition of
  'slave_connections_needed_for_purge'.
  If the variable is changed to 0, a warning will be printed to the error
  log.

Code changes:
- Added THD argument to several purge_logs related functions that needed
  THD.
- Added 'interactive' options to purge_logs functions. This allowed
  me to remove testing of sql_command == SQLCOM_PURGE.
- Changed purge_logs_before_date() to first check if log is applicable
  before calling can_purge_logs(). This ensures we do not get a
  notification for logs that does not match the remove criteria.
- MYSQL_BIN_LOG::can_purge_log() will write notifications to the user
  or error log if a log cannot yet be removed.
- log_in_use() will return reason why a binary log cannot be removed.

Changes to keep code consistent:
- Moved checking of binlog_format for Galera to be after Galera is
  initialized (The old check never worked). If Galera is enabled
  we now change the binlog_format to ROW, with a warning, instead of
  aborting the server. If this change happens a warning will be printed to
  the error log.
- Print a warning if Galera or FLASHBACK changes the binlog_format
  to ROW. Before it was done silently.

Reviewed by: Sergei Golubchik <serg@mariadb.com>,
             Kristian Nielsen <knielsen@knielsen-hq.org>

mysql-test/include/wait_for_purge.inc

@@ -20,7 +20,7 @@
 #
 # SIDE EFFECTS:
 #
-#    Disables --query_log while running, enables it afterwards.
+#    Disables --query_log and warnings while running, enables them afterwards.
 
 --echo include/wait_for_purge.inc "$purge_binlogs_to"
 
@@ -28,6 +28,8 @@ let $_wait_count= 300;
 let $_done= 0;
 
 --disable_query_log
+--disable_warnings
+
 while ($_wait_count)
 {
   dec $_wait_count;
@@ -50,4 +52,5 @@ if (!$_done)
   eval SHOW BINARY LOGS;
   --die ERROR: failed while waiting for binlog purge to $purge_binlogs_to
 }
+--enable_warnings
 --enable_query_log

mysql-test/main/mysqld--help.result

@@ -1308,6 +1308,7 @@ The following specify which files/extra groups are read (specified before remain
  Minimum number of connected slaves required for automatic
  binary log purge with max_binlog_total_size,
  binlog_expire_logs_seconds or binlog_expire_logs_days.
+ Default is 0 when Galera is enabled and 1 otherwise.
  --slave-ddl-exec-mode=name 
  How replication events should be executed. Legal values
  are STRICT and IDEMPOTENT (default). In IDEMPOTENT mode,

mysql-test/suite/binlog/r/binlog_flush_binlogs_delete_domain.result

@@ -49,6 +49,9 @@ ERROR HY000: Could not delete gtid domain. Reason: binlog files may contain gtid
 MDEV-31140: Missing error from DELETE_DOMAIN_ID when gtid_binlog_state partially matches GTID_LIST.
 FLUSH BINARY LOGS;
 PURGE BINARY LOGS TO 'master-bin.000005';
+show binary logs;
+Log_name	File_size
+master-bin.000005	#
 SET @@SESSION.gtid_domain_id=8;
 SET @@SESSION.server_id=10*8 + 1;
 INSERT INTO t SELECT 1+MAX(a) FROM t;

mysql-test/suite/binlog/r/binlog_index.result

@@ -30,6 +30,7 @@ flush logs;
 flush logs;
 *** must be a warning master-bin.000001 was not found ***
 Warnings:
+Note	1375	Binary log 'master-bin.000004' is not purged because it is the current active binlog
 Warning	1612	Being purged log master-bin.000001 was not found
 *** must show one record, of the active binlog, left in the index file after PURGE ***
 show binary logs;

mysql-test/suite/binlog/r/binlog_xa_recover.result

@@ -89,6 +89,8 @@ master-bin.000006	#	Format_desc	#	#	SERVER_VERSION, BINLOG_VERSION
 master-bin.000006	#	Gtid_list	#	#	[#-#-#]
 master-bin.000006	#	Binlog_checkpoint	#	#	master-bin.000004
 PURGE BINARY LOGS TO "master-bin.000006";
+Warnings:
+Note	1375	Binary log 'master-bin.000004' is not purged because it may be needed for crash recovery (XID)
 show binary logs;
 Log_name	File_size
 master-bin.000004	#

mysql-test/suite/binlog/t/binlog_flush_binlogs_delete_domain.test

@@ -91,6 +91,7 @@ while ($domain_cnt)
 FLUSH BINARY LOGS;
 --let $purge_to_binlog= query_get_value(SHOW MASTER STATUS, File, 1)
 --eval PURGE BINARY LOGS TO '$purge_to_binlog'
+--source include/show_binary_logs.inc
 --eval SET @@SESSION.gtid_domain_id=$err_domain_id
 --eval SET @@SESSION.server_id=10*$err_domain_id + $err_server_id
 eval INSERT INTO t SELECT 1+MAX(a) FROM t;

mysql-test/suite/binlog_encryption/binlog_index.result

@@ -30,6 +30,7 @@ flush logs;
 flush logs;
 *** must be a warning master-bin.000001 was not found ***
 Warnings:
+Note	1375	Binary log 'master-bin.000004' is not purged because it is the current active binlog
 Warning	1612	Being purged log master-bin.000001 was not found
 *** must show one record, of the active binlog, left in the index file after PURGE ***
 show binary logs;

mysql-test/suite/binlog_encryption/binlog_xa_recover.result

@@ -93,6 +93,8 @@ master-bin.000006	#	Start_encryption	#	#
 master-bin.000006	#	Gtid_list	#	#	[#-#-#]
 master-bin.000006	#	Binlog_checkpoint	#	#	master-bin.000004
 PURGE BINARY LOGS TO "master-bin.000006";
+Warnings:
+Note	1375	Binary log 'master-bin.000004' is not purged because it may be needed for crash recovery (XID)
 show binary logs;
 Log_name	File_size
 master-bin.000004	#

mysql-test/suite/compat/oracle/r/binlog_ptr_mysqlbinlog.result

+call mtr.add_suppression("Binlog_format changed to.*flashback");
 SET @@SQL_MODE = 'ORACLE';
 ##########################################################################
 #      Test verifies Gtid_log_event/Xid_log_event specific print         #

mysql-test/suite/compat/oracle/t/binlog_ptr_mysqlbinlog.test

@@ -18,6 +18,8 @@
 --source include/have_log_bin.inc
 --source include/have_innodb.inc
 
+call mtr.add_suppression("Binlog_format changed to.*flashback");
+
 let $MYSQLD_DATADIR= `select @@datadir`;
 SET @@SQL_MODE = 'ORACLE';
 

mysql-test/suite/galera/r/basic.result

 connection node_2;
 connection node_1;
+select @@slave_connections_needed_for_purge;
+@@slave_connections_needed_for_purge
+0
+select VARIABLE_NAME, GLOBAL_VALUE, GLOBAL_VALUE_ORIGIN from information_schema.system_variables where variable_name="slave_connections_needed_for_purge";
+VARIABLE_NAME	GLOBAL_VALUE	GLOBAL_VALUE_ORIGIN
+SLAVE_CONNECTIONS_NEEDED_FOR_PURGE	0	AUTO
 USE test;
 CREATE TABLE t1(c1 INT PRIMARY KEY) ENGINE=INNODB;
 INSERT INTO t1 VALUES (1), (2), (3), (4), (5);

mysql-test/suite/galera/t/basic.test

 --source include/galera_cluster.inc
 --source include/have_innodb.inc
 
+select @@slave_connections_needed_for_purge;
+select VARIABLE_NAME, GLOBAL_VALUE, GLOBAL_VALUE_ORIGIN from information_schema.system_variables where variable_name="slave_connections_needed_for_purge";
+
 USE test;
 CREATE TABLE t1(c1 INT PRIMARY KEY) ENGINE=INNODB;
 INSERT INTO t1 VALUES (1), (2), (3), (4), (5);

mysql-test/suite/rpl/r/purge_binlog.result

+include/master-slave.inc
+[connection master]
+#
+# MDEV-34504 PURGE BINARY LOGS not working anymore
+#
+select @@slave_connections_needed_for_purge;
+@@slave_connections_needed_for_purge
+0
+set @old_dbug= @@global.debug_dbug;
+create table t1 (a int, b varchar(32768));
+insert into t1 values(1,repeat("a",32768));
+connection slave;
+select a from t1;
+a
+1
+set @@global.debug_dbug= "+d,pause_before_io_read_event";
+connection master;
+insert into t1 values(2,repeat("b",32768));
+insert into t1 values(3,repeat("c",32768));
+connection slave;
+set debug_sync='now wait_for io_thread_at_read_event';
+select a from t1;
+a
+1
+connection master;
+FLUSH BINARY LOGS;
+show binary logs;
+Log_name	File_size
+master-bin.000001	#
+master-bin.000002	#
+PURGE BINARY LOGS TO 'master-bin.000002';
+Warnings:
+Note	1375	Binary log XXX is not purged because it is in use by a slave thread
+show binary logs;
+Log_name	File_size
+master-bin.000001	#
+master-bin.000002	#
+connection slave;
+set @@global.debug_dbug= @old_dbug;
+set debug_sync='now signal io_thread_continue_read_event';
+connection master;
+connection slave;
+select count(*) from t1;
+count(*)
+153
+connection master;
+PURGE BINARY LOGS TO 'master-bin.000002';
+show binary logs;
+Log_name	File_size
+master-bin.000002	#
+drop table t1;
+include/rpl_end.inc

mysql-test/suite/rpl/r/rpl_rotate_logs.result

@@ -70,6 +70,8 @@ master-bin.000002	#
 master-bin.000003	#
 SELECT @time_for_purge:=DATE_ADD('tmpval', INTERVAL 1 SECOND);
 purge master logs before (@time_for_purge);
+Warnings:
+Note	1375	Binary log 'master-bin.000003' is not purged because it is the current active binlog
 show binary logs;
 Log_name	File_size
 master-bin.000003	#

mysql-test/suite/rpl/t/purge_binlog.test

+--source include/have_debug.inc
+--source include/have_debug_sync.inc
+--source include/master-slave.inc
+--source include/have_binlog_format_row.inc
+
+--echo #
+--echo # MDEV-34504 PURGE BINARY LOGS not working anymore
+--echo #
+
+select @@slave_connections_needed_for_purge;
+set @old_dbug= @@global.debug_dbug;
+
+create table t1 (a int, b varchar(32768));
+insert into t1 values(1,repeat("a",32768));
+--sync_slave_with_master
+select a from t1;
+set @@global.debug_dbug= "+d,pause_before_io_read_event";
+--connection master
+insert into t1 values(2,repeat("b",32768));
+insert into t1 values(3,repeat("c",32768));
+--connection slave
+set debug_sync='now wait_for io_thread_at_read_event';
+select a from t1;
+--connection master
+--disable_query_log
+let $i=150;
+while ($i)
+{
+--eval insert into t1 values($i+4,repeat(char(64+$i),32768));
+--dec $i
+}
+--enable_query_log
+
+FLUSH BINARY LOGS;
+--source include/show_binary_logs.inc
+--let $purge_to_binlog= query_get_value(SHOW MASTER STATUS, File, 1)
+--replace_regex /Binary log.*is not/Binary log XXX is not/
+--eval PURGE BINARY LOGS TO '$purge_to_binlog'
+--source include/show_binary_logs.inc
+--connection slave
+set @@global.debug_dbug= @old_dbug;
+set debug_sync='now signal io_thread_continue_read_event';
+--connection master
+--sync_slave_with_master
+select count(*) from t1;
+--connection master
+--eval PURGE BINARY LOGS TO '$purge_to_binlog'
+--source include/show_binary_logs.inc
+drop table t1;
+
+--source include/rpl_end.inc

mysql-test/suite/rpl/t/rpl_rotate_logs.test

@@ -190,10 +190,12 @@ sync_slave_with_master;
 #
 --error 1220
 show binlog events in 'non existing_binlog_file';
+--disable_warnings
 purge master logs before now();
 --error 1220
 show binlog events in '';
 purge master logs before now();
+--enable_warnings
 
 --echo End of 5.0 tests
 --echo #cleanup

mysql-test/suite/sys_vars/r/sysvars_server_embedded.result

@@ -3435,7 +3435,7 @@ COMMAND_LINE_ARGUMENT	OPTIONAL
 VARIABLE_NAME	SLAVE_CONNECTIONS_NEEDED_FOR_PURGE
 VARIABLE_SCOPE	GLOBAL
 VARIABLE_TYPE	INT UNSIGNED
-VARIABLE_COMMENT	Minimum number of connected slaves required for automatic binary log purge with max_binlog_total_size, binlog_expire_logs_seconds or binlog_expire_logs_days.
+VARIABLE_COMMENT	Minimum number of connected slaves required for automatic binary log purge with max_binlog_total_size, binlog_expire_logs_seconds or binlog_expire_logs_days. Default is 0 when Galera is enabled and 1 otherwise.
 NUMERIC_MIN_VALUE	0
 NUMERIC_MAX_VALUE	4294967295
 NUMERIC_BLOCK_SIZE	1

mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result

@@ -4005,7 +4005,7 @@ COMMAND_LINE_ARGUMENT	OPTIONAL
 VARIABLE_NAME	SLAVE_CONNECTIONS_NEEDED_FOR_PURGE
 VARIABLE_SCOPE	GLOBAL
 VARIABLE_TYPE	INT UNSIGNED
-VARIABLE_COMMENT	Minimum number of connected slaves required for automatic binary log purge with max_binlog_total_size, binlog_expire_logs_seconds or binlog_expire_logs_days.
+VARIABLE_COMMENT	Minimum number of connected slaves required for automatic binary log purge with max_binlog_total_size, binlog_expire_logs_seconds or binlog_expire_logs_days. Default is 0 when Galera is enabled and 1 otherwise.
 NUMERIC_MIN_VALUE	0
 NUMERIC_MAX_VALUE	4294967295
 NUMERIC_BLOCK_SIZE	1

sql/log.cc

@@ -4794,8 +4794,8 @@ int MYSQL_BIN_LOG::purge_first_log(Relay_log_info* rli, bool included)
 
   DBUG_EXECUTE_IF("crash_before_purge_logs", DBUG_SUICIDE(););
 
-  rli->relay_log.purge_logs(to_purge_if_included, included,
-                            0, 0, &log_space_reclaimed);
+  rli->relay_log.purge_logs(current_thd, to_purge_if_included, included,
+                            0, 0, 0, &log_space_reclaimed);
 
   mysql_mutex_lock(&rli->log_space_lock);
   rli->log_space_total-= log_space_reclaimed;
@@ -4862,16 +4862,17 @@ int MYSQL_BIN_LOG::update_log_index(LOG_INFO* log_info, bool need_update_threads
                                 mysql_file_stat() or mysql_file_delete()
 */
 
-int MYSQL_BIN_LOG::purge_logs(const char *to_log, 
+int MYSQL_BIN_LOG::purge_logs(THD *thd,
+                              const char *to_log,
                               bool included,
                               bool need_mutex, 
-                              bool need_update_threads, 
+                              bool need_update_threads,
+                              bool interactive,
                               ulonglong *reclaimed_space)
 {
   int error= 0;
   bool exit_loop= 0;
   LOG_INFO log_info;
-  THD *thd= current_thd;
   DBUG_ENTER("purge_logs");
   DBUG_PRINT("info",("to_log= %s",to_log));
 
@@ -4897,7 +4898,7 @@ int MYSQL_BIN_LOG::purge_logs(const char *to_log,
   if (unlikely((error=find_log_pos(&log_info, NullS, 0 /*no mutex*/))))
     goto err;
   while ((strcmp(to_log,log_info.log_file_name) || (exit_loop=included)) &&
-         can_purge_log(log_info.log_file_name))
+         can_purge_log(log_info.log_file_name, interactive))
   {
     if (unlikely((error= register_purge_index_entry(log_info.log_file_name))))
     {
@@ -5246,13 +5247,13 @@ int MYSQL_BIN_LOG::purge_index_entry(THD *thd, ulonglong *reclaimed_space,
                                 mysql_file_stat() or mysql_file_delete()
 */
 
-int MYSQL_BIN_LOG::purge_logs_before_date(time_t purge_time)
+int MYSQL_BIN_LOG::purge_logs_before_date(THD *thd, time_t purge_time,
+                                          bool interactive)
 {
   int error;
   char to_log[FN_REFLEN];
   LOG_INFO log_info;
   MY_STAT stat_area;
-  THD *thd= current_thd;
   DBUG_ENTER("purge_logs_before_date");
 
   mysql_mutex_lock(&LOCK_index);
@@ -5261,7 +5262,7 @@ int MYSQL_BIN_LOG::purge_logs_before_date(time_t purge_time)
   if (unlikely((error=find_log_pos(&log_info, NullS, 0 /*no mutex*/))))
     goto err;
 
-  while (can_purge_log(log_info.log_file_name))
+  for (;;)
   {
     if (!mysql_file_stat(m_key_file_log,
                          log_info.log_file_name, &stat_area, MYF(0)))
@@ -5299,7 +5300,8 @@ int MYSQL_BIN_LOG::purge_logs_before_date(time_t purge_time)
     }
     else
     {
-      if (stat_area.st_mtime >= purge_time)
+      if (stat_area.st_mtime >= purge_time ||
+          !can_purge_log(log_info.log_file_name, interactive))
         break;
       strmake_buf(to_log, log_info.log_file_name);
     }
@@ -5310,7 +5312,7 @@ int MYSQL_BIN_LOG::purge_logs_before_date(time_t purge_time)
   if (to_log[0])
   {
     ulonglong reclaimed_space= 0;
-    error= purge_logs(to_log, 1, 0, 1, &reclaimed_space);
+    error= purge_logs(thd, to_log, 1, 0, 1, interactive, &reclaimed_space);
     binlog_space_total-= reclaimed_space;
   }
 
@@ -5341,6 +5343,7 @@ int MYSQL_BIN_LOG::real_purge_logs_by_size(ulonglong binlog_pos)
   MY_STAT stat_area;
   char to_log[FN_REFLEN];
   ulonglong found_space= 0;
+  THD *thd= current_thd;
   DBUG_ENTER("real_purge_logs_by_size");
 
   mysql_mutex_lock(&LOCK_index);
@@ -5354,7 +5357,7 @@ int MYSQL_BIN_LOG::real_purge_logs_by_size(ulonglong binlog_pos)
     goto err;
 
   to_log[0] = 0;
-  while (can_purge_log(log_info.log_file_name))
+  while (can_purge_log(log_info.log_file_name, 0))
   {
     if (!mysql_file_stat(m_key_file_log, log_info.log_file_name, &stat_area,
                          MYF(0)))
@@ -5364,7 +5367,6 @@ int MYSQL_BIN_LOG::real_purge_logs_by_size(ulonglong binlog_pos)
         /*
           Other than ENOENT are fatal
         */
-        THD *thd = current_thd;
         if (thd)
         {
           push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
@@ -5397,8 +5399,9 @@ int MYSQL_BIN_LOG::real_purge_logs_by_size(ulonglong binlog_pos)
   if (found_space)
   {
     ulonglong reclaimed_space= 0;
-    purge_logs(to_log, true, false /*need_lock_index=false*/,
+    purge_logs(thd, to_log, true, false /*need_lock_index=false*/,
                true /*need_update_threads=true*/,
+               false /* not interactive */,
                &reclaimed_space);
     DBUG_ASSERT(reclaimed_space == found_space);
     binlog_space_total-= reclaimed_space;
@@ -5417,6 +5420,12 @@ int MYSQL_BIN_LOG::real_purge_logs_by_size(ulonglong binlog_pos)
 }
 
 /*
+  @param  log_file_name_arg  Name of log file to check
+  @param  interactive        True if called by a PURGE BINLOG command.
+
+  @return 0 Log cannot be removed
+  @return 1 Log can be removed
+
   The following variables are here to allows us to quickly check if
   the can_purge_log(log_file_name_arg) name will fail in the
   'log_in_use' call.
@@ -5431,18 +5440,23 @@ int MYSQL_BIN_LOG::real_purge_logs_by_size(ulonglong binlog_pos)
 static bool waiting_for_slave_to_change_binlog= 0;
 static ulonglong purge_sending_new_binlog_file= 0;
 static char purge_binlog_name[FN_REFLEN];
+static bool purge_warning_given= 0;
 
 bool
-MYSQL_BIN_LOG::can_purge_log(const char *log_file_name_arg)
+MYSQL_BIN_LOG::can_purge_log(const char *log_file_name_arg,
+                             bool interactive)
 {
-  THD *thd= current_thd;                        // May be NULL at startup
-  bool res;
+  int res;
+  const char *reason;
 
   if (is_active(log_file_name_arg) ||
       (!is_relay_log && waiting_for_slave_to_change_binlog &&
        purge_sending_new_binlog_file == sending_new_binlog_file &&
        !strcmp(log_file_name_arg, purge_binlog_name)))
-      return false;
+  {
+    reason= "it is the current active binlog";
+    goto error;
+  }
 
   DBUG_ASSERT(!is_relay_log || binlog_xid_count_list.is_empty());
   if (!is_relay_log)
@@ -5458,7 +5472,10 @@ MYSQL_BIN_LOG::can_purge_log(const char *log_file_name_arg)
     }
     mysql_mutex_unlock(&LOCK_xid_list);
     if (b)
-      return false;
+    {
+      reason= "it may be needed for crash recovery (XID)";
+      goto error;
+    }
   }
 
   if (!is_relay_log)
@@ -5467,8 +5484,7 @@ MYSQL_BIN_LOG::can_purge_log(const char *log_file_name_arg)
     purge_sending_new_binlog_file= sending_new_binlog_file;
   }
   if ((res= log_in_use(log_file_name_arg,
-                       (is_relay_log ||
-                        (thd && thd->lex->sql_command == SQLCOM_PURGE)) ?
+                       (is_relay_log || interactive) ?
                        0 : slave_connections_needed_for_purge)))
   {
     if (!is_relay_log)
@@ -5476,9 +5492,39 @@ MYSQL_BIN_LOG::can_purge_log(const char *log_file_name_arg)
       waiting_for_slave_to_change_binlog= 1;
       strmake(purge_binlog_name, log_file_name_arg,
               sizeof(purge_binlog_name)-1);
+      if (res == 1)
+        reason= "it is in use by a slave thread";
+      else
+        reason= "less than 'slave_connections_needed_for_purge' slaves have "
+          "processed it";
+      goto error;
     }
   }
-  return !res;
+  /* We can purge this file, reset for next failure */
+  purge_warning_given= 0;
+  return 1;
+
+error:
+  if (!is_relay_log && (interactive || !purge_warning_given))
+  {
+    /* Remove directory (to keep things shorter and compatible */
+    log_file_name_arg+= dirname_length(log_file_name_arg);
+
+    /* purge_warning_given is reset after next sucessful purge */
+    purge_warning_given= 1;
+    if (interactive)
+    {
+      my_printf_error(ER_BINLOG_PURGE_PROHIBITED,
+                      "Binary log '%s' is not purged because %s",
+                      MYF(ME_NOTE), log_file_name_arg, reason);
+    }
+    else
+    {
+      sql_print_information("Binary log '%s' is not purged because %s",
+                            log_file_name_arg, reason);
+    }
+  }
+  return 0;
 }
 #endif /* HAVE_REPLICATION */
 
@@ -7614,16 +7660,17 @@ void MYSQL_BIN_LOG::purge(bool all)
 {
   mysql_mutex_assert_not_owner(&LOCK_log);
 #ifdef HAVE_REPLICATION
+  THD *thd= current_thd;
   if (binlog_expire_logs_seconds)
   {
-    DEBUG_SYNC(current_thd, "at_purge_logs_before_date");
+    DEBUG_SYNC(thd, "at_purge_logs_before_date");
     time_t purge_time= my_time(0) - binlog_expire_logs_seconds;
     DBUG_EXECUTE_IF("expire_logs_always", { purge_time = my_time(0); });
     if (purge_time >= 0)
     {
-      purge_logs_before_date(purge_time);
+      purge_logs_before_date(thd, purge_time, 0);
     }
-    DEBUG_SYNC(current_thd, "after_purge_logs_before_date");
+    DEBUG_SYNC(thd, "after_purge_logs_before_date");
   }
   if (all && binlog_space_limit)
   {

sql/log.h

@@ -1035,7 +1035,7 @@ class MYSQL_BIN_LOG: public TC_LOG, private Event_log
   void mark_xid_done(ulong cookie, bool write_checkpoint);
   void make_log_name(char* buf, const char* log_ident);
   bool is_active(const char* log_file_name);
-  bool can_purge_log(const char *log_file_name);
+  bool can_purge_log(const char *log_file_name, bool interactive);
   int update_log_index(LOG_INFO* linfo, bool need_update_threads);
   int rotate(bool force_rotate, bool* check_purge);
   void checkpoint_and_purge(ulong binlog_id);
@@ -1054,10 +1054,10 @@ class MYSQL_BIN_LOG: public TC_LOG, private Event_log
      @retval other Failure
   */
   bool flush_and_sync(bool *synced);
-  int purge_logs(const char *to_log, bool included,
-                 bool need_mutex, bool need_update_threads,
+  int purge_logs(THD *thd, const char *to_log, bool included,
+                 bool need_mutex, bool need_update_threads, bool interactive,
                  ulonglong *decrease_log_space);
-  int purge_logs_before_date(time_t purge_time);
+  int purge_logs_before_date(THD *thd, time_t purge_time, bool interactive);
   int purge_first_log(Relay_log_info* rli, bool included);
   int count_binlog_space();
   void count_binlog_space_with_mutex()

sql/mysqld.cc

@@ -521,6 +521,7 @@ uint default_password_lifetime;
 my_bool disconnect_on_expired_password;
 
 bool max_user_connections_checking=0;
+
 /**
   Limit of the total number of prepared statements in the server.
   Is necessary to protect the server against out-of-memory attacks.
@@ -5869,7 +5870,28 @@ int mysqld_main(int argc, char **argv)
 #ifdef WITH_WSREP
   wsrep_set_wsrep_on(nullptr);
   if (WSREP_ON && wsrep_check_opts()) unireg_abort(1);
-#endif
+
+  if (!opt_bootstrap && WSREP_PROVIDER_EXISTS && WSREP_ON)
+  {
+    if (global_system_variables.binlog_format != BINLOG_FORMAT_ROW)
+    {
+      sql_print_information("Binlog_format changed to \"ROW\" because of "
+                            "Galera");
+      SYSVAR_AUTOSIZE(global_system_variables.binlog_format, BINLOG_FORMAT_ROW);
+    }
+    binlog_format_used= 1;
+    if (IS_SYSVAR_AUTOSIZE(&internal_slave_connections_needed_for_purge))
+    {
+      slave_connections_needed_for_purge=
+        internal_slave_connections_needed_for_purge= 0;
+      SYSVAR_AUTOSIZE(internal_slave_connections_needed_for_purge, 0);
+      sql_print_information(
+      "slave_connections_needed_for_purge changed to 0 because "
+      "of Galera. Change it to 1 or higher if this Galera node "
+      "is also Master in a normal replication setup");
+    }
+  }
+#endif /* WITH_WSREP */
 
 #ifdef _WIN32
   /* 
@@ -8241,7 +8263,6 @@ mysqld_get_one_option(const struct my_option *opt, const char *argument,
       ((enum_slave_parallel_mode)opt_slave_parallel_mode);
     break;
   }
-
   case (int)OPT_BINLOG_IGNORE_DB:
   {
     binlog_filter->add_ignore_db(argument);
@@ -8744,18 +8765,14 @@ static int get_options(int *argc_ptr, char ***argv_ptr)
     opt_bin_log= opt_bin_log_used= 1;
 
     /* Force format to row */
+    if (global_system_variables.binlog_format != BINLOG_FORMAT_ROW)
+    {
+      sql_print_information("Binlog_format changed to \"ROW\" because of "
+                            "flashback");
+      SYSVAR_AUTOSIZE(global_system_variables.binlog_format,
+                      BINLOG_FORMAT_ROW);
+    }
     binlog_format_used= 1;
-    global_system_variables.binlog_format= BINLOG_FORMAT_ROW;
-  }
-
-  if (!opt_bootstrap && WSREP_PROVIDER_EXISTS && WSREP_ON &&
-      global_system_variables.binlog_format != BINLOG_FORMAT_ROW)
-  {
-
-    WSREP_ERROR ("Only binlog_format = 'ROW' is currently supported. "
-                 "Configured value: '%s'. Please adjust your configuration.",
-                 binlog_format_names[global_system_variables.binlog_format]);
-    return 1;
   }
 
   // Synchronize @@global.autocommit on --autocommit

sql/sql_repl.cc

@@ -612,14 +612,26 @@ static my_bool log_in_use_callback(THD *thd, st_log_in_use *arg)
 }
 
 
-bool log_in_use(const char* log_name, uint min_connected)
+/*
+  Check if a log is in use.
+
+  @return 0  Not used
+  @return 1  A slave is reading from the log
+  @return 2  There are less than 'min_connected' slaves that
+             has recived the log.
+*/
+
+int log_in_use(const char* log_name, uint min_connected)
 {
   st_log_in_use arg;
   arg.log_name= log_name;
   arg.connected_slaves= 0;
 
-  return ((server_threads.iterate(log_in_use_callback, &arg) ||
-           arg.connected_slaves < min_connected));
+  if (server_threads.iterate(log_in_use_callback, &arg))
+    return 1;
+  if (arg.connected_slaves < min_connected)
+    return 2;
+  return 0;
 }
 
 
@@ -659,8 +671,8 @@ bool purge_master_logs(THD* thd, const char* to_log)
 
   mysql_bin_log.make_log_name(search_file_name, to_log);
   return purge_error_message(thd,
-			     mysql_bin_log.purge_logs(search_file_name, 0, 1,
-						      1, NULL));
+			     mysql_bin_log.purge_logs(thd, search_file_name,
+                                                      0, 1, 1, 1, NULL));
 }
 
 
@@ -683,7 +695,9 @@ bool purge_master_logs_before_date(THD* thd, time_t purge_time)
     return 0;
   }
   return purge_error_message(thd,
-                             mysql_bin_log.purge_logs_before_date(purge_time));
+                             mysql_bin_log.purge_logs_before_date(thd,
+                                                                  purge_time,
+                                                                  1));
 }
 
 void set_read_error(binlog_send_info *info, int error)

sql/sql_repl.h

@@ -38,7 +38,7 @@ int reset_master(THD* thd, rpl_gtid *init_state, uint32 init_state_len,
                  ulong next_log_number);
 bool purge_master_logs(THD* thd, const char* to_log);
 bool purge_master_logs_before_date(THD* thd, time_t purge_time);
-bool log_in_use(const char* log_name, uint min_connections);
+int log_in_use(const char* log_name, uint min_connections);
 void adjust_linfo_offsets(my_off_t purge_offset);
 void show_binlogs_get_fields(THD *thd, List<Item> *field_list);
 bool show_binlogs(THD* thd);

sql/sys_vars.cc

@@ -704,6 +704,7 @@ Sys_binlog_format(
        NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(binlog_format_check),
        ON_UPDATE(fix_binlog_format_after_update));
 
+
 static bool binlog_direct_check(sys_var *self, THD *thd, set_var *var)
 {
   if (var->type == OPT_GLOBAL)
@@ -1265,7 +1266,7 @@ static bool update_binlog_space_limit(sys_var *, THD *,
       mysql_bin_log.count_binlog_space();
     /* Inform can_purge_log() that it should do a recheck of log_in_use() */
     sending_new_binlog_file++;
-     mysql_bin_log.unlock_index();
+    mysql_bin_log.unlock_index();
     mysql_bin_log.purge(1);
     return 0;
   }
@@ -1274,6 +1275,7 @@ static bool update_binlog_space_limit(sys_var *, THD *,
   return 0;
 }
 
+
 static Sys_var_on_access_global<Sys_var_ulonglong,
                                 PRIV_SET_SYSTEM_GLOBAL_VAR_MAX_BINLOG_CACHE_SIZE>
 Sys_max_binlog_total_size(
@@ -1303,13 +1305,15 @@ Sys_slave_connections_needed_for_purge(
       "slave_connections_needed_for_purge",
       "Minimum number of connected slaves required for automatic binary "
       "log purge with max_binlog_total_size, binlog_expire_logs_seconds "
-      "or binlog_expire_logs_days.",
+      "or binlog_expire_logs_days. Default is 0 when Galera is enabled and 1 "
+      "otherwise.",
        GLOBAL_VAR(internal_slave_connections_needed_for_purge),
        CMD_LINE(REQUIRED_ARG),
        VALID_RANGE(0, UINT_MAX), DEFAULT(1), BLOCK_SIZE(1),
        NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(0),
        ON_UPDATE(update_binlog_space_limit));
 
+
 static Sys_var_mybool Sys_flush(
        "flush", "Flush MyISAM tables to disk between SQL commands",
        GLOBAL_VAR(myisam_flush),