-
Notifications
You must be signed in to change notification settings - Fork 1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Fixed Akka.Remote.ResendUnfulfillableException: Unable to fulfill res…
…end request since negatively acknowledged payload is no longer in buffer. (#3914) close #3905 - Fixed Akka.Remote.ResendUnfulfillableException: Unable to fulfill resend request since negatively acknowledged payload is no longer in buffer.
- Loading branch information
1 parent
9cbcbcf
commit 65b4f55
Showing
10 changed files
with
347 additions
and
169 deletions.
There are no files selected for viewing
155 changes: 155 additions & 0 deletions
155
src/core/Akka.Remote.Tests.MultiNode/TransportFailSpec.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,155 @@ | ||
| using System; | ||
| using System.Threading.Tasks; | ||
| using Akka.Actor; | ||
| using Akka.Configuration; | ||
| using Akka.Event; | ||
| using Akka.Remote.TestKit; | ||
| using Akka.Util; | ||
|
|
||
| namespace Akka.Remote.Tests.MultiNode | ||
| { | ||
| public class TransportFailSpecConfig : MultiNodeConfig | ||
| { | ||
| public RoleName First { get; } | ||
| public RoleName Second { get; } | ||
|
|
||
| public TransportFailSpecConfig() | ||
| { | ||
| First = Role("first"); | ||
| Second = Role("second"); | ||
|
|
||
| CommonConfig = DebugConfig(true).WithFallback(ConfigurationFactory.ParseString(@" | ||
| akka.loglevel = INFO | ||
| akka.remote{ | ||
| transport-failure-detector { | ||
| implementation-class = """+ typeof(TestFailureDetector).AssemblyQualifiedName + @""" | ||
| heartbeat-interval = 1 s | ||
| } | ||
| retry-gate-closed-for = 3 s | ||
| # Don't trigger watch Terminated | ||
| watch-failure-detector.acceptable-heartbeat-pause = 60 s | ||
| #use-passive-connections = off | ||
| } | ||
| ")); | ||
| } | ||
|
|
||
| internal static AtomicBoolean FdAvailable = new AtomicBoolean(true); | ||
|
|
||
| /// <summary> | ||
| /// Failure detector implementation that will fail when <see cref="FdAvailable"/> is false. | ||
| /// </summary> | ||
| public class TestFailureDetector : FailureDetector | ||
| { | ||
| public TestFailureDetector(Config config, EventStream eventStream) | ||
| { | ||
|
|
||
| } | ||
|
|
||
| private volatile bool _active = false; | ||
|
|
||
| public override bool IsAvailable => _active ? FdAvailable.Value : true; | ||
|
|
||
| public override bool IsMonitoring => _active; | ||
|
|
||
| public override void HeartBeat() | ||
| { | ||
| _active = true; | ||
| } | ||
| } | ||
|
|
||
| public class Subject : ReceiveActor | ||
| { | ||
| public Subject() | ||
| { | ||
| ReceiveAny(_ => Sender.Tell(_)); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public class TransportFailSpec : MultiNodeSpec | ||
| { | ||
| private readonly TransportFailSpecConfig _config; | ||
|
|
||
| public TransportFailSpec() : this(new TransportFailSpecConfig()) { } | ||
|
|
||
| private TransportFailSpec(TransportFailSpecConfig config) : base(config, typeof(TransportFailSpecConfig)) | ||
| { | ||
| _config = config; | ||
| } | ||
|
|
||
| protected override int InitialParticipantsValueFactory => 2; | ||
|
|
||
| private IActorRef Identify(RoleName role, string actorName) | ||
| { | ||
| var p = CreateTestProbe(); | ||
| Sys.ActorSelection(Node(role) / "user" / actorName).Tell(new Identify(actorName), p.Ref); | ||
| return p.ExpectMsg<ActorIdentity>(RemainingOrDefault).Subject; | ||
| } | ||
|
|
||
| [MultiNodeFact] | ||
| public void TransportFail_should_reconnect() | ||
| { | ||
| RunOn(() => | ||
| { | ||
| EnterBarrier("actors-started"); | ||
| var subject = Identify(_config.Second, "subject"); | ||
| Watch(subject); | ||
| subject.Tell("hello"); | ||
| ExpectMsg("hello"); | ||
| }, _config.First); | ||
|
|
||
| RunOn(() => | ||
| { | ||
| Sys.ActorOf(Props.Create(() => new TransportFailSpecConfig.Subject()), "subject"); | ||
| EnterBarrier("actors-started"); | ||
| }, _config.Second); | ||
|
|
||
| EnterBarrier("watch-established"); | ||
|
|
||
| // trigger transport failure detector | ||
| TransportFailSpecConfig.FdAvailable.GetAndSet(false); | ||
|
|
||
| // wait for ungated (also later awaitAssert retry) | ||
| Task.Delay(RARP.For(Sys).Provider.RemoteSettings.RetryGateClosedFor).Wait(); | ||
| TransportFailSpecConfig.FdAvailable.GetAndSet(true); | ||
|
|
||
| RunOn(() => | ||
| { | ||
| EnterBarrier("actors-started2"); | ||
| var quarantineProbe = CreateTestProbe(); | ||
| Sys.EventStream.Subscribe(quarantineProbe.Ref, typeof(QuarantinedEvent)); | ||
| IActorRef subject2 = null; | ||
| AwaitAssert(() => | ||
| { | ||
| // TODO: harden | ||
| Within(TimeSpan.FromSeconds(3), () => | ||
| { | ||
| AwaitCondition(() => | ||
| { | ||
| subject2 = Identify(_config.Second, "subject2"); | ||
| return subject2 != null; | ||
| }, RemainingOrDefault, TimeSpan.FromSeconds(1)); | ||
| }); | ||
| }, TimeSpan.FromSeconds(5)); | ||
| Watch(subject2); | ||
| quarantineProbe.ExpectNoMsg(TimeSpan.FromSeconds(1)); | ||
| subject2.Tell("hello2"); | ||
| ExpectMsg("hello2"); | ||
| EnterBarrier("watch-established2"); | ||
| ExpectTerminated(subject2); | ||
| }, _config.First); | ||
|
|
||
| RunOn(() => | ||
| { | ||
| var subject2 = Sys.ActorOf(Props.Create(() => new TransportFailSpecConfig.Subject()), "subject2"); | ||
| EnterBarrier("actors-started2"); | ||
| EnterBarrier("watch-established2"); | ||
| subject2.Tell(PoisonPill.Instance); | ||
| }, _config.Second); | ||
|
|
||
| EnterBarrier("done"); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.