From 0493196ecce1a153b33d3290df09a4c0cbc9c685 Mon Sep 17 00:00:00 2001
From: Johan Wikman <johan.wikman@mariadb.com>
Date: Wed, 30 Mar 2016 16:05:19 +0300
Subject: [PATCH] Disconnect slave if duplicate event detected

If a duplicate event is detected the state of the slave is set
to BLRS_ERRORED and the connection is closed. That way the
duplicate event will not break the slave, and it will pick
up its state when it reconnects.
---
 server/modules/routing/binlog/blr_master.c | 40 +++++++++++++++-------
 server/modules/routing/binlog/blr_slave.c  | 17 +++++++++
 2 files changed, 44 insertions(+), 13 deletions(-)

diff --git a/server/modules/routing/binlog/blr_master.c b/server/modules/routing/binlog/blr_master.c
index 9da0f2734..020ab0f53 100644
--- a/server/modules/routing/binlog/blr_master.c
+++ b/server/modules/routing/binlog/blr_master.c
@@ -1981,24 +1981,37 @@ blr_distribute_binlog_record(ROUTER_INSTANCE *router, REP_HEADER *hdr, uint8_t *
                     blr_slave_rotate(router, slave, ptr);
                 }
 
-                blr_send_event(BLR_THREAD_ROLE_MASTER, binlog_name, binlog_pos, slave, hdr, ptr);
-
-                spinlock_acquire(&slave->catch_lock);
-                if (hdr->event_type != ROTATE_EVENT)
+                if (blr_send_event(BLR_THREAD_ROLE_MASTER, binlog_name, binlog_pos, slave, hdr, ptr))
                 {
-                    slave->binlog_pos = hdr->next_pos;
-                }
-                if (slave->overrun)
-                {
-                    slave->stats.n_overrun++;
-                    slave->overrun = 0;
-                    poll_fake_write_event(slave->dcb);
+                    spinlock_acquire(&slave->catch_lock);
+                    if (hdr->event_type != ROTATE_EVENT)
+                    {
+                        slave->binlog_pos = hdr->next_pos;
+                    }
+                    if (slave->overrun)
+                    {
+                        slave->stats.n_overrun++;
+                        slave->overrun = 0;
+                        poll_fake_write_event(slave->dcb);
+                    }
+                    else
+                    {
+                        slave->cstate &= ~CS_BUSY;
+                    }
+                    spinlock_release(&slave->catch_lock);
                 }
                 else
                 {
-                    slave->cstate &= ~CS_BUSY;
+                    MXS_WARNING("Slave %s:%i, server-id %d, binlog '%s, position %u: "
+                                "Master-thread could not send event to slave, closing connection.",
+                                slave->dcb->remote,
+                                ntohs((slave->dcb->ipv4).sin_port),
+                                slave->serverid,
+                                binlog_name,
+                                binlog_pos);
+                    slave->state = BLRS_ERRORED;
+                    dcb_close(slave->dcb);
                 }
-                spinlock_release(&slave->catch_lock);
                 break;
 
             case SLAVE_EVENT_ALREADY_SENT:
@@ -2677,6 +2690,7 @@ bool blr_send_event(blr_thread_role_t role,
                   role == BLR_THREAD_ROLE_MASTER ? "master" : "slave",
                   slave->lsi_sender_tid,
                   slave->lsi_sender_role == BLR_THREAD_ROLE_MASTER ? "master" : "slave");
+        return false;
     }
 
     /** Check if the event and the OK byte fit into a single packet  */
diff --git a/server/modules/routing/binlog/blr_slave.c b/server/modules/routing/binlog/blr_slave.c
index fcf0d7ca5..c30ceaeb4 100644
--- a/server/modules/routing/binlog/blr_slave.c
+++ b/server/modules/routing/binlog/blr_slave.c
@@ -2303,6 +2303,23 @@ blr_slave_catchup(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, bool large)
             slave->stats.n_events++;
             burst_size -= hdr.event_size;
         }
+        else
+        {
+            MXS_WARNING("Slave %s:%i, server-id %d, binlog '%s, position %u: "
+                        "Slave-thread could not send event to slave, closing connection.",
+                        slave->dcb->remote,
+                        ntohs((slave->dcb->ipv4).sin_port),
+                        slave->serverid,
+                        binlog_name,
+                        binlog_pos);
+#ifndef BLFILE_IN_SLAVE
+            blr_close_binlog(router, file);
+#endif
+            slave->state = BLRS_ERRORED;
+            dcb_close(slave->dcb);
+            return 0;
+        }
+
         gwbuf_free(record);
         record = NULL;