Supress ActorSelectionMessage with DeadLetterSuppression, #28226 (#28341)

* for example the Cluster InitJoin message is marked with DeadLetterSuppression
  but was anyway logged because sent with actorSelection
* for other WrappedMessage than ActorSelectionMessage we shouldn't unwrap and publish
  the inner in SuppressedDeadLetter because that might loose some information
* therefore those are silenced in the DeadLetterListener instead
This commit is contained in:
Patrik Nordwall 2019-12-11 08:58:25 +01:00 committed by Christopher Batey
parent 87b94b65fd
commit 2a0e967be3
3 changed files with 65 additions and 23 deletions

View file

@ -50,6 +50,15 @@ class DeadLetterSupressionSpec extends AkkaSpec with ImplicitSender {
allListener.expectMsg(SuppressedDeadLetter(SuppressedMsg, testActor, system.deadLetters)) allListener.expectMsg(SuppressedDeadLetter(SuppressedMsg, testActor, system.deadLetters))
allListener.expectMsg(DeadLetter(NormalMsg, testActor, deadActor)) allListener.expectMsg(DeadLetter(NormalMsg, testActor, deadActor))
allListener.expectNoMessage() allListener.expectNoMessage()
// unwrap for ActorSelection
system.actorSelection(deadActor.path) ! SuppressedMsg
system.actorSelection(deadActor.path) ! NormalMsg
// the recipient ref isn't the same as deadActor here so only checking the message
deadListener.expectMsgType[DeadLetter].message should ===(NormalMsg)
suppressedListener.expectMsgType[SuppressedDeadLetter].message should ===(SuppressedMsg)
deadListener.expectNoMessage()
} }
s"must suppress message from default dead-letters logging (sent to dead: ${Logging.simpleName(system.deadLetters)})" in { s"must suppress message from default dead-letters logging (sent to dead: ${Logging.simpleName(system.deadLetters)})" in {
@ -76,5 +85,13 @@ class DeadLetterSupressionSpec extends AkkaSpec with ImplicitSender {
deadListener.expectNoMessage(Duration.Zero) deadListener.expectNoMessage(Duration.Zero)
suppressedListener.expectNoMessage(Duration.Zero) suppressedListener.expectNoMessage(Duration.Zero)
allListener.expectNoMessage(Duration.Zero) allListener.expectNoMessage(Duration.Zero)
// unwrap for ActorSelection
system.actorSelection(system.deadLetters.path) ! SuppressedMsg
system.actorSelection(system.deadLetters.path) ! NormalMsg
deadListener.expectMsg(DeadLetter(NormalMsg, testActor, system.deadLetters))
suppressedListener.expectMsg(SuppressedDeadLetter(SuppressedMsg, testActor, system.deadLetters))
deadListener.expectNoMessage()
} }
} }

View file

@ -608,14 +608,23 @@ private[akka] class EmptyLocalActorRef(
case Some(identify) => case Some(identify) =>
if (!sel.wildcardFanOut) sender ! ActorIdentity(identify.messageId, None) if (!sel.wildcardFanOut) sender ! ActorIdentity(identify.messageId, None)
case None => case None =>
eventStream.publish(DeadLetter(sel.msg, if (sender eq Actor.noSender) provider.deadLetters else sender, this)) sel.msg match {
case m: DeadLetterSuppression => publishSupressedDeadLetter(m, sender)
case _ =>
eventStream.publish(
DeadLetter(sel.msg, if (sender eq Actor.noSender) provider.deadLetters else sender, this))
}
} }
true true
case m: DeadLetterSuppression => case m: DeadLetterSuppression =>
eventStream.publish(SuppressedDeadLetter(m, if (sender eq Actor.noSender) provider.deadLetters else sender, this)) publishSupressedDeadLetter(m, sender)
true true
case _ => false case _ => false
} }
private def publishSupressedDeadLetter(msg: DeadLetterSuppression, sender: ActorRef): Unit = {
eventStream.publish(SuppressedDeadLetter(msg, if (sender eq Actor.noSender) provider.deadLetters else sender, this))
}
} }
/** /**

View file

@ -13,6 +13,7 @@ import akka.actor.ActorRef
import akka.actor.AllDeadLetters import akka.actor.AllDeadLetters
import akka.actor.DeadLetter import akka.actor.DeadLetter
import akka.actor.DeadLetterActorRef import akka.actor.DeadLetterActorRef
import akka.actor.DeadLetterSuppression
import akka.actor.Dropped import akka.actor.Dropped
import akka.actor.WrappedMessage import akka.actor.WrappedMessage
import akka.event.Logging.Info import akka.event.Logging.Info
@ -59,12 +60,15 @@ class DeadLetterListener extends Actor {
private def receiveWithAlwaysLogging: Receive = { private def receiveWithAlwaysLogging: Receive = {
case d: AllDeadLetters => case d: AllDeadLetters =>
if (!isWrappedSuppressed(d)) {
incrementCount() incrementCount()
logDeadLetter(d, doneMsg = "") logDeadLetter(d, doneMsg = "")
} }
}
private def receiveWithMaxCountLogging: Receive = { private def receiveWithMaxCountLogging: Receive = {
case d: AllDeadLetters => case d: AllDeadLetters =>
if (!isWrappedSuppressed(d)) {
incrementCount() incrementCount()
if (count == maxCount) { if (count == maxCount) {
logDeadLetter(d, ", no more dead letters will be logged") logDeadLetter(d, ", no more dead letters will be logged")
@ -73,9 +77,11 @@ class DeadLetterListener extends Actor {
logDeadLetter(d, "") logDeadLetter(d, "")
} }
} }
}
private def receiveWithSuspendLogging(suspendDuration: FiniteDuration): Receive = { private def receiveWithSuspendLogging(suspendDuration: FiniteDuration): Receive = {
case d: AllDeadLetters => case d: AllDeadLetters =>
if (!isWrappedSuppressed(d)) {
incrementCount() incrementCount()
if (count == maxCount) { if (count == maxCount) {
val doneMsg = s", no more dead letters will be logged in next [${suspendDuration.pretty}]" val doneMsg = s", no more dead letters will be logged in next [${suspendDuration.pretty}]"
@ -84,9 +90,11 @@ class DeadLetterListener extends Actor {
} else } else
logDeadLetter(d, "") logDeadLetter(d, "")
} }
}
private def receiveWhenSuspended(suspendDuration: FiniteDuration, suspendDeadline: Deadline): Receive = { private def receiveWhenSuspended(suspendDuration: FiniteDuration, suspendDeadline: Deadline): Receive = {
case d: AllDeadLetters => case d: AllDeadLetters =>
if (!isWrappedSuppressed(d)) {
incrementCount() incrementCount()
if (suspendDeadline.isOverdue()) { if (suspendDeadline.isOverdue()) {
val doneMsg = s", of which ${count - maxCount - 1} were not logged. The counter will be reset now" val doneMsg = s", of which ${count - maxCount - 1} were not logged. The counter will be reset now"
@ -95,6 +103,7 @@ class DeadLetterListener extends Actor {
context.become(receiveWithSuspendLogging(suspendDuration)) context.become(receiveWithSuspendLogging(suspendDuration))
} }
} }
}
private def logDeadLetter(d: AllDeadLetters, doneMsg: String): Unit = { private def logDeadLetter(d: AllDeadLetters, doneMsg: String): Unit = {
val origin = if (isReal(d.sender)) s" from ${d.sender}" else "" val origin = if (isReal(d.sender)) s" from ${d.sender}" else ""
@ -126,4 +135,11 @@ class DeadLetterListener extends Actor {
(snd ne ActorRef.noSender) && (snd ne context.system.deadLetters) && !snd.isInstanceOf[DeadLetterActorRef] (snd ne ActorRef.noSender) && (snd ne context.system.deadLetters) && !snd.isInstanceOf[DeadLetterActorRef]
} }
private def isWrappedSuppressed(d: AllDeadLetters): Boolean = {
d.message match {
case w: WrappedMessage if w.message.isInstanceOf[DeadLetterSuppression] => true
case _ => false
}
}
} }