-
Notifications
You must be signed in to change notification settings - Fork 1k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
pass Akka.Cluster.Cluster
into IDowningProvider
directly
#5965
Changes from all commits
ddfe27f
bad566d
3d2aacb
a636343
119f90d
7d93324
a6fa54c
6bd120f
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,88 @@ | ||
// //----------------------------------------------------------------------- | ||
// // <copyright file="Bugfix5962Spec.cs" company="Akka.NET Project"> | ||
// // Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com> | ||
// // Copyright (C) 2013-2022 .NET Foundation <https://github.com/akkadotnet/akka.net> | ||
// // </copyright> | ||
// //----------------------------------------------------------------------- | ||
|
||
using System; | ||
using System.Threading.Tasks; | ||
using Akka.Configuration; | ||
using Akka.TestKit; | ||
using FluentAssertions; | ||
using FluentAssertions.Extensions; | ||
using Xunit; | ||
using Xunit.Abstractions; | ||
using static FluentAssertions.FluentActions; | ||
|
||
|
||
namespace Akka.Cluster.Tests | ||
{ | ||
public class Bugfix5962Spec : TestKit.Xunit2.TestKit | ||
{ | ||
private static readonly Config Config = ConfigurationFactory.ParseString(@" | ||
akka { | ||
loglevel = INFO | ||
actor { | ||
provider = cluster | ||
default-dispatcher = { | ||
executor = channel-executor | ||
channel-executor.priority = normal | ||
} | ||
# Adding this part in combination with the SplitBrainResolverProvider causes the error | ||
internal-dispatcher = { | ||
executor = channel-executor | ||
channel-executor.priority = high | ||
} | ||
} | ||
remote { | ||
dot-netty.tcp { | ||
port = 15508 | ||
hostname = ""127.0.0.1"" | ||
} | ||
default-remote-dispatcher { | ||
executor = channel-executor | ||
channel-executor.priority = high | ||
} | ||
backoff-remote-dispatcher { | ||
executor = channel-executor | ||
channel-executor.priority = low | ||
} | ||
} | ||
cluster { | ||
seed-nodes = [""akka.tcp://Bugfix5962Spec@127.0.0.1:15508""] | ||
downing-provider-class = ""Akka.Cluster.SBR.SplitBrainResolverProvider, Akka.Cluster"" | ||
} | ||
}"); | ||
|
||
private readonly Type _timerMsgType; | ||
|
||
public Bugfix5962Spec(ITestOutputHelper output): base(Config, nameof(Bugfix5962Spec), output) | ||
{ | ||
_timerMsgType = Type.GetType("Akka.Actor.Scheduler.TimerScheduler+TimerMsg, Akka"); | ||
} | ||
|
||
[Fact] | ||
public async Task SBR_Should_work_with_channel_executor() | ||
{ | ||
var latch = new TestLatch(1); | ||
var cluster = Cluster.Get(Sys); | ||
cluster.RegisterOnMemberUp(() => | ||
{ | ||
latch.CountDown(); | ||
}); | ||
|
||
var selection = Sys.ActorSelection("akka://Bugfix5962Spec/system/cluster/core/daemon/downingProvider"); | ||
|
||
await Awaiting(() => selection.ResolveOne(1.Seconds())) | ||
.Should().NotThrowAsync("Downing provider should be alive. ActorSelection will throw an ActorNotFoundException if this fails"); | ||
|
||
// There should be no TimerMsg being sent to dead letter, this signals that the downing provider is dead | ||
await EventFilter.DeadLetter(_timerMsgType).ExpectAsync(0, async () => | ||
{ | ||
latch.Ready(1.Seconds()); | ||
await Task.Delay(2.Seconds()); | ||
}); | ||
} | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -7,13 +7,15 @@ | |
|
||
using System; | ||
using System.Threading; | ||
using System.Threading.Tasks; | ||
using Akka.Actor; | ||
using Akka.Actor.Dsl; | ||
using Akka.Configuration; | ||
using Akka.Event; | ||
using Akka.TestKit; | ||
using Akka.Util; | ||
using Xunit; | ||
using Xunit.Abstractions; | ||
|
||
namespace Akka.Cluster.Tests | ||
{ | ||
|
@@ -25,12 +27,13 @@ public class StartupWithOneThreadSpec : AkkaSpec | |
akka.actor.default-dispatcher.dedicated-thread-pool.thread-count = 1 | ||
akka.actor.provider = ""Akka.Cluster.ClusterActorRefProvider, Akka.Cluster"" | ||
akka.remote.dot-netty.tcp.port = 0 | ||
akka.cluster.downing-provider-class = ""Akka.Cluster.SBR.SplitBrainResolverProvider, Akka.Cluster"" | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Added SBR to the single thread startup spec to see if the deadletter'd |
||
akka.cluster.split-brain-resolver.active-strategy = keep-majority | ||
"); | ||
|
||
private long _startTime; | ||
|
||
public StartupWithOneThreadSpec() : base(Configuration) | ||
{ | ||
public StartupWithOneThreadSpec(ITestOutputHelper output) : base(Configuration, output) { | ||
_startTime = MonotonicClock.GetTicks(); | ||
} | ||
|
||
|
@@ -53,7 +56,7 @@ private Props TestProps | |
} | ||
|
||
[Fact] | ||
public void A_cluster_must_startup_with_one_dispatcher_thread() | ||
public async Task A_cluster_must_startup_with_one_dispatcher_thread() | ||
{ | ||
// This test failed before fixing https://github.com/akkadotnet/akka.net/issues/1959 when adding a sleep before the | ||
// Await of GetClusterCoreRef in the Cluster extension constructor. | ||
|
@@ -75,6 +78,11 @@ public void A_cluster_must_startup_with_one_dispatcher_thread() | |
ExpectMsg("hello"); | ||
ExpectMsg("hello"); | ||
ExpectMsg("hello"); | ||
|
||
// perform a self-join | ||
var cts = new CancellationTokenSource(TimeSpan.FromSeconds((3))); | ||
var selfAddress = cluster.SelfAddress; | ||
await cluster.JoinSeedNodesAsync(new[] { selfAddress }, cts.Token); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Force actual cluster formation to complete successfully as part of spec. |
||
} | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -132,7 +132,7 @@ public Cluster(ActorSystemImpl system) | |
Scheduler = CreateScheduler(system); | ||
|
||
// it has to be lazy - otherwise if downing provider will init a cluster itself, it will deadlock | ||
_downingProvider = new Lazy<IDowningProvider>(() => Akka.Cluster.DowningProvider.Load(Settings.DowningProviderType, system), LazyThreadSafetyMode.ExecutionAndPublication); | ||
_downingProvider = new Lazy<IDowningProvider>(() => Akka.Cluster.DowningProvider.Load(Settings.DowningProviderType, system, this), LazyThreadSafetyMode.ExecutionAndPublication); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is where |
||
|
||
//create supervisor for daemons under path "/system/cluster" | ||
_clusterDaemons = system.SystemActorOf(Props.Create(() => new ClusterDaemon(Settings)).WithDeploy(Deploy.Local), "cluster"); | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -45,20 +45,18 @@ public interface IDowningProvider | |
public sealed class NoDowning : IDowningProvider | ||
{ | ||
private readonly ActorSystem _system; | ||
|
||
/// <summary> | ||
/// TBD | ||
/// </summary> | ||
/// <param name="system">TBD</param> | ||
public NoDowning(ActorSystem system) | ||
private readonly Cluster _cluster; | ||
|
||
public NoDowning(ActorSystem system, Cluster cluster) | ||
{ | ||
_system = system; | ||
_cluster = cluster; | ||
} | ||
|
||
/// <summary> | ||
/// TBD | ||
/// </summary> | ||
public TimeSpan DownRemovalMargin => Cluster.Get(_system).Settings.DownRemovalMargin; | ||
public TimeSpan DownRemovalMargin => _cluster.Settings.DownRemovalMargin; | ||
|
||
/// <summary> | ||
/// TBD | ||
|
@@ -72,20 +70,25 @@ public NoDowning(ActorSystem system) | |
internal static class DowningProvider | ||
{ | ||
/// <summary> | ||
/// TBD | ||
/// Loads the <see cref="IDowningProvider"/> from configuration and instantiates it via reflection. | ||
/// </summary> | ||
/// <param name="downingProviderType">TBD</param> | ||
/// <param name="system">TBD</param> | ||
/// <param name="cluster">The current cluster object.</param> | ||
/// <exception cref="ConfigurationException"> | ||
/// This exception is thrown when the specified <paramref name="downingProviderType"/> does not implement <see cref="IDowningProvider"/>. | ||
/// </exception> | ||
/// <returns>TBD</returns> | ||
public static IDowningProvider Load(Type downingProviderType, ActorSystem system) | ||
/// <returns>The activated <see cref="IDowningProvider"/></returns> | ||
/// <remarks> | ||
/// Required to pass in <see cref="Akka.Cluster.Cluster"/> manually here since https://github.com/akkadotnet/akka.net/issues/5962 | ||
/// can cause the SBR startup to fail when running with the `channel-executor`. | ||
/// </remarks> | ||
public static IDowningProvider Load(Type downingProviderType, ActorSystem system, Cluster cluster) | ||
{ | ||
var extendedSystem = system as ExtendedActorSystem; | ||
try | ||
{ | ||
return (IDowningProvider)Activator.CreateInstance(downingProviderType, extendedSystem); | ||
return (IDowningProvider)Activator.CreateInstance(downingProviderType, extendedSystem, cluster); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Pass this into the new constructor via reflection - this line will throw and fail the Akka.Cluster startup sequence for any |
||
} | ||
catch (Exception e) | ||
{ | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -27,22 +27,22 @@ internal class SplitBrainResolver : SplitBrainResolverBase | |
{ | ||
private Cluster _cluster; | ||
|
||
public SplitBrainResolver(TimeSpan stableAfter, DowningStrategy strategy) | ||
public SplitBrainResolver(TimeSpan stableAfter, DowningStrategy strategy, Cluster cluster) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Modern SBR updates. |
||
: base(stableAfter, strategy) | ||
{ | ||
_cluster = cluster; | ||
} | ||
|
||
public override UniqueAddress SelfUniqueAddress => _cluster.SelfUniqueAddress; | ||
|
||
public static Props Props2(TimeSpan stableAfter, DowningStrategy strategy) | ||
public static Props Props2(TimeSpan stableAfter, DowningStrategy strategy, Cluster cluster) | ||
{ | ||
return Props.Create(() => new SplitBrainResolver(stableAfter, strategy)); | ||
return Props.Create(() => new SplitBrainResolver(stableAfter, strategy, cluster)); | ||
} | ||
|
||
// re-subscribe when restart | ||
protected override void PreStart() | ||
{ | ||
_cluster = Cluster.Get(Context.System); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Bugfix 2. |
||
_cluster.Subscribe(Self, InitialStateAsEvents, typeof(IClusterDomainEvent)); | ||
|
||
base.PreStart(); | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Breaking API change, by design - forces the
Cluster
to be passed into the SBR directly rather than resolved viaCluster.Get
, which is what triggers this issue.