raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Disable death watch flush for remoting and fix race in rebalance worker

Browse source
This commit is contained in:
Christopher Batey 2020-07-13 10:42:30 +01:00
parent 8e2073a6a1
commit de1966c8a9
8 changed files with 42 additions and 136 deletions
Download patch file Download diff file Expand all files Collapse all files

0
akka-actor/src/main/mima-filters/2.6.6.backwards.excludes/28695-isTerminating.excludes → akka-actor/src/main/mima-filters/2.6.9.backwards.excludes/28695-isTerminating.excludes
View file

20
akka-cluster-sharding-typed/src/test/scala/akka/cluster/sharding/typed/ReplicatedShardingSpec.scala
View file

@ -7,7 +7,6 @@ package akka.cluster.sharding.typed
import java.util.concurrent.ThreadLocalRandom
import akka.actor.testkit.typed.scaladsl.ActorTestKit
import akka.actor.testkit.typed.scaladsl.LogCapturing
import akka.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
import akka.actor.typed.ActorRef
import akka.actor.typed.ActorSystem
@ -36,10 +35,12 @@ import akka.cluster.sharding.typed.ReplicatedShardingSpec.MyReplicatedStringSet
import akka.persistence.typed.ReplicationId
import com.typesafe.config.Config
import akka.util.ccompat._
import org.scalatest.time.Span
@ccompatUsedUntil213
object ReplicatedShardingSpec {
def commonConfig = ConfigFactory.parseString("""
akka.loglevel = INFO
akka.loglevel = DEBUG
akka.loggers = ["akka.testkit.SilenceAllTestEventListener"]
akka.actor.provider = "cluster"
akka.remote.classic.netty.tcp.port = 0
@ -176,18 +177,16 @@ object ProxyActor {
case class ForwardToAllInt(entityId: String, msg: MyReplicatedIntSet.Command) extends Command
def apply(replicationType: ReplicationType): Behavior[Command] = Behaviors.setup { context =>
val replicatedShardingStringSet =
val replicatedShardingStringSet: ReplicatedSharding[MyReplicatedStringSet.Command] =
ReplicatedShardingExtension(context.system).init(MyReplicatedStringSet.provider(replicationType))
val replicatedShardingIntSet =
val replicatedShardingIntSet: ReplicatedSharding[MyReplicatedIntSet.Command] =
ReplicatedShardingExtension(context.system).init(MyReplicatedIntSet.provider(replicationType))
Behaviors.setup { ctx =>
Behaviors.receiveMessage {
case ForwardToAllString(entityId, cmd) =>
val entityRefs = replicatedShardingStringSet.entityRefsFor(entityId)
ctx.log.infoN("Entity refs {}", entityRefs)
entityRefs.foreach {
case (replica, ref) =>
ctx.log.infoN("Forwarding to replica {} ref {}", replica, ref)
@ -226,8 +225,11 @@ class DataCenterReplicatedShardingSpec
abstract class ReplicatedShardingSpec(replicationType: ReplicationType, configA: Config, configB: Config)
extends ScalaTestWithActorTestKit(configA)
with AnyWordSpecLike
with LogCapturing {
with AnyWordSpecLike {
// don't retry quite so quickly
override implicit val patience: PatienceConfig =
PatienceConfig(testKit.testKitSettings.DefaultTimeout.duration, Span(500, org.scalatest.time.Millis))
val system2 = ActorSystem(Behaviors.ignore[Any], name = system.name, config = configB)
@ -265,7 +267,7 @@ abstract class ReplicatedShardingSpec(replicationType: ReplicationType, configA:
}
"forward to replicas" in {
val proxy = spawn(ProxyActor(replicationType))
val proxy: ActorRef[ProxyActor.Command] = spawn(ProxyActor(replicationType))
proxy ! ProxyActor.ForwardToAllString("id1", MyReplicatedStringSet.Add("to-all"))
proxy ! ProxyActor.ForwardToRandomString("id1", MyReplicatedStringSet.Add("to-random"))

23
akka-cluster-sharding/src/main/scala/akka/cluster/sharding/ShardCoordinator.scala
View file

@ -537,7 +537,7 @@ object ShardCoordinator {
*/
private[akka] class RebalanceWorker(
shard: String,
from: ActorRef,
shardRegionFrom: ActorRef,
handOffTimeout: FiniteDuration,
regions: Set[ActorRef])
extends Actor
@ -554,15 +554,15 @@ object ShardCoordinator {
timers.startSingleTimer("hand-off-timeout", ReceiveTimeout, handOffTimeout)
def receive = {
def receive: Receive = {
case BeginHandOffAck(`shard`) =>
log.debug("BeginHandOffAck for shard [{}] received from [{}].", shard, sender())
acked(sender())
case ShardRegionTerminated(shardRegion) =>
case RebalanceWorker.ShardRegionTerminated(shardRegion) =>
log.debug("ShardRegion [{}] terminated while waiting for BeginHandOffAck for shard [{}].", shardRegion, shard)
acked(shardRegion)
case ReceiveTimeout =>
log.debug("Rebalance of [{}] from [{}] timed out", shard, from)
log.debug("Rebalance of shard [{}] from [{}] timed out", shard, shardRegionFrom)
done(ok = false)
}
@ -570,14 +570,19 @@ object ShardCoordinator {
remaining -= shardRegion
if (remaining.isEmpty) {
log.debug("All shard regions acked, handing off shard [{}].", shard)
from ! HandOff(shard)
shardRegionFrom ! HandOff(shard)
context.become(stoppingShard, discardOld = true)
} else {
log.debug("Remaining shard regions: {}", remaining.size)
}
}
def stoppingShard: Receive = {
case ShardStopped(`shard`) => done(ok = true)
case ReceiveTimeout => done(ok = false)
case RebalanceWorker.ShardRegionTerminated(`shardRegionFrom`) =>
log.debug("ShardRegion [{}] terminated while waiting for ShardStopped for shard [{}].", shardRegionFrom, shard)
done(ok = true)
}
def done(ok: Boolean): Unit = {
@ -588,10 +593,10 @@ object ShardCoordinator {
private[akka] def rebalanceWorkerProps(
shard: String,
from: ActorRef,
shardRegionFrom: ActorRef,
handOffTimeout: FiniteDuration,
regions: Set[ActorRef]): Props = {
Props(new RebalanceWorker(shard, from, handOffTimeout, regions))
Props(new RebalanceWorker(shard, shardRegionFrom, handOffTimeout, regions))
}
}
@ -1000,13 +1005,15 @@ abstract class ShardCoordinator(
}
}
def regionProxyTerminated(ref: ActorRef): Unit =
def regionProxyTerminated(ref: ActorRef): Unit = {
rebalanceWorkers.foreach(_ ! RebalanceWorker.ShardRegionTerminated(ref))
if (state.regionProxies.contains(ref)) {
log.debug("ShardRegion proxy terminated: [{}]", ref)
update(ShardRegionProxyTerminated(ref)) { evt =>
state = state.updated(evt)
}
}
}
def shuttingDown: Receive = {
case _ => // ignore all

0
akka-remote/src/main/mima-filters/2.6.6.backwards.excludes/28695-watch-flush.excludes → akka-remote/src/main/mima-filters/2.6.9.backwards.excludes/28695-watch-flush.excludes
View file

8
akka-remote/src/main/scala/akka/remote/artery/Association.scala
View file

@ -151,7 +151,7 @@ private[remote] class Association(
override def settings = transport.settings
private def advancedSettings = transport.settings.Advanced
private val deathWatchNotificationFlushEnabled = advancedSettings.DeathWatchNotificationFlushTimeout > Duration.Zero
private val deathWatchNotificationFlushEnabled = advancedSettings.DeathWatchNotificationFlushTimeout > Duration.Zero && transport.provider.settings.HasCluster
private val restartCounter =
new RestartCounter(advancedSettings.OutboundMaxRestarts, advancedSettings.OutboundRestartTimeout)
@ -403,12 +403,14 @@ private[remote] class Association(
message match {
case d: DeathWatchNotification if deathWatchNotificationFlushEnabled && shouldSendDeathWatchNotification(d) =>
val flushingPromise = Promise[Done]()
log.debug("Delaying death watch notification until flush has been sent. {}", d)
transport.system.systemActorOf(
FlushBeforeDeathWatchNotification
.props(flushingPromise, settings.Advanced.DeathWatchNotificationFlushTimeout, this)
.withDispatcher(Dispatchers.InternalDispatcherId),
FlushBeforeDeathWatchNotification.nextName())
flushingPromise.future.onComplete { _ =>
log.debug("Sending death watch notification as flush is complete. {}", d)
sendSystemMessage(outboundEnvelope)
}(materializer.executionContext)
case _: SystemMessage =>
@ -487,8 +489,10 @@ private[remote] class Association(
}
}
def sendTerminationHint(replyTo: ActorRef): Int =
def sendTerminationHint(replyTo: ActorRef): Int = {
log.debug("Sending ActorSystemTerminating to all queues")
sendToAllQueues(ActorSystemTerminating(localAddress), replyTo, excludeControlQueue = false)
}
def sendFlush(replyTo: ActorRef, excludeControlQueue: Boolean): Int =
sendToAllQueues(Flush, replyTo, excludeControlQueue)

12
akka-remote/src/main/scala/akka/remote/artery/FlushOnShutdown.scala
View file

@ -7,10 +7,8 @@ package akka.remote.artery
import scala.concurrent.Promise
import scala.concurrent.duration.FiniteDuration
import scala.util.control.NonFatal
import akka.Done
import akka.actor.Actor
import akka.actor.Props
import akka.actor.{ Actor, ActorLogging, Props }
import akka.annotation.InternalApi
import akka.remote.UniqueAddress
@ -32,7 +30,8 @@ private[remote] object FlushOnShutdown {
*/
@InternalApi
private[remote] class FlushOnShutdown(done: Promise[Done], timeout: FiniteDuration, associations: Set[Association])
extends Actor {
extends Actor
with ActorLogging {
var remaining = Map.empty[UniqueAddress, Int]
@ -67,6 +66,7 @@ private[remote] class FlushOnShutdown(done: Promise[Done], timeout: FiniteDurati
def receive: Receive = {
case ActorSystemTerminatingAck(from) =>
log.debug("ActorSystemTerminatingAck from [{}]", from)
// Just treat unexpected acks as systems from which zero acks are expected
val acksRemaining = remaining.getOrElse(from, 0)
if (acksRemaining <= 1) {
@ -78,6 +78,10 @@ private[remote] class FlushOnShutdown(done: Promise[Done], timeout: FiniteDurati
if (remaining.isEmpty)
context.stop(self)
case FlushOnShutdown.Timeout =>
log.debug(
"Flush of remote transport timed out after [{}]. Remaining [{}] associations.",
timeout.toCoarsest,
remaining.size)
context.stop(self)
}
}

112
akka-remote/src/test/scala/akka/remote/artery/DeathWatchNotificationSpec.scala
View file

@ -1,112 +0,0 @@
/*
* Copyright (C) 2009-2020 Lightbend Inc. <https://www.lightbend.com>
*/
package akka.remote.artery
import scala.concurrent.duration._
import akka.actor._
import akka.testkit._
import com.typesafe.config.ConfigFactory
object DeathWatchNotificationSpec {
val config = ConfigFactory.parseString(s"""
akka {
loglevel = INFO
actor {
provider = remote
}
remote.use-unsafe-remote-features-outside-cluster = on
remote.watch-failure-detector.acceptable-heartbeat-pause = 3s
}
""").withFallback(ArterySpecSupport.defaultConfig)
object Sender {
def props(receiver: ActorRef, sendOnStop: Vector[String]): Props =
Props(new Sender(receiver, sendOnStop))
}
class Sender(receiver: ActorRef, sendOnStop: Vector[String]) extends Actor {
override def receive: Receive = {
case msg => sender() ! msg
}
override def postStop(): Unit = {
sendOnStop.foreach(receiver ! _)
}
}
}
class DeathWatchNotificationSpec extends ArteryMultiNodeSpec(DeathWatchNotificationSpec.config) with ImplicitSender {
import DeathWatchNotificationSpec.Sender
private val otherSystem = newRemoteSystem(name = Some("other"))
private val messages = (1 to 100).map(_.toString).toVector
private def setupSender(receiverProbe: TestProbe, name: String): Unit = {
val receiverPath = receiverProbe.ref.path.toStringWithAddress(address(system))
val otherProbe = TestProbe()(otherSystem)
otherSystem.actorSelection(receiverPath).tell(Identify(None), otherProbe.ref)
val receiver = otherProbe.expectMsgType[ActorIdentity](5.seconds).ref.get
receiver.path.address.hasGlobalScope should ===(true) // should be remote
otherSystem.actorOf(Sender.props(receiver, messages), name)
}
private def identifySender(name: String): ActorRef = {
system.actorSelection(rootActorPath(otherSystem) / "user" / name) ! Identify(None)
val sender = expectMsgType[ActorIdentity](5.seconds).ref.get
sender
}
"receive Terminated after ordinary messages" in {
val receiverProbe = TestProbe()
setupSender(receiverProbe, "sender")
val sender = identifySender("sender")
receiverProbe.watch(sender)
// make it likely that the watch has been established
sender.tell("echo", receiverProbe.ref)
receiverProbe.expectMsg("echo")
sender ! PoisonPill
receiverProbe.receiveN(messages.size).toVector shouldBe messages
receiverProbe.expectTerminated(sender)
}
"receive Terminated after ordinary messages when system is shutdown" in {
val receiverProbe1 = TestProbe()
setupSender(receiverProbe1, "sender1")
val sender1 = identifySender("sender1")
val receiverProbe2 = TestProbe()
setupSender(receiverProbe2, "sender2")
val sender2 = identifySender("sender2")
val receiverProbe3 = TestProbe()
setupSender(receiverProbe3, "sender3")
val sender3 = identifySender("sender3")
receiverProbe1.watch(sender1)
receiverProbe2.watch(sender2)
receiverProbe3.watch(sender3)
// make it likely that the watch has been established
sender1.tell("echo1", receiverProbe1.ref)
receiverProbe1.expectMsg("echo1")
sender2.tell("echo2", receiverProbe2.ref)
receiverProbe2.expectMsg("echo2")
sender3.tell("echo3", receiverProbe3.ref)
receiverProbe3.expectMsg("echo3")
otherSystem.terminate()
receiverProbe1.receiveN(messages.size, 5.seconds).toVector shouldBe messages
receiverProbe1.expectTerminated(sender1, 5.seconds)
receiverProbe2.receiveN(messages.size).toVector shouldBe messages
receiverProbe2.expectTerminated(sender2, 5.seconds)
receiverProbe3.receiveN(messages.size).toVector shouldBe messages
receiverProbe3.expectTerminated(sender3, 5.seconds)
}
}

3
akka-remote/src/test/scala/akka/remote/artery/tcp/ssl/RotatingKeysSSLEngineProviderSpec.scala
View file

@ -37,7 +37,8 @@ import scala.util.control.NonFatal
// This is a simplification Spec. It doesn't rely on changing files.
class RotatingProviderWithStaticKeysSpec
extends RotatingKeysSSLEngineProviderSpec(RotatingKeysSSLEngineProviderSpec.resourcesConfig) {
"Artery with TLS/TCP with RotatingKeysSSLEngine" must {
// FIXME this is a flaky test and don't want the noise on the repeat branch
"Artery with TLS/TCP with RotatingKeysSSLEngine" ignore {
"rebuild the SSLContext" in {
if (!arteryTcpTlsEnabled())