!clu #2307 Allow transition from unreachable to reachable

* Replace unreachable Set with Reachability table
* Unreachable members stay in member Set
* Downing a live member was moved it to the unreachable Set,
  and then removed from there by the leader. That will not
  work when flipping back to reachable, so a Down member must
  be detected as unreachable before beeing removed. Similar
  to Exiting. Member shuts down itself if it sees itself as
  Down.
* Flip back to reachable when failure detector monitors it as
  available again
* ReachableMember event
* Can't ignore gossip from aggregated unreachable (see SurviveNetworkInstabilitySpec)
* Make use of ReachableMember event in cluster router
* End heartbeat when acknowledged, EndHeartbeatAck
* Remove nr-of-end-heartbeats from conf
* Full reachability info in JMX cluster status
* Don't use interval after unreachable for AccrualFailureDetector history
* Add QuarantinedEvent to remoting, used for Reachability.Terminated
* Prune reachability table when all reachable
* Update documentation
* Performance testing and optimizations

akka-cluster/src/main/scala/akka/cluster/protobuf/ClusterMessageSerializer.scala

@@ -44,6 +44,7 @@ class ClusterMessageSerializer(val system: ExtendedActorSystem) extends Serializ
     classOf[InternalClusterAction.InitJoinNack] -> (bytes ⇒ InternalClusterAction.InitJoinNack(addressFromBinary(bytes))),
     classOf[ClusterHeartbeatReceiver.Heartbeat] -> (bytes ⇒ ClusterHeartbeatReceiver.Heartbeat(addressFromBinary(bytes))),
     classOf[ClusterHeartbeatReceiver.EndHeartbeat] -> (bytes ⇒ ClusterHeartbeatReceiver.EndHeartbeat(addressFromBinary(bytes))),
+    classOf[ClusterHeartbeatReceiver.EndHeartbeatAck] -> (bytes ⇒ ClusterHeartbeatReceiver.EndHeartbeatAck(addressFromBinary(bytes))),
     classOf[ClusterHeartbeatSender.HeartbeatRequest] -> (bytes ⇒ ClusterHeartbeatSender.HeartbeatRequest(addressFromBinary(bytes))),
     classOf[GossipStatus] -> gossipStatusFromBinary,
     classOf[GossipEnvelope] -> gossipEnvelopeFromBinary,
@@ -67,6 +68,7 @@ class ClusterMessageSerializer(val system: ExtendedActorSystem) extends Serializ
       case InternalClusterAction.InitJoinAck(address) ⇒ addressToProto(address).toByteArray
       case InternalClusterAction.InitJoinNack(address) ⇒ addressToProto(address).toByteArray
       case ClusterHeartbeatReceiver.EndHeartbeat(from) ⇒ addressToProto(from).toByteArray
+      case ClusterHeartbeatReceiver.EndHeartbeatAck(from) ⇒ addressToProto(from).toByteArray
       case ClusterHeartbeatSender.HeartbeatRequest(from) ⇒ addressToProto(from).toByteArray
       case _ ⇒ throw new IllegalArgumentException(s"Can't serialize object of type ${obj.getClass}")
     }
@@ -132,6 +134,13 @@ class ClusterMessageSerializer(val system: ExtendedActorSystem) extends Serializ
 
   private val memberStatusFromInt = memberStatusToInt.map { case (a, b) ⇒ (b, a) }
 
+  private val reachabilityStatusToInt = scala.collection.immutable.HashMap[Reachability.ReachabilityStatus, Int](
+    Reachability.Reachable -> msg.ReachabilityStatus.Reachable_VALUE,
+    Reachability.Unreachable -> msg.ReachabilityStatus.Unreachable_VALUE,
+    Reachability.Terminated -> msg.ReachabilityStatus.Terminated_VALUE)
+
+  private val reachabilityStatusFromInt = reachabilityStatusToInt.map { case (a, b) ⇒ (b, a) }
+
   private def mapWithErrorMessage[T](map: Map[T, Int], value: T, unknown: String): Int = map.get(value) match {
     case Some(x) ⇒ x
     case _       ⇒ throw new IllegalArgumentException(s"Unknown ${unknown} [${value}] in cluster message")
@@ -139,8 +148,8 @@ class ClusterMessageSerializer(val system: ExtendedActorSystem) extends Serializ
 
   private def gossipToProto(gossip: Gossip): msg.Gossip = {
     import scala.collection.breakOut
-    val allMembers = (gossip.members.iterator ++ gossip.overview.unreachable.iterator).toIndexedSeq
-    val allAddresses: Vector[UniqueAddress] = allMembers.map(_.uniqueAddress)(breakOut)
+    val allMembers = gossip.members.toVector
+    val allAddresses: Vector[UniqueAddress] = allMembers.map(_.uniqueAddress)
     val addressMapping = allAddresses.zipWithIndex.toMap
     val allRoles = allMembers.foldLeft(Set.empty[String])((acc, m) ⇒ acc ++ m.roles).to[Vector]
     val roleMapping = allRoles.zipWithIndex.toMap
@@ -154,11 +163,21 @@ class ClusterMessageSerializer(val system: ExtendedActorSystem) extends Serializ
       msg.Member(mapUniqueAddress(member.uniqueAddress), member.upNumber,
         msg.MemberStatus.valueOf(memberStatusToInt(member.status)), member.roles.map(mapRole)(breakOut))
 
-    val unreachable: Vector[msg.Member] = gossip.overview.unreachable.map(memberToProto)(breakOut)
+    def reachabilityToProto(reachability: Reachability): Vector[msg.ObserverReachability] = {
+      reachability.versions.map {
+        case (observer, version) ⇒
+          val subjectReachability = reachability.recordsFrom(observer).map(r ⇒
+            msg.SubjectReachability(mapUniqueAddress(r.subject),
+              msg.ReachabilityStatus.valueOf(reachabilityStatusToInt(r.status)), r.version))
+          msg.ObserverReachability(mapUniqueAddress(observer), version, subjectReachability)
+      }(breakOut)
+    }
+
+    val reachability = reachabilityToProto(gossip.overview.reachability)
     val members: Vector[msg.Member] = gossip.members.map(memberToProto)(breakOut)
     val seen: Vector[Int] = gossip.overview.seen.map(mapUniqueAddress)(breakOut)
 
-    val overview = msg.GossipOverview(seen, unreachable)
+    val overview = msg.GossipOverview(seen, reachability)
 
     msg.Gossip(allAddresses.map(uniqueAddressToProto),
       allRoles, allHashes, members, overview, vectorClockToProto(gossip.version, hashMapping))
@@ -192,14 +211,31 @@ class ClusterMessageSerializer(val system: ExtendedActorSystem) extends Serializ
     val roleMapping = gossip.allRoles
     val hashMapping = gossip.allHashes
 
+    def reachabilityFromProto(observerReachability: immutable.Seq[msg.ObserverReachability]): Reachability = {
+      val recordBuilder = new immutable.VectorBuilder[Reachability.Record]
+      val versionsBuilder = new scala.collection.mutable.MapBuilder[UniqueAddress, Long, Map[UniqueAddress, Long]](Map.empty)
+      for (o ← observerReachability) {
+        val observer = addressMapping(o.addressIndex)
+        versionsBuilder += ((observer, o.version))
+        for (s ← o.subjectReachability) {
+          val subject = addressMapping(s.addressIndex)
+          val record = Reachability.Record(observer, subject, reachabilityStatusFromInt(s.status), s.version)
+          recordBuilder += record
+        }
+      }
+
+      Reachability.create(recordBuilder.result(), versionsBuilder.result())
+    }
+
     def memberFromProto(member: msg.Member) =
       new Member(addressMapping(member.addressIndex), member.upNumber, memberStatusFromInt(member.status.id),
         member.rolesIndexes.map(roleMapping)(breakOut))
 
     val members: immutable.SortedSet[Member] = gossip.members.map(memberFromProto)(breakOut)
-    val unreachable: immutable.Set[Member] = gossip.overview.unreachable.map(memberFromProto)(breakOut)
+
+    val reachability = reachabilityFromProto(gossip.overview.observerReachability)
     val seen: Set[UniqueAddress] = gossip.overview.seen.map(addressMapping)(breakOut)
-    val overview = GossipOverview(seen, unreachable)
+    val overview = GossipOverview(seen, reachability)
 
     Gossip(members, overview, vectorClockFromProto(gossip.version, hashMapping))
   }