Handle SQL thread crash in vt/vttablet/tabletserver/repltracker

go/mysql/flavor.go

@@ -299,8 +299,12 @@ func parseReplicationStatus(fields map[string]string) ReplicationStatus {
 	status.MasterPort = int(parseInt)
 	parseInt, _ = strconv.ParseInt(fields["Connect_Retry"], 10, 0)
 	status.MasterConnectRetry = int(parseInt)
-	parseUint, _ := strconv.ParseUint(fields["Seconds_Behind_Master"], 10, 0)
+	parseUint, _ := strconv.ParseUint(fields["Last_SQL_Errno"], 10, 0)
+	status.LastSQLErrno = uint(parseUint)
+	parseUint, _ = strconv.ParseUint(fields["Seconds_Behind_Master"], 10, 0)
 	status.SecondsBehindMaster = uint(parseUint)
+	parseUint, _ = strconv.ParseUint(fields["Skip_Counter"], 10, 0)
+	status.SkipCounter = uint(parseUint)
 	parseUint, _ = strconv.ParseUint(fields["Master_Server_Id"], 10, 0)
 	status.MasterServerID = uint(parseUint)
 

go/mysql/replication_status.go

@@ -36,7 +36,9 @@ type ReplicationStatus struct {
 	MasterServerID       uint
 	IOThreadRunning      bool
 	SQLThreadRunning     bool
+	LastSQLErrno         uint
 	SecondsBehindMaster  uint
+	SkipCounter          uint
 	MasterHost           string
 	MasterPort           int
 	MasterConnectRetry   int
@@ -49,6 +51,14 @@ func (s *ReplicationStatus) ReplicationRunning() bool {
 	return s.IOThreadRunning && s.SQLThreadRunning
 }
 
+// HasReplicationSQLThreadError returns true if the replication SQL thread stopped on an
+// error (ie: Slave_SQL_Running: no + Last_SQL_Errno > 0) and sql_slave_skip_counter is
+// disabled (0). This suggests the replication SQL thread stopped on an error that can't
+// be recovered/skipped automatically. A node in this state may have inconsistent data.
+func (s *ReplicationStatus) HasReplicationSQLThreadError() bool {
+	return !s.SQLThreadRunning && s.LastSQLErrno > 0 && s.SkipCounter == 0
+}
+
 // ReplicationStatusToProto translates a Status to proto3.
 func ReplicationStatusToProto(s ReplicationStatus) *replicationdatapb.Status {
 	return &replicationdatapb.Status{
@@ -59,6 +69,8 @@ func ReplicationStatusToProto(s ReplicationStatus) *replicationdatapb.Status {
 		MasterServerId:       uint32(s.MasterServerID),
 		IoThreadRunning:      s.IOThreadRunning,
 		SqlThreadRunning:     s.SQLThreadRunning,
+		LastSqlErrno:         uint32(s.LastSQLErrno),
+		SkipCounter:          uint32(s.SkipCounter),
 		SecondsBehindMaster:  uint32(s.SecondsBehindMaster),
 		MasterHost:           s.MasterHost,
 		MasterPort:           int32(s.MasterPort),

go/mysql/replication_status_test.go

@@ -42,6 +42,17 @@ func TestStatusIOThreadNotRunning(t *testing.T) {
 	}
 }
 
+func TestStatusHasReplicationSQLThreadError(t *testing.T) {
+	input := &ReplicationStatus{
+		LastSQLErrno:     1032,
+		SQLThreadRunning: false,
+	}
+	want := true
+	if got := input.HasReplicationSQLThreadError(); got != want {
+		t.Errorf("%v#v.HasReplicationSQLThreadError() = %v, want %v", input, got, want)
+	}
+}
+
 func TestStatusSQLThreadNotRunning(t *testing.T) {
 	input := &ReplicationStatus{
 		IOThreadRunning:  true,

go/vt/vttablet/tabletserver/repltracker/poller.go

@@ -46,7 +46,9 @@ func (p *poller) Status() (time.Duration, error) {
 		return 0, err
 	}
 
-	if !status.ReplicationRunning() {
+	if status.HasReplicationSQLThreadError() {
+		return 0, vterrors.Errorf(vtrpcpb.Code_UNAVAILABLE, "replication sql thread error")
+	} else if !status.ReplicationRunning() {
 		if p.timeRecorded.IsZero() {
 			return 0, vterrors.Errorf(vtrpcpb.Code_UNAVAILABLE, "replication is not running")
 		}

proto/replicationdata.proto

@@ -37,6 +37,8 @@ message Status {
   string file_relay_log_position = 10;
   uint32 master_server_id = 11;
   string master_uuid = 12;
+  uint32 last_sql_errno = 13;
+  uint32 skip_counter = 14;
 }
 
 // StopReplicationStatus represents the replication status before calling StopReplication, and the replication status collected immediately after