Revert "perf: prioritize handling of Terminated after updating ddata" (#32766)

This reverts commit a9916f3.

Author: pvlugter

akka-cluster-sharding/src/main/scala/akka/cluster/sharding/ShardCoordinator.scala

@@ -1755,15 +1755,14 @@ private[akka] class DDataShardCoordinator(
       shardId: Option[ShardId],
       waitingForStateWrite: Boolean,
       waitingForRememberShard: Boolean,
-      afterUpdateCallback: E => Unit,
-      terminatingShardRegion: Option[Terminated]): Receive = {
+      afterUpdateCallback: E => Unit): Receive = {
 
     case UpdateSuccess(CoordinatorStateKey, Some(`evt`)) =>
       updateStateRetries = 0
       if (!waitingForRememberShard) {
         log.debug("{}: The coordinator state was successfully updated with {}", typeName, evt)
         if (shardId.isDefined) timers.cancel(RememberEntitiesTimeoutKey)
-        unbecomeAfterUpdate(evt, afterUpdateCallback, terminatingShardRegion)
+        unbecomeAfterUpdate(evt, afterUpdateCallback)
       } else {
         log.debug(
           "{}: The coordinator state was successfully updated with {}, waiting for remember shard update",
@@ -1775,8 +1774,7 @@ private[akka] class DDataShardCoordinator(
             shardId,
             waitingForStateWrite = false,
             waitingForRememberShard = true,
-            afterUpdateCallback = afterUpdateCallback,
-            terminatingShardRegion = terminatingShardRegion))
+            afterUpdateCallback = afterUpdateCallback))
       }
 
     case UpdateTimeout(CoordinatorStateKey, Some(`evt`)) =>
@@ -1823,25 +1821,6 @@ private[akka] class DDataShardCoordinator(
       if (!handleGetShardHome(shard))
         stashGetShardHomeRequest(sender(), g) // must wait for update that is in progress
 
-    case t @ Terminated(ref) if terminatingShardRegion.isEmpty && state.regions.get(ref).fold(0)(_.size) > 0 =>
-      // ensure that if this termination results in any state updates,
-      // those updates will be the next state update
-      // the effect is to bulk-deallocate shards associated with this region
-      // if there was a rebalance before this (graceful leaving) with pending
-      // RebalanceDone (which also deallocates shards), the handling for that
-      // accounts for processing Terminated before RebalanceDone
-      //
-      // Subsequent Terminateds while waiting for update will be handled like
-      // any other message (viz. stashed)
-      context.become(
-        waitingForUpdate(
-          evt,
-          shardId,
-          waitingForStateWrite = waitingForStateWrite,
-          waitingForRememberShard = waitingForRememberShard,
-          afterUpdateCallback = afterUpdateCallback,
-          terminatingShardRegion = Some(t)))
-
     case ShardCoordinator.Internal.Terminate =>
       log.debug("{}: The ShardCoordinator received termination message while waiting for update", typeName)
       terminating = true
@@ -1858,7 +1837,7 @@ private[akka] class DDataShardCoordinator(
         if (!waitingForStateWrite) {
           log.debug("{}: The ShardCoordinator saw remember shard start successfully written {}", typeName, evt)
           if (shardId.isDefined) timers.cancel(RememberEntitiesTimeoutKey)
-          unbecomeAfterUpdate(evt, afterUpdateCallback, terminatingShardRegion)
+          unbecomeAfterUpdate(evt, afterUpdateCallback)
         } else {
           log.debug(
             "{}: The ShardCoordinator saw remember shard start successfully written {}, waiting for state update",
@@ -1870,8 +1849,7 @@ private[akka] class DDataShardCoordinator(
               shardId,
               waitingForStateWrite = true,
               waitingForRememberShard = false,
-              afterUpdateCallback = afterUpdateCallback,
-              terminatingShardRegion = terminatingShardRegion))
+              afterUpdateCallback = afterUpdateCallback))
         }
       }
 
@@ -1907,18 +1885,8 @@ private[akka] class DDataShardCoordinator(
     case _ => stash()
   }
 
-  private def unbecomeAfterUpdate[E <: DomainEvent](
-      evt: E,
-      afterUpdateCallback: E => Unit,
-      unprocessedTermination: Option[Terminated]): Unit = {
+  private def unbecomeAfterUpdate[E <: DomainEvent](evt: E, afterUpdateCallback: E => Unit): Unit = {
     context.unbecome()
-
-    // want an unprocessed termination to be in the mailbox before
-    // anything else could be unstashed in the callback
-    unprocessedTermination.foreach { t =>
-      self.tell(t, ActorRef.noSender)
-    }
-
     afterUpdateCallback(evt)
     if (verboseDebug)
       log.debug("{}: New coordinator state after [{}]: [{}]", typeName, evt, state)
@@ -1991,8 +1959,7 @@ private[akka] class DDataShardCoordinator(
             shardId = Some(s.shard),
             waitingForStateWrite = true,
             waitingForRememberShard = true,
-            afterUpdateCallback = f,
-            terminatingShardRegion = None)
+            afterUpdateCallback = f)
 
         case _ =>
           // no update of shards, already known
@@ -2001,8 +1968,7 @@ private[akka] class DDataShardCoordinator(
             shardId = None,
             waitingForStateWrite = true,
             waitingForRememberShard = false,
-            afterUpdateCallback = f,
-            terminatingShardRegion = None)
+            afterUpdateCallback = f)
       }
     context.become(waitingReceive, discardOld = false)
   }