Convert Akka.Cluster.Sharding.Tests.MultiNode to async (#7758)

* Convert Akka.Cluster.Sharding.Tests.MultiNode to async

- Convert all test methods to async Task
- Replace TestConductor blocking calls with async versions
- Replace EnterBarrier with EnterBarrierAsync
- Use RunOnAsync for async operations
- Use WithinAsync for async timing constraints
- Add using System.Threading.Tasks

This migration prevents thread pool starvation in CI environments
by replacing blocking .Wait() calls with proper async/await patterns.

* Fix async impl

---------

Co-authored-by: Gregorius Soedharmo <[email protected]>

Author: Aaronontheweb

src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardCoordinatorDowning2Spec.cs

@@ -6,6 +6,7 @@
 //-----------------------------------------------------------------------
 
 using System;
+using System.Threading.Tasks;
 using Akka.Actor;
 using Akka.Configuration;
 using Akka.MultiNode.TestAdapter;
@@ -14,123 +15,122 @@
 using Akka.TestKit;
 using FluentAssertions;
 
-namespace Akka.Cluster.Sharding.Tests
+namespace Akka.Cluster.Sharding.Tests;
+
+public class ClusterShardingSingleShardPerEntitySpecConfig : MultiNodeClusterShardingConfig
 {
-    public class ClusterShardingSingleShardPerEntitySpecConfig : MultiNodeClusterShardingConfig
-    {
-        public RoleName First { get; }
-        public RoleName Second { get; }
-        public RoleName Third { get; }
-        public RoleName Fourth { get; }
-        public RoleName Fifth { get; }
+    public RoleName First { get; }
+    public RoleName Second { get; }
+    public RoleName Third { get; }
+    public RoleName Fourth { get; }
+    public RoleName Fifth { get; }
 
-        public Config R1Config { get; }
-        public Config R2Config { get; }
+    public Config R1Config { get; }
+    public Config R2Config { get; }
 
-        public ClusterShardingSingleShardPerEntitySpecConfig()
-            : base(loglevel: "DEBUG", additionalConfig: @"
+    public ClusterShardingSingleShardPerEntitySpecConfig()
+        : base(loglevel: "DEBUG", additionalConfig: @"
                 akka.cluster.sharding.updating-state-timeout = 1s
             ")
-        {
-            First = Role("first");
-            Second = Role("second");
-            Third = Role("third");
-            Fourth = Role("fourth");
-            Fifth = Role("fifth");
-
-            TestTransport = true;
-        }
+    {
+        First = Role("first");
+        Second = Role("second");
+        Third = Role("third");
+        Fourth = Role("fourth");
+        Fifth = Role("fifth");
+
+        TestTransport = true;
     }
+}
 
-    public class ClusterShardingSingleShardPerEntitySpec : MultiNodeClusterShardingSpec<ClusterShardingSingleShardPerEntitySpecConfig>
-    {
-        #region setup
+public class ClusterShardingSingleShardPerEntitySpec : MultiNodeClusterShardingSpec<ClusterShardingSingleShardPerEntitySpecConfig>
+{
+    #region setup
 
-        private readonly Lazy<IActorRef> _region;
+    private readonly Lazy<IActorRef> _region;
 
-        public ClusterShardingSingleShardPerEntitySpec()
-            : this(new ClusterShardingSingleShardPerEntitySpecConfig(), typeof(ClusterShardingSingleShardPerEntitySpec))
-        {
-        }
+    public ClusterShardingSingleShardPerEntitySpec()
+        : this(new ClusterShardingSingleShardPerEntitySpecConfig(), typeof(ClusterShardingSingleShardPerEntitySpec))
+    {
+    }
 
-        protected ClusterShardingSingleShardPerEntitySpec(ClusterShardingSingleShardPerEntitySpecConfig config, Type type)
-            : base(config, type)
-        {
-            _region = new Lazy<IActorRef>(() => ClusterSharding.Get(Sys).ShardRegion("Entity"));
-        }
+    protected ClusterShardingSingleShardPerEntitySpec(ClusterShardingSingleShardPerEntitySpecConfig config, Type type)
+        : base(config, type)
+    {
+        _region = new Lazy<IActorRef>(() => ClusterSharding.Get(Sys).ShardRegion("Entity"));
+    }
 
-        private void Join(RoleName from, RoleName to)
-        {
-            Join(
-                from,
-                to,
-                () => StartSharding(
-                    Sys,
-                    typeName: "Entity",
-                    entityProps: Props.Create(() => new ShardedEntity())));
-        }
-
-        private void JoinAndAllocate(RoleName node, int entityId)
+    private Task JoinAsync(RoleName from, RoleName to)
+    {
+        return JoinAsync(
+            from,
+            to,
+            () => StartSharding(
+                Sys,
+                typeName: "Entity",
+                entityProps: Props.Create(() => new ShardedEntity())));
+    }
+
+    private async Task JoinAndAllocate(RoleName node, int entityId)
+    {
+        await WithinAsync(TimeSpan.FromSeconds(10), async () =>
         {
-            Within(TimeSpan.FromSeconds(10), () =>
+            await JoinAsync(node, Config.First);
+            await RunOnAsync(async () =>
             {
-                Join(node, Config.First);
-                RunOn(() =>
-                {
-                    _region.Value.Tell(entityId);
+                _region.Value.Tell(entityId);
 
-                    ExpectMsg(entityId);
+                await ExpectMsgAsync(entityId);
 
-                    LastSender.Path.Should().Be(_region.Value.Path / $"{entityId}" / $"{entityId}");
-                }, node);
-            });
-            EnterBarrier($"started-{entityId}");
-        }
+                LastSender.Path.Should().Be(_region.Value.Path / $"{entityId}" / $"{entityId}");
+            }, node);
+        });
+        await EnterBarrierAsync($"started-{entityId}");
+    }
 
 
-        #endregion
+    #endregion
 
-        [MultiNodeFact]
-        public void Cluster_sharding_with_single_shard_per_entity_specs()
-        {
-            Cluster_sharding_with_single_shard_per_entity_must_use_specified_region();
-        }
+    [MultiNodeFact]
+    public async Task Cluster_sharding_with_single_shard_per_entity_specs()
+    {
+        await Cluster_sharding_with_single_shard_per_entity_must_use_specified_region();
+    }
 
-        private void Cluster_sharding_with_single_shard_per_entity_must_use_specified_region()
-        {
-            JoinAndAllocate(Config.First, 1);
-            JoinAndAllocate(Config.Second, 2);
-            JoinAndAllocate(Config.Third, 3);
-            JoinAndAllocate(Config.Fourth, 4);
-            JoinAndAllocate(Config.Fifth, 5);
+    private async Task Cluster_sharding_with_single_shard_per_entity_must_use_specified_region()
+    {
+        await JoinAndAllocate(Config.First, 1);
+        await JoinAndAllocate(Config.Second, 2);
+        await JoinAndAllocate(Config.Third, 3);
+        await JoinAndAllocate(Config.Fourth, 4);
+        await JoinAndAllocate(Config.Fifth, 5);
 
-            RunOn(() =>
-            {
-                // coordinator is on 'first', blackhole 3 other means that it can't update with WriteMajority
-                TestConductor.Blackhole(Config.First, Config.Third, ThrottleTransportAdapter.Direction.Both).Wait();
-                TestConductor.Blackhole(Config.First, Config.Fourth, ThrottleTransportAdapter.Direction.Both).Wait();
-                TestConductor.Blackhole(Config.First, Config.Fifth, ThrottleTransportAdapter.Direction.Both).Wait();
-
-                // shard 6 not allocated yet and due to the blackhole it will not be completed
-                _region.Value.Tell(6);
-
-                // shard 1 location is know by 'first' region, not involving coordinator
-                _region.Value.Tell(1);
-                ExpectMsg(1);
-
-                // shard 2 location not known at 'first' region yet, but coordinator is on 'first' and should
-                // be able to answer GetShardHome even though previous request for shard 4 has not completed yet
-                _region.Value.Tell(2);
-                ExpectMsg(2);
-                LastSender.Path.Should().Be(Node(Config.Second) / "system" / "sharding" / "Entity" / "2" / "2");
-
-                TestConductor.PassThrough(Config.First, Config.Third, ThrottleTransportAdapter.Direction.Both).Wait();
-                TestConductor.PassThrough(Config.First, Config.Fourth, ThrottleTransportAdapter.Direction.Both).Wait();
-                TestConductor.PassThrough(Config.First, Config.Fifth, ThrottleTransportAdapter.Direction.Both).Wait();
-                ExpectMsg(6, TimeSpan.FromSeconds(10));
-            }, Config.First);
-
-            EnterBarrier("after-1");
-        }
+        await RunOnAsync(async () =>
+        {
+            // coordinator is on 'first', blackhole 3 other means that it can't update with WriteMajority
+            await TestConductor.BlackholeAsync(Config.First, Config.Third, ThrottleTransportAdapter.Direction.Both);
+            await TestConductor.BlackholeAsync(Config.First, Config.Fourth, ThrottleTransportAdapter.Direction.Both);
+            await TestConductor.BlackholeAsync(Config.First, Config.Fifth, ThrottleTransportAdapter.Direction.Both);
+
+            // shard 6 not allocated yet and due to the blackhole it will not be completed
+            _region.Value.Tell(6);
+
+            // shard 1 location is know by 'first' region, not involving coordinator
+            _region.Value.Tell(1);
+            await ExpectMsgAsync(1);
+
+            // shard 2 location not known at 'first' region yet, but coordinator is on 'first' and should
+            // be able to answer GetShardHome even though previous request for shard 4 has not completed yet
+            _region.Value.Tell(2);
+            await ExpectMsgAsync(2);
+            LastSender.Path.Should().Be(await NodeAsync(Config.Second) / "system" / "sharding" / "Entity" / "2" / "2");
+
+            await TestConductor.PassThroughAsync(Config.First, Config.Third, ThrottleTransportAdapter.Direction.Both);
+            await TestConductor.PassThroughAsync(Config.First, Config.Fourth, ThrottleTransportAdapter.Direction.Both);
+            await TestConductor.PassThroughAsync(Config.First, Config.Fifth, ThrottleTransportAdapter.Direction.Both);
+            await ExpectMsgAsync(6, TimeSpan.FromSeconds(10));
+        }, Config.First);
+
+        await EnterBarrierAsync("after-1");
     }
-}
+}