From 8f02ab0fedb15d3aa57a699e08bd35f7d429c459 Mon Sep 17 00:00:00 2001
From: rj03hou <rj03hou@gmail.com>
Date: Mon, 5 Dec 2016 19:42:16 +0800
Subject: [PATCH] check the slave status when find recursive find the master,
 so support if the dba using reset slave instead of reset slave all.

---
 go/logic/inspect.go |  1 +
 go/mysql/utils.go   | 18 ++++++++++++++++++
 2 files changed, 19 insertions(+)

diff --git a/go/logic/inspect.go b/go/logic/inspect.go
index 4519225..9b1ff5e 100644
--- a/go/logic/inspect.go
+++ b/go/logic/inspect.go
@@ -692,6 +692,7 @@ func (this *Inspector) readChangelogState() (map[string]string, error) {
 }
 
 func (this *Inspector) getMasterConnectionConfig() (applierConfig *mysql.ConnectionConfig, err error) {
+	log.Infof("Start find master, recursive to execute show slave status from the host to find the master.")
 	visitedKeys := mysql.NewInstanceKeyMap()
 	return mysql.GetMasterConnectionConfigSafe(this.connectionConfig, visitedKeys, this.migrationContext.AllowedMasterMaster)
 }
diff --git a/go/mysql/utils.go b/go/mysql/utils.go
index ae22e3a..de12e46 100644
--- a/go/mysql/utils.go
+++ b/go/mysql/utils.go
@@ -83,12 +83,30 @@ func GetMasterKeyFromSlaveStatus(connectionConfig *ConnectionConfig) (masterKey
 		return nil, err
 	}
 	err = sqlutils.QueryRowsMap(db, `show slave status`, func(rowMap sqlutils.RowMap) error {
+		// Using reset slave not reset slave all, the Master_Log_File is empty.
+		if rowMap.GetString("Master_Log_File") == "" {
+			return nil
+		}
+
+		slaveIORunning := rowMap.GetString("Slave_IO_Running")
+		slaveSQLRunning := rowMap.GetString("Slave_SQL_Running")
+
+		// If not using reset slave or reset slave all and slave is break, something is wrong, so we stop.
+		if slaveIORunning != "Yes" || slaveSQLRunning != "Yes" {
+			return fmt.Errorf("The slave %s is break: Slave_IO_Running:%s Slave_SQL_Running:%s please make sure the replication is correct.",
+				connectionConfig.Key.Hostname,
+				slaveIORunning,
+				slaveSQLRunning,
+			)
+		}
+
 		masterKey = &InstanceKey{
 			Hostname: rowMap.GetString("Master_Host"),
 			Port:     rowMap.GetInt("Master_Port"),
 		}
 		return nil
 	})
+
 	return masterKey, err
 }