[API-1824] Fix Compact Serialization as per QA (#787)

* Add Compact serialization tests

* Fix compact following QA

* Fix compact following QA + associated fixesÃ

* Misc improvements of tests infrastructures

* Refactor cluster code for issues identified after events re-ordering

* build fix

* refactor fix

* refactor fix

* refactor fix

* refactor fix

* refactor fix

* cleanup

Author: zpqrtbnk

src/Hazelcast.Net.Testing/Accessors/MemberConnectionAccessor.cs

@@ -22,10 +22,10 @@ internal class MemberConnectionAccessor: AccessorBase<MemberConnection>
         public MemberConnectionAccessor(MemberConnection instance): base(instance)
         { }
 
-        public bool Active
+        public bool Connected
         {
-            get => GetField<bool>("_active");
-            set => SetField("_active", value);
+            get => GetField<bool>("_connected");
+            set => SetField("_connected", value);
         }
 
         public Guid MemberId

src/Hazelcast.Net.Testing/RemoteControllerClientExtensions.cs

@@ -160,8 +160,14 @@ public static async Task StopMemberWaitRemovedAsync(this IRemoteControllerClient
             var removed = new SemaphoreSlim(0);
 
             var subscriptionId = await clientInternal.SubscribeAsync(on => on
+                    .StateChanged((sender, args) =>
+                    {
+                        // removing the only member disconnects the client and triggers this event
+                        if (args.State == ClientState.Disconnected) removed.Release();
+                    })
                     .MembersUpdated((sender, args) =>
                     {
+                        // removing one member amongst others triggers this event (but does not disconnect the client)
                         if (args.RemovedMembers.Count > 0) removed.Release();
                     }))
                 .CfAwait();

src/Hazelcast.Net.Tests/Clustering/ConnectMembersTests.cs

@@ -61,6 +61,7 @@ async Task ConnectMembers(MemberConnectionQueue memberConnectionQueue, Cancellat
             var cancellation = new CancellationTokenSource();
             var connecting = ConnectMembers(queue, cancellation.Token);
 
+            queue.Resume(); // the queue is initially suspended, resume
 
             // -- connects
 
@@ -211,6 +212,8 @@ async Task ConnectMembers(MemberConnectionQueue memberConnectionQueue, Cancellat
             var cancellation = new CancellationTokenSource();
             var connecting = ConnectMembers(queue, cancellation.Token);
 
+            queue.Resume(); // the queue is initially suspended, resume
+
             // -- connects
 
             var stopwatch = Stopwatch.StartNew();

src/Hazelcast.Net.Tests/Clustering/MemberConnectionTests.cs

@@ -397,7 +397,7 @@ private MemberConnection NewActiveMemberConnection(MemberInfo member, Authentica
                 new SslOptions(), new Int64Sequence(), loggerFactory
             );
 
-            connection.Accessor().Active = true;
+            connection.Accessor().Connected = true;
             connection.Accessor().MemberId = member.Id;
 
             return connection;

src/Hazelcast.Net.Tests/Clustering/SubscriptionCollectTests.cs

@@ -24,6 +24,7 @@
 using Hazelcast.Protocol.Models;
 using Hazelcast.Serialization;
 using Hazelcast.Testing;
+using Hazelcast.Testing.Logging;
 using Hazelcast.Testing.TestServer;
 using Microsoft.Extensions.Logging.Abstractions;
 using NUnit.Framework;
@@ -92,7 +93,9 @@ public async Task SubscriptionIsCollected()
 
             HConsole.WriteLine(this, "Start client");
 
-            var options = new HazelcastOptionsBuilder().With(options =>
+            var options = new HazelcastOptionsBuilder()
+                .WithHConsoleLogger()
+                .With(options =>
             {
                 options.Networking.Addresses.Add("127.0.0.1:11001");
                 options.Networking.Addresses.Add("127.0.0.1:11002");

src/Hazelcast.Net.Tests/Remote/ClientTests.cs

@@ -23,7 +23,7 @@
 namespace Hazelcast.Tests.Remote
 {
     [TestFixture]
-    [Timeout(10_000)]
+    [Timeout(30_000)]
     public class ClientTests : SingleMemberRemoteTestBase
     {
         [Test]
@@ -40,15 +40,15 @@ public async Task ClientCanConnect()
         }
 
         [Test]
-        public async Task ClientStaringClientWithConfig()
+        public async Task ClientStartingClientWithConfig()
         {
             var clientStarting =  HazelcastClientFactory.GetNewStartingClient(CreateHazelcastOptions());
             await clientStarting.Task;
             await clientStarting.Client.DisposeAsync();
         }
 
         [Test]
-        public async Task ClientStaringClientWithConfig2()
+        public async Task ClientStatringClientWithConfig2()
         {
             var o = CreateHazelcastOptions();
             var clientStarting =  HazelcastClientFactory.GetNewStartingClient(options=>

src/Hazelcast.Net.Tests/Serialization/Compact/CompactQaTests.cs

@@ -0,0 +1,253 @@
+// Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved.
+// 
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// 
+// http://www.apache.org/licenses/LICENSE-2.0
+// 
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Hazelcast.Core;
+using Hazelcast.Exceptions;
+using Hazelcast.Messaging;
+using Hazelcast.Networking;
+using Hazelcast.Partitioning;
+using Hazelcast.Protocol.Codecs;
+using Hazelcast.Serialization;
+using Hazelcast.Testing;
+using Hazelcast.Testing.Logging;
+using NUnit.Framework;
+
+namespace Hazelcast.Tests.Serialization.Compact;
+
+[TestFixture]
+public class CompactQaTests : ClusterRemoteTestBase
+{
+    private IDisposable UseHConsole() => HConsole.Capture(o => o
+        //.WithFilename("console.out")
+        .Configure().SetMaxLevel().EnableTimeStamp(origin: DateTime.Now)
+        .Configure(this).SetPrefix("TEST")
+        .Configure<SocketConnectionBase>().SetIndent(8).SetPrefix("SOCKET").SetLevel(0)
+        .Configure<ClientMessageConnection>().SetMinLevel()
+        .Configure<AsyncContext>().SetMinLevel()
+        .Configure<Partitioner>().SetLevel(1));
+
+    [Test]
+    [Explicit("See comment in test.")]
+    public async Task MemberAddressMatch()
+    {
+        using var _ = UseHConsole();
+
+        var member = await RcClient.StartMemberAsync(RcCluster).CfAwait();
+        await using var cleanup = new DisposeAsyncAction(async () => await RcClient.StopMemberAsync(RcCluster, member));
+
+        var options = CreateHazelcastOptions();
+        options.Networking.ConnectionRetry.ClusterConnectionTimeoutMilliseconds = 10_000;
+        options.Networking.Addresses.Clear();
+        options.Networking.Addresses.Add("127.0.0.2:5701");
+
+        // this relies on the connect queue running all the time, but we have now disabled
+        // the queue when disconnected - so this test cannot work anymore, we will need to
+        // find a different way to do address re-routing in the future, if other clients
+        // do it too.
+
+        await using var client = await HazelcastClientFactory.StartNewClientAsync(options);
+    }
+
+    [TestCase(true)]
+    [TestCase(false)]
+    [Explicit]
+    public async Task ExceptionPreventsClientFromReconnecting(bool recover)
+    {
+        using var _ = UseHConsole();
+
+        // in case a test times out, NUnit just reports that it failed, without further
+        // details - in fact, it does not even say that the test timed out - so by wrapping
+        // the actual test method this way, we make sure that the test "fails" instead of
+        // timing out, and we get proper output (eg HConsole output)
+        await ExceptionPreventsClientFromReconnectingTask(recover).CfAwait(TimeSpan.FromSeconds(60));
+    }
+
+    private async Task ExceptionPreventsClientFromReconnectingTask(bool recover)
+    {
+        var throwException = false;
+
+        // start a member
+        var member = await RcClient.StartMemberAsync(RcCluster).CfAwait();
+        await using var cleanup = new DisposeAsyncAction(async () => await RcClient.StopMemberAsync(RcCluster, member));
+
+        // use a clean client + hook into cluster messaging to capture messages (before client starts)
+        var options = CreateHazelcastOptions();
+        options.Messaging.RetryTimeoutSeconds = 20;
+        await using var client = HazelcastClientFactory.CreateClient(options);
+        // use an internal-level handler, exceptions in user-level handlers are caught
+        client.Cluster.Connections.ConnectionOpened += (_, _, _, _) =>
+        {
+            if (throwException) throw new Exception("bang!");
+            return default;
+        };
+        await client.StartAsync(CancellationToken.None);
+
+        var map = await client.GetMapAsync<int, IGenericRecord>("bar");
+        await map.PutAsync(1, GenericRecordBuilder.Compact("bar1").Build());
+
+        HConsole.WriteLine(this, "-------- STOP MEMBER --------");
+
+        // stop the member, wait until it is actually removed (else we might reconnect to it)
+        await RcClient.StopMemberWaitRemovedAsync(client, RcCluster, member).CfAwait();
+
+        // trigger exceptions
+        throwException = true;
+        if (recover)
+        {
+            async Task RecoverAfter(int delayMilliseconds)
+            {
+                await Task.Delay(delayMilliseconds);
+                throwException = false;
+            }
+
+            var recoverTask = RecoverAfter(5000); // fire-and-forget
+        }
+
+        HConsole.WriteLine(this, "-------- START MEMBER --------");
+
+        // start another member
+        member = await RcClient.StartMemberAsync(RcCluster).CfAwait();
+
+        HConsole.WriteLine(this, "-------- PUT ASYNC --------");
+
+        if (recover)
+        {
+            // this will eventually complete once we're able to reconnect
+            await map.PutAsync(2, GenericRecordBuilder.Compact("bar2").Build()).CfAwait();
+        }
+        else
+        {
+            // this will never complete because we'll never be able to reconnect
+            //
+            // Java: ReconnectMode can be OFF, ON (blocking invocations) or ASYNC (not blocking, triggers HazelcastClientOfflineException)
+            // .NET: ReconnectMode can be DoNotReconnect = OFF, ReconnectSync or ReconnectAsync - but these two have the same effect
+            //
+            // In Java, ASYNC causes any invocation to *immediately* fail with HazelcastClientOfflineException if the client is
+            // reconnecting, whereas ON causes the invocation to be retried, and it may eventually fail with OperationTimeoutException.
+            //
+            // In .NET, invocations are tried (and retried) while the client is reconnecting, until either the client reconnects and
+            // the invocation succeeds, or it times out. So, essentially, .NET is ON and it makes sense that we get a TaskTimeoutException
+            // below.
+
+            await AssertEx.ThrowsAsync<TaskTimeoutException>(async () => await map.PutAsync(2, GenericRecordBuilder.Compact("bar2").Build()).CfAwait());
+        }
+
+        HConsole.WriteLine(this, "-------- STOP MEMBER --------");
+
+        await RcClient.StopMemberAsync(RcCluster, member).CfAwait();
+
+        HConsole.WriteLine(this, "-------- END --------");
+    }
+
+    [Test]
+    public async Task ClusterRestart()
+    {
+        //using var _ = UseHConsole();
+
+        // start a member
+        var member = await RcClient.StartMemberAsync(RcCluster).CfAwait();
+        await using var cleanup = new DisposeAsyncAction(async () => await RcClient.StopMemberAsync(RcCluster, member));
+
+        var messages = new List<ClientMessage>();
+
+        // use a clean client + hook into cluster messaging to capture messages (before client starts)
+        await using var client = HazelcastClientFactory.CreateClient(CreateHazelcastOptions());
+        client.Cluster.Messaging.SendingMessage += message =>
+        {
+            messages.Add(message);
+            return default;
+        };
+        await client.StartAsync(CancellationToken.None);
+
+        var clientMembers = client.Members;
+        Assert.That(clientMembers.Count, Is.EqualTo(1));
+        Assert.That(clientMembers.First().Member.Id, Is.EqualTo(Guid.Parse(member.Uuid)));
+        HConsole.WriteLine(this, $"CONNECTED TO MEMBER {member.Uuid}");
+
+        var map = await client.GetMapAsync<int, IGenericRecord>("bar");
+
+        // add values = will publish the corresponding schemas
+        await map.PutAsync(1, GenericRecordBuilder.Compact("bar1").Build());
+        await map.PutAsync(2, GenericRecordBuilder.Compact("bar2").Build());
+
+        // stop the member, wait until it is actually removed (else we might reconnect to it)
+        await RcClient.StopMemberWaitRemovedAsync(client, RcCluster, member);
+
+        messages.Clear();
+
+        // start another member
+        member = await RcClient.StartMemberAsync(RcCluster);
+
+        // ensure that, eventually, the client is going to connect to *that* other member
+        // and not, because of some timing issues, on the previous one that would not stop
+        // fast enough - StopMemberWaitRemoved waits for the client to lose its connection
+        // to the member but the remote controller has no wait of notifying us that the
+        // member is actually dead for real and not going to accept connections anymore.
+        await AssertEx.SucceedsEventually(() =>
+        {
+            clientMembers = client.Members;
+            Assert.That(clientMembers.Count, Is.EqualTo(1));
+            Assert.That(clientMembers.First().Member.Id, Is.EqualTo(Guid.Parse(member.Uuid)));
+        }, 60_000, 1_000);
+
+        // new member, values are lost
+        var value1 = await map.GetAsync(1);
+        Assert.That(value1, Is.Null);
+
+        // yet we've been reconnected, schemas have been republished
+        Assert.That(messages.Any(x => x.MessageType == ClientSendAllSchemasCodec.RequestMessageType));
+
+        await RcClient.StopMemberAsync(RcCluster, member).CfAwait();
+    }
+
+    [Test]
+    public async Task ReadSchemaAfterWrite_withObjectValueType()
+    {
+        //using var _ = UseHConsole();
+
+        // start a member
+        var member = await RcClient.StartMemberAsync(RcCluster).CfAwait();
+        await using var cleanup = new DisposeAsyncAction(async () => await RcClient.StopMemberAsync(RcCluster, member));
+
+        // use a clean client
+        await using var client = await CreateAndStartClientAsync();
+
+        var map = await client.GetMapAsync<int, object>("bar");
+
+        await map.PutAsync(1, GenericRecordBuilder.Compact("bar1").Build());
+        var value = await map.PutAsync(1, GenericRecordBuilder.Compact("bar2").Build());
+        Assert.That(value, Is.Not.Null);
+        Assert.That(value, Is.InstanceOf<IGenericRecord>());
+
+        await RcClient.StopMemberAsync(RcCluster, member).CfAwait();
+    }
+
+    protected override HazelcastOptions CreateHazelcastOptions()
+    {
+        var options = base.CreateHazelcastOptions();
+        options.Networking.ReconnectMode = ReconnectMode.ReconnectSync;
+        options.Messaging.RetryTimeoutSeconds = 10;
+        return options;
+    }
+
+    protected override HazelcastOptionsBuilder CreateHazelcastOptionsBuilder()
+    {
+        return base.CreateHazelcastOptionsBuilder().WithHConsoleLogger();
+    }
+}

src/Hazelcast.Net.Tests/Serialization/Compact/CompactSerializationSerializerTests.cs

@@ -234,8 +234,18 @@ public void ReadImpossibleType()
 
             var input = new ObjectDataInput(output.ToByteArray(), orw, Endianness.BigEndian);
 
-            // fails because the type cannot be constructed
-            Assert.Throws<SerializationException>(() => serializer.Read(input));
+            // fails because the type cannot be constructed, and therefore we cannot create a
+            // proper compact serialization registration producing an ImpossibleType = we don't
+            // know how to deserialize.
+            Assert.Throws<SerializationException>(() => serializer.Read<ImpossibleType>(input));
+
+            // on the other hand, if we don't specify the type, we indicate that we accept any
+            // object, and therefore since we fail to create a proper registration we fall back
+            // to generic record.
+            input.Position = 0;
+            var obj = serializer.Read(input);
+            Assert.That(obj, Is.Not.Null);
+            Assert.That(obj, Is.InstanceOf<IGenericRecord>());
         }
 
         public class ImpossibleType