Fixes for Kafka Tests following the Proactor/Reactor split

src/Paramore.Brighter.MessagingGateway.Kafka/KafkaDefaultMessageHeaderBuilder.cs

@@ -47,10 +47,13 @@ public Headers Build(Message message)
                 new Header(HeaderNames.MESSAGE_ID, message.Header.MessageId.ToByteArray()),
             };
 
-            if (message.Header.TimeStamp != default)
-                headers.Add(HeaderNames.TIMESTAMP, message.Header.TimeStamp.ToString().ToByteArray());
-            else
-                headers.Add(HeaderNames.TIMESTAMP, DateTimeOffset.UtcNow.ToString().ToByteArray());
+            string timeStampAsString = DateTimeOffset.UtcNow.DateTime.ToString(CultureInfo.InvariantCulture);
+            if (message.Header.TimeStamp.DateTime != default)
+            {
+                timeStampAsString = message.Header.TimeStamp.DateTime.ToString(CultureInfo.InvariantCulture);
+            }
+
+            headers.Add(HeaderNames.TIMESTAMP, timeStampAsString.ToByteArray());
 
             if (message.Header.CorrelationId != string.Empty)
                 headers.Add(HeaderNames.CORRELATION_ID, message.Header.CorrelationId.ToByteArray());
@@ -64,7 +67,7 @@ public Headers Build(Message message)
             if (!string.IsNullOrEmpty(message.Header.ReplyTo))
                 headers.Add(HeaderNames.REPLY_TO, message.Header.ReplyTo.ToByteArray());
 
-            headers.Add(HeaderNames.DELAYED_MILLISECONDS, message.Header.Delayed.ToString().ToByteArray());
+            headers.Add(HeaderNames.DELAYED_MILLISECONDS, message.Header.Delayed.TotalMilliseconds.ToString(CultureInfo.InvariantCulture).ToByteArray());
 
             headers.Add(HeaderNames.HANDLED_COUNT, message.Header.HandledCount.ToString().ToByteArray());
 

src/Paramore.Brighter.MessagingGateway.Kafka/KafkaMessageConsumer.cs

@@ -206,7 +206,7 @@ public KafkaMessageConsumer(
                 .Build();
 
             s_logger.LogInformation("Kafka consumer subscribing to {Topic}", Topic);
-            _consumer.Subscribe(new []{ Topic.Value });
+            _consumer.Subscribe([Topic.Value]);
 
             _creator = new KafkaMessageCreator();
 

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_a_set_of_messages_is_sent_preserve_order_on_a_confluent_cluster_async.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Confluent/Proactor/When_a_set_of_messages_is_sent_preserve_order_on_a_confluent_cluster_async.cs

@@ -10,7 +10,7 @@
 using Xunit;
 using Xunit.Abstractions;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Confluent.Proactor;
 
 [Trait("Category", "Kafka")]
 [Trait("Category", "Confluent")]
@@ -131,15 +131,17 @@ private async Task<string> SendMessageAsync()
 
     private async Task<IEnumerable<Message>> ConsumeMessagesAsync(IAmAMessageConsumerAsync consumer)
     {
-        var messages = new Message[0];
+        var messages = Array.Empty<Message>();
         int maxTries = 0;
         do
         {
             try
             {
                 maxTries++;
-                await Task.Delay(500); //Let topic propagate in the broker
-                messages = await consumer.ReceiveAsync(TimeSpan.FromMilliseconds(1000));
+                //Let topic propagate in the broker
+                await Task.Delay(500); 
+                //use TimeSpan.Zero to avoid blocking
+                messages = await consumer.ReceiveAsync(TimeSpan.Zero);
 
                 if (messages[0].Header.MessageType != MessageType.MT_NONE)
                     break;

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_consumer_assumes_topic_but_missing_on_a_confluent_cluster_async.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Confluent/Proactor/When_consumer_assumes_topic_but_missing_on_a_confluent_cluster_async.cs

@@ -1,17 +1,11 @@
 using System;
-using System.Linq;
 using System.Runtime.InteropServices;
 using System.Threading.Tasks;
-using Confluent.Kafka;
 using FluentAssertions;
-using Paramore.Brighter.Kafka.Tests.TestDoubles;
 using Paramore.Brighter.MessagingGateway.Kafka;
 using Xunit;
-using Xunit.Abstractions;
-using SaslMechanism = Paramore.Brighter.MessagingGateway.Kafka.SaslMechanism;
-using SecurityProtocol = Paramore.Brighter.MessagingGateway.Kafka.SecurityProtocol;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Confluent.Proactor;
 
 [Trait("Category", "Kafka")]
 [Trait("Category", "Confluent")]

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_consumer_declares_topic_on_a_confluent_cluster_async.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Confluent/Proactor/When_consumer_declares_topic_on_a_confluent_cluster_async.cs

@@ -9,7 +9,7 @@
 using SaslMechanism = Paramore.Brighter.MessagingGateway.Kafka.SaslMechanism;
 using SecurityProtocol = Paramore.Brighter.MessagingGateway.Kafka.SecurityProtocol;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Confluent.Proactor;
 
 [Trait("Category", "Kafka")]
 [Trait("Category", "Confluent")]

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_posting_a_message_to_a_confluent_cluster_async.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Confluent/Proactor/When_posting_a_message_to_a_confluent_cluster_async.cs

@@ -9,7 +9,7 @@
 using SaslMechanism = Paramore.Brighter.MessagingGateway.Kafka.SaslMechanism;
 using SecurityProtocol = Paramore.Brighter.MessagingGateway.Kafka.SecurityProtocol;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Confluent.Proactor;
 
 [Trait("Category", "Kafka")]
 [Trait("Category", "Confluent")]

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_a_set_of_messages_is_sent_preserve_order_on_a_confluent_cluster.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Confluent/Reactor/When_a_set_of_messages_is_sent_preserve_order_on_a_confluent_cluster.cs

@@ -10,7 +10,7 @@
 using Xunit;
 using Xunit.Abstractions;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Confluent.Reactor;
 
 [Trait("Category", "Kafka")]
 [Trait("Category", "Confluent")]
@@ -129,14 +129,15 @@ private string SendMessage()
 
     private IEnumerable<Message> ConsumeMessages(IAmAMessageConsumerSync consumer)
     {
-        var messages = new Message[0];
+        var messages = Array.Empty<Message>();
         int maxTries = 0;
         do
         {
             try
             {
                 maxTries++;
-                Task.Delay(500).Wait(); //Let topic propagate in the broker
+                //Let topic propagate in the broker
+                Task.Delay(500).Wait(); 
                 messages = consumer.Receive(TimeSpan.FromMilliseconds(1000));
 
                 if (messages[0].Header.MessageType != MessageType.MT_NONE)

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_consumer_assumes_topic_but_missing_on_a_confluent_cluster.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Confluent/Reactor/When_consumer_assumes_topic_but_missing_on_a_confluent_cluster.cs

@@ -23,19 +23,13 @@ THE SOFTWARE. */
 #endregion
 
 using System;
-using System.Linq;
 using System.Runtime.InteropServices;
 using System.Threading.Tasks;
-using Confluent.Kafka;
 using FluentAssertions;
-using Paramore.Brighter.Kafka.Tests.TestDoubles;
 using Paramore.Brighter.MessagingGateway.Kafka;
 using Xunit;
-using Xunit.Abstractions;
-using SaslMechanism = Paramore.Brighter.MessagingGateway.Kafka.SaslMechanism;
-using SecurityProtocol = Paramore.Brighter.MessagingGateway.Kafka.SecurityProtocol;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Confluent.Reactor;
 
 [Trait("Category", "Kafka")]
 [Trait("Category", "Confluent")]

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_consumer_declares_topic_on_a_confluent_cluster.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Confluent/Reactor/When_consumer_declares_topic_on_a_confluent_cluster.cs

@@ -9,7 +9,7 @@
 using SaslMechanism = Paramore.Brighter.MessagingGateway.Kafka.SaslMechanism;
 using SecurityProtocol = Paramore.Brighter.MessagingGateway.Kafka.SecurityProtocol;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Confluent.Reactor;
 
 [Trait("Category", "Kafka")]
 [Trait("Category", "Confluent")]

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_posting_a_message_to_a_confluent_cluster.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Confluent/Reactor/When_posting_a_message_to_a_confluent_cluster.cs

@@ -9,7 +9,7 @@
 using SaslMechanism = Paramore.Brighter.MessagingGateway.Kafka.SaslMechanism;
 using SecurityProtocol = Paramore.Brighter.MessagingGateway.Kafka.SecurityProtocol;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Confluent.Reactor;
 
 [Trait("Category", "Kafka")]
 [Trait("Category", "Confluent")]

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_a_message_is_acknowledged_update_offset_async.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Local/Proactor/When_a_message_is_acknowledged_update_offset_async.cs

@@ -8,7 +8,7 @@
 using Xunit;
 using Xunit.Abstractions;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Local.Proactor;
 
 [Trait("Category", "Kafka")]
 [Trait("Fragile", "CI")]
@@ -68,7 +68,7 @@ public async Task When_a_message_is_acknowldgede_update_offset()
         }
 
         //yield to broker to catch up
-        await Task.Delay(TimeSpan.FromSeconds(5));
+        await Task.Delay(TimeSpan.FromMilliseconds(500));
 
         //This will create a new consumer
         Message[] newMessages = await ConsumeMessagesAsync(groupId, batchLimit: 5);
@@ -107,15 +107,17 @@ private async Task<Message[]> ConsumeMessagesAsync(string groupId, int batchLimi
 
         async Task<Message> ConsumeMessageAsync(IAmAMessageConsumerAsync consumer)
         {
-            Message[] messages = new []{new Message()};
+            Message[] messages = [new Message()];
             int maxTries = 0;
             do
             {
                 try
                 {
                     maxTries++;
-                    await Task.Delay(500); //Let topic propagate in the broker
-                    messages = await consumer.ReceiveAsync(TimeSpan.FromMilliseconds(1000));
+                    //Let topic propagate in the broker
+                    await Task.Delay(500); 
+                    //use TimeSpan.Zero to avoid blocking
+                    messages = await consumer.ReceiveAsync(TimeSpan.Zero);
 
                     if (messages[0].Header.MessageType != MessageType.MT_NONE)
                     {

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_a_set_of_messages_is_sent_preserve_order_async.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Local/Proactor/When_a_set_of_messages_is_sent_preserve_order_async.cs

@@ -9,7 +9,7 @@
 using Xunit;
 using Xunit.Abstractions;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Local.Proactor;
 
 [Trait("Category", "Kafka")]
 [Collection("Kafka")]   //Kafka doesn't like multiple consumers of a partition
@@ -111,15 +111,17 @@ private async Task<string> SendMessageAsync()
 
     private async Task<IEnumerable<Message>> ConsumeMessagesAsync(IAmAMessageConsumerAsync consumer)
     {
-        var messages = new Message[0];
+        var messages = Array.Empty<Message>();
         int maxTries = 0;
         do
         {
             try
             {
                 maxTries++;
-                await Task.Delay(500); //Let topic propagate in the broker
-                messages = await consumer.ReceiveAsync(TimeSpan.FromMilliseconds(1000));
+                //Let topic propagate in the broker
+                await Task.Delay(500); 
+                //use TimeSpan.Zero to avoid blocking
+                messages = await consumer.ReceiveAsync(TimeSpan.Zero);
 
                 if (messages[0].Header.MessageType != MessageType.MT_NONE)
                     break;

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_consumer_assumes_topic_but_missing_async.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Local/Proactor/When_consumer_assumes_topic_but_missing_async.cs

@@ -5,7 +5,7 @@
 using Xunit;
 using Xunit.Abstractions;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Local.Proactor;
 
 [Trait("Category", "Kafka")]
 [Collection("Kafka")]   //Kafka doesn't like multiple consumers of a partition
@@ -26,7 +26,8 @@ public KafkaProducerAssumeTestsAsync(ITestOutputHelper output)
                 Name = "Kafka Producer Send Test",
                 BootStrapServers = new[] {"localhost:9092"}
             },
-            new KafkaPublication[] {new KafkaPublication
+            [
+                new KafkaPublication
             {
                 Topic = new RoutingKey(_topic),
                 NumPartitions = 1,
@@ -36,7 +37,8 @@ public KafkaProducerAssumeTestsAsync(ITestOutputHelper output)
                 MessageTimeoutMs = 2000,
                 RequestTimeoutMs = 2000,
                 MakeChannels = OnMissingChannel.Create
-            }}).Create();
+            }
+            ]).Create();
     }
 
     //[Fact(Skip = "Does not fail on docker container as has topic creation set to true")]

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_consumer_declares_topic_async.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Local/Proactor/When_consumer_declares_topic_async.cs

@@ -6,7 +6,7 @@
 using Xunit;
 using Xunit.Abstractions;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Local.Proactor;
 
 [Trait("Category", "Kafka")]
 [Collection("Kafka")]   //Kafka doesn't like multiple consumers of a partition
@@ -82,8 +82,10 @@ public async Task When_a_consumer_declares_topics()
             try
             {
                 maxTries++;
-                await Task.Delay(500); //Let topic propagate in the broker
-                messages = await _consumer.ReceiveAsync(TimeSpan.FromMilliseconds(10000));
+                //Let topic propagate in the broker
+                await Task.Delay(500); 
+                //use TimeSpan.Zero to avoid blocking
+                messages = await _consumer.ReceiveAsync(TimeSpan.Zero);
                 await _consumer.AcknowledgeAsync(messages[0]);
 
                 if (messages[0].Header.MessageType != MessageType.MT_NONE)

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_offsets_awaiting_next_acknowledge_sweep_them_async.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Local/Proactor/When_offsets_awaiting_next_acknowledge_sweep_them_async.cs

@@ -7,7 +7,7 @@
 using Xunit;
 using Xunit.Abstractions;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Local.Proactor;
 
 [Trait("Category", "Kafka")]
 [Collection("Kafka")]   //Kafka doesn't like multiple consumers of a partition

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_posting_a_message_async.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Local/Proactor/When_posting_a_message_async.cs

@@ -8,7 +8,7 @@
 using Xunit;
 using Xunit.Abstractions;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Local.Proactor;
 
 [Trait("Category", "Kafka")]
 [Collection("Kafka")] //Kafka doesn't like multiple consumers of a partition
@@ -96,8 +96,8 @@ public async Task When_posting_a_message()
         receivedMessage.Header.PartitionKey.Should().Be(_partitionKey);
         receivedMessage.Body.Bytes.Should().Equal(message.Body.Bytes);
         receivedMessage.Body.Value.Should().Be(message.Body.Value);
-        receivedMessage.Header.TimeStamp.ToString("yyyy-MM-ddTHH:mm:ssZ")
-            .Should().Be(message.Header.TimeStamp.ToString("yyyy-MM-ddTHH:mm:ssZ"));
+        receivedMessage.Header.TimeStamp.ToString("yyyy-MM-ddTHH:mm:Z")
+            .Should().Be(message.Header.TimeStamp.ToString("yyyy-MM-ddTHH:mm:Z"));
         receivedCommand.Id.Should().Be(command.Id);
         receivedCommand.Value.Should().Be(command.Value);
     }
@@ -111,8 +111,10 @@ private async Task<Message> GetMessageAsync()
             try
             {
                 maxTries++;
-                await Task.Delay(500); //Let topic propagate in the broker
-                messages = await _consumer.ReceiveAsync(TimeSpan.FromMilliseconds(1000));
+                //Let topic propagate in the broker
+                await Task.Delay(500); 
+                //set timespan to zero so that we will not block
+                messages = await _consumer.ReceiveAsync(TimeSpan.Zero);
 
                 if (messages[0].Header.MessageType != MessageType.MT_NONE)
                 {

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_posting_a_message_with_header_bytes_async.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Local/Proactor/When_posting_a_message_with_header_bytes_async.cs

@@ -11,7 +11,7 @@
 using Xunit;
 using Xunit.Abstractions;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Local.Proactor;
 
 [Trait("Category", "Kafka")]
 [Collection("Kafka")]   //Kafka doesn't like multiple consumers of a partition

tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/When_recieving_a_message_without_partition_key_header_async.cs → tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/Local/Proactor/When_recieving_a_message_without_partition_key_header_async.cs

@@ -10,7 +10,7 @@
 using Xunit.Abstractions;
 using Acks = Confluent.Kafka.Acks;
 
-namespace Paramore.Brighter.Kafka.Tests.MessagingGateway;
+namespace Paramore.Brighter.Kafka.Tests.MessagingGateway.Local.Proactor;
 
 public class KafkaMessageProducerMissingHeaderTestsAsync : IAsyncDisposable
 {
@@ -40,7 +40,7 @@ public KafkaMessageProducerMissingHeaderTestsAsync(ITestOutputHelper output)
             LingerMs = 5,
             MessageTimeoutMs = 5000,
             MessageSendMaxRetries = 3,
-            Partitioner = Confluent.Kafka.Partitioner.ConsistentRandom,
+            Partitioner = global::Confluent.Kafka.Partitioner.ConsistentRandom,
             QueueBufferingMaxMessages = 10,
             QueueBufferingMaxKbytes = 1048576,
             RequestTimeoutMs = 500,