[SPARK-34273][CORE] Do not reregister BlockManager when SparkContext is stopped

### What changes were proposed in this pull request?

This PR aims to prevent `HeartbeatReceiver` asks `Executor` to re-register blocker manager when the SparkContext is already stopped.

### Why are the changes needed?

Currently, `HeartbeatReceiver` blindly asks re-registration for the new heartbeat message.
However, when SparkContext is stopped, we don't need to re-register new block manager.
Re-registration causes unnecessary executors' logs and and a delay on job termination.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Pass the CIs with the newly added test case.

Closes #31373 from dongjoon-hyun/SPARK-34273.

Authored-by: Dongjoon Hyun <dhyun@apple.com>
Signed-off-by: Dongjoon Hyun <dhyun@apple.com>

Author: dongjoon-hyun

core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala

@@ -128,14 +128,16 @@ private[spark] class HeartbeatReceiver(sc: SparkContext, clock: Clock)
 
     // Messages received from executors
     case heartbeat @ Heartbeat(executorId, accumUpdates, blockManagerId, executorUpdates) =>
+      var reregisterBlockManager = !sc.isStopped
       if (scheduler != null) {
         if (executorLastSeen.contains(executorId)) {
           executorLastSeen(executorId) = clock.getTimeMillis()
           eventLoopThread.submit(new Runnable {
             override def run(): Unit = Utils.tryLogNonFatalError {
               val unknownExecutor = !scheduler.executorHeartbeatReceived(
                 executorId, accumUpdates, blockManagerId, executorUpdates)
-              val response = HeartbeatResponse(reregisterBlockManager = unknownExecutor)
+              reregisterBlockManager &= unknownExecutor
+              val response = HeartbeatResponse(reregisterBlockManager)
               context.reply(response)
             }
           })
@@ -145,14 +147,14 @@ private[spark] class HeartbeatReceiver(sc: SparkContext, clock: Clock)
           // not log warning here. Otherwise there may be a lot of noise especially if
           // we explicitly remove executors (SPARK-4134).
           logDebug(s"Received heartbeat from unknown executor $executorId")
-          context.reply(HeartbeatResponse(reregisterBlockManager = true))
+          context.reply(HeartbeatResponse(reregisterBlockManager))
         }
       } else {
         // Because Executor will sleep several seconds before sending the first "Heartbeat", this
         // case rarely happens. However, if it really happens, log it and ask the executor to
         // register itself again.
         logWarning(s"Dropping $heartbeat because TaskScheduler is not ready yet")
-        context.reply(HeartbeatResponse(reregisterBlockManager = true))
+        context.reply(HeartbeatResponse(reregisterBlockManager))
       }
   }
 

core/src/test/scala/org/apache/spark/HeartbeatReceiverSuite.scala

@@ -219,6 +219,24 @@ class HeartbeatReceiverSuite
     fakeSchedulerBackend.stop()
   }
 
+  test("SPARK-34273: Do not reregister BlockManager when SparkContext is stopped") {
+    val blockManagerId = BlockManagerId(executorId1, "localhost", 12345)
+
+    heartbeatReceiverRef.askSync[Boolean](TaskSchedulerIsSet)
+    val response = heartbeatReceiverRef.askSync[HeartbeatResponse](
+      Heartbeat(executorId1, Array.empty, blockManagerId, mutable.Map.empty))
+    assert(response.reregisterBlockManager)
+
+    try {
+      sc.stopped.set(true)
+      val response = heartbeatReceiverRef.askSync[HeartbeatResponse](
+        Heartbeat(executorId1, Array.empty, blockManagerId, mutable.Map.empty))
+      assert(!response.reregisterBlockManager)
+    } finally {
+      sc.stopped.set(false)
+    }
+  }
+
   /** Manually send a heartbeat and return the response. */
   private def triggerHeartbeat(
       executorId: String,