DistributedPubSub keeps sending to exited nodes
97 views
Skip to first unread message
Christian Hoffmeister
Jun 23, 2016, 4:27:45 PM6/23/16
to Akka User List
Hi,
I have a sample application where two nodes join an Akka cluster. On both nodes there is an actor running registered to DistributedPubSubMediator and both are sending a Ping once per second (that hits either itself or the other node).
Then I gracefully shut down Node B (leave cluster, wait for MemberRemoved event, wait some seconds, terminate actor system, wait for termination, exit the jvm).
But even though both nodes see Node B leaving through gossip convergence at 22:16:58, Node A keeps the left Node B in the PubSub (you see this with the RECEIVED PONG(nr, ping sender, pong responder)).
Why does PubSub not remove actors from exited nodes, but only actors from removed nodes (remove of Node B on Node A happens 4 seconds after exiting of Node B). In my case I only do not loose messages, because Node B is waiting another 10 seconds after it sees its one remove (which again, is 4 seconds earlier than Node A sees Node B removed).
Thanks in advance.
Christian
[info] Thu Jun 23 22:16:23 CEST 2016 RECEIVE Pong(0,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:23 CEST 2016 CLUSTER MemberUp(Member(address = akka.tcp://airf...@127.0.0.1:2551, status = Up))[info] Thu Jun 23 22:16:23 CEST 2016 CLUSTER LeaderChanged(Some(akka.tcp://airf...@127.0.0.1:2551))[info] Thu Jun 23 22:16:23 CEST 2016 CLUSTER RoleLeaderChanged(api,Some(akka.tcp://airf...@127.0.0.1:2551))[info] Thu Jun 23 22:16:23 CEST 2016 CLUSTER SeenChanged(true,Set(akka.tcp://airf...@127.0.0.1:2551))[info] Thu Jun 23 22:16:24 CEST 2016 RECEIVE Pong(1,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:25 CEST 2016 RECEIVE Pong(2,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:26 CEST 2016 RECEIVE Pong(3,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:28 CEST 2016 RECEIVE Pong(4,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:29 CEST 2016 RECEIVE Pong(5,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:30 CEST 2016 RECEIVE Pong(6,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:31 CEST 2016 RECEIVE Pong(7,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:32 CEST 2016 RECEIVE Pong(8,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:33 CEST 2016 RECEIVE Pong(9,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:34 CEST 2016 RECEIVE Pong(10,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:35 CEST 2016 RECEIVE Pong(11,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:36 CEST 2016 RECEIVE Pong(12,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:37 CEST 2016 RECEIVE Pong(13,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:38 CEST 2016 RECEIVE Pong(14,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:39 CEST 2016 RECEIVE Pong(15,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:40 CEST 2016 RECEIVE Pong(16,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:41 CEST 2016 RECEIVE Pong(17,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:42 CEST 2016 RECEIVE Pong(18,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:43 CEST 2016 RECEIVE Pong(19,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:44 CEST 2016 RECEIVE Pong(20,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:45 CEST 2016 RECEIVE Pong(21,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:46 CEST 2016 RECEIVE Pong(22,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:47 CEST 2016 RECEIVE Pong(23,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:48 CEST 2016 RECEIVE Pong(24,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER MemberJoined(Member(address = akka.tcp://airf...@127.0.0.1:2552, status = Joining))[info] Thu Jun 23 22:16:49 CEST 2016 RECEIVE Pong(25,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER SeenChanged(true,Set(akka.tcp://airf...@127.0.0.1:2551, akka.tcp://airf...@127.0.0.1:2552))[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER ReachabilityChanged()[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER MemberUp(Member(address = akka.tcp://airf...@127.0.0.1:2552, status = Up))[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER SeenChanged(false,Set(akka.tcp://airf...@127.0.0.1:2551))[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER SeenChanged(true,Set(akka.tcp://airf...@127.0.0.1:2551, akka.tcp://airf...@127.0.0.1:2552))[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER ReachabilityChanged()[info] Thu Jun 23 22:16:50 CEST 2016 RECEIVE Pong(26,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2552)[info] Thu Jun 23 22:16:51 CEST 2016 RECEIVE Pong(27,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:52 CEST 2016 RECEIVE Pong(28,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:53 CEST 2016 RECEIVE Pong(29,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2552)[info] Thu Jun 23 22:16:54 CEST 2016 RECEIVE Pong(30,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2552)[info] Thu Jun 23 22:16:55 CEST 2016 RECEIVE Pong(31,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:56 CEST 2016 RECEIVE Pong(32,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:57 CEST 2016 RECEIVE Pong(33,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:58 CEST 2016 RECEIVE Pong(34,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER MemberLeft(Member(address = akka.tcp://airf...@127.0.0.1:2552, status = Leaving))[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER ReachabilityChanged()[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER MemberExited(Member(address = akka.tcp://airf...@127.0.0.1:2552, status = Exiting))[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER SeenChanged(true,Set(akka.tcp://airf...@127.0.0.1:2551))[info] Thu Jun 23 22:16:59 CEST 2016 RECEIVE Pong(35,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2552)[info] Thu Jun 23 22:17:00 CEST 2016 RECEIVE Pong(36,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:17:01 CEST 2016 RECEIVE Pong(37,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2552)[info] Thu Jun 23 22:17:02 CEST 2016 RECEIVE Pong(38,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2552)[info] Thu Jun 23 22:17:02 CEST 2016 CLUSTER UnreachableMember(Member(address = akka.tcp://airf...@127.0.0.1:2552, status = Exiting))[info] Thu Jun 23 22:17:02 CEST 2016 CLUSTER ReachabilityChanged(akka.tcp://airf...@127.0.0.1:2551 -> akka.tcp://airf...@127.0.0.1:2552: Unreachable [Unreachable] (1))[info] Thu Jun 23 22:17:02 CEST 2016 CLUSTER MemberRemoved(Member(address = akka.tcp://airf...@127.0.0.1:2552, status = Removed),Exiting)[info] Thu Jun 23 22:17:02 CEST 2016 CLUSTER ReachabilityChanged()[info] Thu Jun 23 22:17:03 CEST 2016 RECEIVE Pong(39,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:17:04 CEST 2016 RECEIVE Pong(40,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:17:05 CEST 2016 RECEIVE Pong(41,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:17:06 CEST 2016 RECEIVE Pong(42,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:17:07 CEST 2016 RECEIVE Pong(43,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:17:08 CEST 2016 RECEIVE Pong(44,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:17:09 CEST 2016 RECEIVE Pong(45,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:17:10 CEST 2016 RECEIVE Pong(46,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:17:10 CEST 2016 CLUSTER MemberLeft(Member(address = akka.tcp://airf...@127.0.0.1:2551, status = Leaving))[info] Thu Jun 23 22:17:11 CEST 2016 RECEIVE Pong(47,akka.tcp://airf...@127.0.0.1:2551,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:17:11 CEST 2016 CLUSTER MemberExited(Member(address = akka.tcp://airf...@127.0.0.1:2551, status = Exiting))[info] Thu Jun 23 22:17:11 CEST 2016 CLUSTER ClusterShuttingDown[info] Thu Jun 23 22:17:11 CEST 2016 CLUSTER MemberRemoved(Member(address = akka.tcp://airf...@127.0.0.1:2551, status = Removed),Exiting)[info] Thu Jun 23 22:17:11 CEST 2016 CLUSTER LeaderChanged(None)[info] Thu Jun 23 22:17:11 CEST 2016 CLUSTER RoleLeaderChanged(api,None)[info] Thu Jun 23 22:17:11 CEST 2016 CLUSTER SeenChanged(true,Set())[info] Thu Jun 23 22:17:11 CEST 2016 CLUSTER ReachabilityChanged()Output Node B
[info] Thu Jun 23 22:16:48 CEST 2016 RECEIVE Pong(0,akka.tcp://airf...@127.0.0.1:2552,akka.tcp://airf...@127.0.0.1:2552)[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER MemberUp(Member(address = akka.tcp://airf...@127.0.0.1:2551, status = Up))[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER MemberJoined(Member(address = akka.tcp://airf...@127.0.0.1:2552, status = Joining))[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER LeaderChanged(Some(akka.tcp://airf...@127.0.0.1:2551))[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER RoleLeaderChanged(api,Some(akka.tcp://airf...@127.0.0.1:2551))[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER SeenChanged(true,Set(akka.tcp://airf...@127.0.0.1:2551, akka.tcp://airf...@127.0.0.1:2552))[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER ReachabilityChanged()[info] Thu Jun 23 22:16:49 CEST 2016 RECEIVE Pong(1,akka.tcp://airf...@127.0.0.1:2552,akka.tcp://airf...@127.0.0.1:2552)[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER MemberUp(Member(address = akka.tcp://airf...@127.0.0.1:2552, status = Up))[info] Thu Jun 23 22:16:49 CEST 2016 CLUSTER ReachabilityChanged()[info] Thu Jun 23 22:16:50 CEST 2016 RECEIVE Pong(2,akka.tcp://airf...@127.0.0.1:2552,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:51 CEST 2016 RECEIVE Pong(3,akka.tcp://airf...@127.0.0.1:2552,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:52 CEST 2016 RECEIVE Pong(4,akka.tcp://airf...@127.0.0.1:2552,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:53 CEST 2016 RECEIVE Pong(5,akka.tcp://airf...@127.0.0.1:2552,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:54 CEST 2016 RECEIVE Pong(6,akka.tcp://airf...@127.0.0.1:2552,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:55 CEST 2016 RECEIVE Pong(7,akka.tcp://airf...@127.0.0.1:2552,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:56 CEST 2016 RECEIVE Pong(8,akka.tcp://airf...@127.0.0.1:2552,akka.tcp://airf...@127.0.0.1:2551)[info] Thu Jun 23 22:16:57 CEST 2016 RECEIVE Pong(9,akka.tcp://airf...@127.0.0.1:2552,akka.tcp://airf...@127.0.0.1:2552)[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER MemberLeft(Member(address = akka.tcp://airf...@127.0.0.1:2552, status = Leaving))[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER SeenChanged(false,Set(akka.tcp://airf...@127.0.0.1:2552))[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER SeenChanged(true,Set(akka.tcp://airf...@127.0.0.1:2552, akka.tcp://airf...@127.0.0.1:2551))[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER ReachabilityChanged()[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER MemberExited(Member(address = akka.tcp://airf...@127.0.0.1:2552, status = Exiting))[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER ReachabilityChanged()[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER ClusterShuttingDown[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER MemberRemoved(Member(address = akka.tcp://airf...@127.0.0.1:2551, status = Removed),Up)[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER MemberRemoved(Member(address = akka.tcp://airf...@127.0.0.1:2552, status = Removed),Exiting)[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER LeaderChanged(None)[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER RoleLeaderChanged(api,None)[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER SeenChanged(true,Set())[info] Thu Jun 23 22:16:58 CEST 2016 CLUSTER ReachabilityChanged()Christian Hoffmeister
Jun 23, 2016, 4:48:42 PM6/23/16
to Akka User List
Ok, my assumption is right (see https://github.com/akka/akka/blob/master/akka-cluster-tools/src/main/scala/akka/cluster/pubsub/DistributedPubSubMediator.scala#L669). Is there a way to work around this? I don't want to wait some random time interval after leaving before actually shutting down. If not just the removing but already the exiting would remove the node, then one would have a more robust way (without random waits) to gracefully leave a cluster/pubsub.
Also another question arises here: Wouldn't it be good, to temporarily remove nodes from pubsub that are unreachable and readd them when they are reachable again?
Am Donnerstag, 23. Juni 2016 22:27:45 UTC+2 schrieb Christian Hoffmeister:
Hi,I have a sample application where two nodes join an Akka cluster. On both nodes there is an actor running registered to DistributedPubSubMediator and both are sending a Ping once per second (that hits either itself or the other node).Then I gracefully shut down Node B (leave cluster, wait for MemberRemoved event, wait some seconds, terminate actor system, wait for termination, exit the jvm).But even though both nodes see Node B leaving through gossip convergence at 22:16:58, Node A keeps the left Node B in the PubSub (you see this with the RECEIVED PONG(nr, ping sender, pong responder)).Why does PubSub not remove actors from exited nodes, but only actors from removed nodes (remove of Node B on Node A happens 4 seconds after exiting of Node B). In my case I only do not loose messages, because Node B is waiting another 10 seconds after it sees its one remove (which again, is 4 seconds earlier than Node A sees Node B removed).Thanks in advance.ChristianOutput Node A
Output Node B
Christian Hoffmeister
Jun 24, 2016, 4:11:37 AM6/24/16
to Akka User List
Have chatted with @ktoso about this and seems, that my thoughts are not totally wrong. Will create a ticket on GitHub for that.
Am Donnerstag, 23. Juni 2016 22:27:45 UTC+2 schrieb Christian Hoffmeister:
Hi,I have a sample application where two nodes join an Akka cluster. On both nodes there is an actor running registered to DistributedPubSubMediator and both are sending a Ping once per second (that hits either itself or the other node).Then I gracefully shut down Node B (leave cluster, wait for MemberRemoved event, wait some seconds, terminate actor system, wait for termination, exit the jvm).But even though both nodes see Node B leaving through gossip convergence at 22:16:58, Node A keeps the left Node B in the PubSub (you see this with the RECEIVED PONG(nr, ping sender, pong responder)).Why does PubSub not remove actors from exited nodes, but only actors from removed nodes (remove of Node B on Node A happens 4 seconds after exiting of Node B). In my case I only do not loose messages, because Node B is waiting another 10 seconds after it sees its one remove (which again, is 4 seconds earlier than Node A sees Node B removed).Thanks in advance.ChristianOutput Node A
Output Node B
Reply all
Reply to author
Forward
0 new messages