From d317dd2dde04b2bd913d5abe2f7e07ffe234d6b6 Mon Sep 17 00:00:00 2001
From: Agniva De Sarker <agnivade@yahoo.co.in>
Date: Fri, 22 May 2020 23:52:25 +0530
Subject: [PATCH] MM-25154: Fix data race in InvokeClusterLeaderChangedListener
 (#14610)

The order of booting up a server should be
Job scheduler -> Cluster.

And shutdown should be the opposite. This is needed because the job scheduler
initializes certain data structures that are later called by ClusterLeaderChanged
event handlers. And the event handlers run in a separate goroutine.

Therefore, if the cluster initialization happens before the job scheduler,
then a race happens between accessing the jobs variable and setting it.

To prevent this race, we fix the order of startup, and add comments in both
places to prevent things from regressing again.
---
 app/cluster.go             | 4 ++++
 app/server.go              | 6 ++++--
 app/server_app_adapters.go | 2 +-
 3 files changed, 9 insertions(+), 3 deletions(-)

diff --git a/app/cluster.go b/app/cluster.go
index 5ca16c9364..4813dd9441 100644
--- a/app/cluster.go
+++ b/app/cluster.go
@@ -23,6 +23,10 @@ func (s *Server) RemoveClusterLeaderChangedListener(id string) {
 
 func (s *Server) InvokeClusterLeaderChangedListeners() {
 	s.Log.Info("Cluster leader changed. Invoking ClusterLeaderChanged listeners.")
+	// This needs to be run in a separate goroutine otherwise a recursive lock happens
+	// because the listener function eventually ends up calling .IsLeader().
+	// Fixing this would require the changed event to pass the leader directly, but that
+	// requires a lot of work.
 	s.Go(func() {
 		s.clusterLeaderListeners.Range(func(_, listener interface{}) bool {
 			listener.(func())()
diff --git a/app/server.go b/app/server.go
index 0f20b06171..851fc86976 100644
--- a/app/server.go
+++ b/app/server.go
@@ -350,6 +350,9 @@ func NewServer(options ...Option) (*Server, error) {
 		mlog.Error("Error to reset the server status.", mlog.Err(err))
 	}
 
+	// Scheduler must be started before cluster.
+	s.initJobs()
+
 	if s.joinCluster && s.Cluster != nil {
 		s.FakeApp().registerAllClusterMessageHandlers()
 		s.Cluster.StartInterNodeCommunication()
@@ -374,8 +377,6 @@ func NewServer(options ...Option) (*Server, error) {
 		}
 	}
 
-	s.initJobs()
-
 	if s.runjobs {
 		s.Go(func() {
 			runSecurityJob(s)
@@ -490,6 +491,7 @@ func (s *Server) Shutdown() error {
 		s.Metrics.StopServer()
 	}
 
+	// This must be done after the cluster is stopped.
 	if s.Jobs != nil && s.runjobs {
 		s.Jobs.StopWorkers()
 		s.Jobs.StopSchedulers()
diff --git a/app/server_app_adapters.go b/app/server_app_adapters.go
index e88d93f1b7..23fd4ff5f6 100644
--- a/app/server_app_adapters.go
+++ b/app/server_app_adapters.go
@@ -114,7 +114,7 @@ func (s *Server) RunOldAppInitialization() error {
 
 	s.clusterLeaderListenerId = s.AddClusterLeaderChangedListener(func() {
 		mlog.Info("Cluster leader changed. Determining if job schedulers should be running:", mlog.Bool("isLeader", s.FakeApp().IsLeader()))
-		if s.Jobs != nil {
+		if s.Jobs != nil && s.Jobs.Schedulers != nil {
 			s.Jobs.Schedulers.HandleClusterLeaderChange(s.FakeApp().IsLeader())
 		}
 	})