|
| 1 | +/* |
| 2 | + * Licensed to the Apache Software Foundation (ASF) under one or more |
| 3 | + * license agreements; and to You under the Apache License, version 2.0: |
| 4 | + * |
| 5 | + * https://www.apache.org/licenses/LICENSE-2.0 |
| 6 | + * |
| 7 | + * This file is part of the Apache Pekko project, which was derived from Akka. |
| 8 | + */ |
| 9 | + |
| 10 | +/* |
| 11 | + * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com> |
| 12 | + */ |
| 13 | + |
| 14 | +package org.apache.pekko.remote.artery |
| 15 | + |
| 16 | +import scala.concurrent.Future |
| 17 | +import scala.concurrent.Promise |
| 18 | + |
| 19 | +import org.scalatest.concurrent.Eventually |
| 20 | +import org.scalatest.time.Span |
| 21 | + |
| 22 | +import org.apache.pekko |
| 23 | +import pekko.actor.ActorRef |
| 24 | +import pekko.actor.ActorSystem |
| 25 | +import pekko.actor.Address |
| 26 | +import pekko.actor.RootActorPath |
| 27 | +import pekko.remote.RARP |
| 28 | +import pekko.remote.UniqueAddress |
| 29 | +import pekko.testkit.ImplicitSender |
| 30 | +import pekko.testkit.TestActors |
| 31 | +import pekko.testkit.TestProbe |
| 32 | + |
| 33 | +class HarmlessQuarantineSpec extends ArteryMultiNodeSpec(""" |
| 34 | + pekko.loglevel=INFO |
| 35 | + pekko.remote.artery.propagate-harmless-quarantine-events = off |
| 36 | + pekko.remote.artery.advanced { |
| 37 | + stop-idle-outbound-after = 1 s |
| 38 | + connection-timeout = 2 s |
| 39 | + remove-quarantined-association-after = 1 s |
| 40 | + compression { |
| 41 | + actor-refs.advertisement-interval = 5 seconds |
| 42 | + } |
| 43 | + } |
| 44 | + """) with ImplicitSender with Eventually { |
| 45 | + |
| 46 | + override implicit val patience: PatienceConfig = { |
| 47 | + import pekko.testkit.TestDuration |
| 48 | + PatienceConfig(testKitSettings.DefaultTimeout.duration.dilated * 2, Span(200, org.scalatest.time.Millis)) |
| 49 | + } |
| 50 | + |
| 51 | + private def futureUniqueRemoteAddress(association: Association): Future[UniqueAddress] = { |
| 52 | + val p = Promise[UniqueAddress]() |
| 53 | + association.associationState.addUniqueRemoteAddressListener(a => p.success(a)) |
| 54 | + p.future |
| 55 | + } |
| 56 | + |
| 57 | + "Harmless Quarantine Events" should { |
| 58 | + |
| 59 | + "eliminate quarantined association when not used - echo test" in withAssociation { |
| 60 | + (remoteSystem, remoteAddress, _, localArtery, localProbe) => |
| 61 | + // event to watch out for, indicator of the issue |
| 62 | + remoteSystem.eventStream.subscribe(testActor, classOf[ThisActorSystemQuarantinedEvent]) |
| 63 | + |
| 64 | + val remoteEcho = remoteSystem.actorSelection("/user/echo").resolveOne(remainingOrDefault).futureValue |
| 65 | + |
| 66 | + val localAddress = RARP(system).provider.getDefaultAddress |
| 67 | + |
| 68 | + val localEchoRef = |
| 69 | + remoteSystem.actorSelection(RootActorPath(localAddress) / localProbe.ref.path.elements).resolveOne( |
| 70 | + remainingOrDefault).futureValue |
| 71 | + remoteEcho.tell("ping", localEchoRef) |
| 72 | + localProbe.expectMsg("ping") |
| 73 | + |
| 74 | + val association = localArtery.association(remoteAddress) |
| 75 | + val remoteUid = futureUniqueRemoteAddress(association).futureValue.uid |
| 76 | + localArtery.quarantine(remoteAddress, Some(remoteUid), "Test") |
| 77 | + association.associationState.isQuarantined(remoteUid) shouldBe true |
| 78 | + association.associationState.quarantinedButHarmless(remoteUid) shouldBe false |
| 79 | + |
| 80 | + remoteEcho.tell("ping", localEchoRef) // trigger sending message from remote to local, which will trigger local to wrongfully notify remote that it is quarantined |
| 81 | + eventually { |
| 82 | + expectMsgType[ThisActorSystemQuarantinedEvent] // this is what remote emits when it learns it is quarantined by local |
| 83 | + } |
| 84 | + } |
| 85 | + |
| 86 | + "eliminate quarantined association when not used - echo test (harmless=true)" in withAssociation { |
| 87 | + (remoteSystem, remoteAddress, _, localArtery, localProbe) => |
| 88 | + // event to watch out for, indicator of the issue |
| 89 | + remoteSystem.eventStream.subscribe(testActor, classOf[ThisActorSystemQuarantinedEvent]) |
| 90 | + |
| 91 | + val remoteEcho = remoteSystem.actorSelection("/user/echo").resolveOne(remainingOrDefault).futureValue |
| 92 | + |
| 93 | + val localAddress = RARP(system).provider.getDefaultAddress |
| 94 | + |
| 95 | + val localEchoRef = |
| 96 | + remoteSystem.actorSelection(RootActorPath(localAddress) / localProbe.ref.path.elements).resolveOne( |
| 97 | + remainingOrDefault).futureValue |
| 98 | + remoteEcho.tell("ping", localEchoRef) |
| 99 | + localProbe.expectMsg("ping") |
| 100 | + |
| 101 | + val association = localArtery.association(remoteAddress) |
| 102 | + val remoteUid = futureUniqueRemoteAddress(association).futureValue.uid |
| 103 | + localArtery.quarantine(remoteAddress, Some(remoteUid), "HarmlessTest", harmless = true) |
| 104 | + association.associationState.isQuarantined(remoteUid) shouldBe true |
| 105 | + association.associationState.quarantinedButHarmless(remoteUid) shouldBe true |
| 106 | + |
| 107 | + remoteEcho.tell("ping", localEchoRef) // trigger sending message from remote to local, which will trigger local to wrongfully notify remote that it is quarantined |
| 108 | + eventually { |
| 109 | + expectNoMessage() |
| 110 | + } |
| 111 | + } |
| 112 | + |
| 113 | + /** |
| 114 | + * Test setup fixture: |
| 115 | + * 1. A 'remote' ActorSystem is created to spawn an Echo actor, |
| 116 | + * 2. A TestProbe is spawned locally to initiate communication with the Echo actor |
| 117 | + * 3. Details (remoteAddress, remoteEcho, localArtery, localProbe) are supplied to the test |
| 118 | + */ |
| 119 | + def withAssociation(test: (ActorSystem, Address, ActorRef, ArteryTransport, TestProbe) => Any): Unit = { |
| 120 | + val remoteSystem = newRemoteSystem() |
| 121 | + try { |
| 122 | + remoteSystem.actorOf(TestActors.echoActorProps, "echo") |
| 123 | + val remoteAddress = RARP(remoteSystem).provider.getDefaultAddress |
| 124 | + |
| 125 | + def remoteEcho = system.actorSelection(RootActorPath(remoteAddress) / "user" / "echo") |
| 126 | + |
| 127 | + val echoRef = remoteEcho.resolveOne(remainingOrDefault).futureValue |
| 128 | + val localProbe = new TestProbe(localSystem) |
| 129 | + |
| 130 | + echoRef.tell("ping", localProbe.ref) |
| 131 | + localProbe.expectMsg("ping") |
| 132 | + |
| 133 | + val artery = RARP(system).provider.transport.asInstanceOf[ArteryTransport] |
| 134 | + |
| 135 | + test(remoteSystem, remoteAddress, echoRef, artery, localProbe) |
| 136 | + |
| 137 | + } finally { |
| 138 | + shutdown(remoteSystem) |
| 139 | + } |
| 140 | + } |
| 141 | + } |
| 142 | +} |
0 commit comments