One TTL task losing heartbeat will block other tasks from getting heartbeat

[YangKeao]

See the following codes:

// updateHeartBeat updates the heartbeat for all tasks with current instance as owner
func (m *taskManager) updateHeartBeat(ctx context.Context, se session.Session, now time.Time) error {
	for _, task := range m.runningTasks {
		state := &cache.TTLTaskState{
			TotalRows:   task.statistics.TotalRows.Load(),
			SuccessRows: task.statistics.SuccessRows.Load(),
			ErrorRows:   task.statistics.ErrorRows.Load(),
		}
		if task.result != nil && task.result.err != nil {
			state.ScanTaskErr = task.result.err.Error()
		}

		intest.Assert(se.GetSessionVars().Location().String() == now.Location().String())
		sql, args, err := updateTTLTaskHeartBeatSQL(task.JobID, task.ScanID, now, state, m.id)
		if err != nil {
			return err
		}
		_, err = se.ExecuteSQL(ctx, sql, args...)
		if err != nil {
			return errors.Wrapf(err, "execute sql: %s", sql)
		}

		if se.GetSessionVars().StmtCtx.AffectedRows() != 1 {
			return errors.Errorf("fail to update task status, maybe the owner is not myself (%s), affected rows: %d",
				m.id, se.GetSessionVars().StmtCtx.AffectedRows())
		}
	}
	return nil
}

it'll return error when one of the tasks fail to update. However, we should log the error and continue to try the next task.

[YangKeao]

Job has similar issue, but it will not occur in normal cases, so it's not a big problem for the job manager. I should also fix it.