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

@@ -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())()

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()

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())
 		}
 	})