Resolve Kafka Unit Testing Issues

src/Foundatio.Kafka/Messaging/KafkaMessageBus.cs

@@ -112,7 +112,6 @@ private async Task OnMessageAsync(IConsumer<string, byte[]> consumer, ConsumeRes
 
             // What to do if message type is null?
             var message = ConvertToMessage(messageType, consumeResult.Message);
-
             await SendMessageToSubscribersAsync(message).AnyContext();
 
             if (!_subscribers.IsEmpty)

tests/Foundatio.Kafka.Tests/Messaging/KafkaMessageBusFanOutTests.cs

@@ -26,13 +26,13 @@ protected override IMessageBus GetMessageBus(Func<SharedMessageBusOptions, Share
         );
     }
 
-    [Fact(Skip = "https://github.com/confluentinc/confluent-kafka-dotnet/issues/1832")]
+    [Fact]
     public override Task CanReceiveFromMultipleSubscribersAsync()
     {
         return base.CanReceiveFromMultipleSubscribersAsync();
     }
 
-    [Fact(Skip = "https://github.com/confluentinc/confluent-kafka-dotnet/issues/1832")]
+    [Fact]
     public override Task CanReceiveMessagesConcurrentlyAsync()
     {
         return base.CanReceiveMessagesConcurrentlyAsync();

tests/Foundatio.Kafka.Tests/Messaging/KafkaMessageBusTests.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Threading;
 using System.Threading.Tasks;
 using Foundatio.AsyncEx;
@@ -21,7 +21,7 @@ public override Task CanUseMessageOptionsAsync()
         return base.CanUseMessageOptionsAsync();
     }
 
-    [Fact(Skip = "https://github.com/confluentinc/confluent-kafka-dotnet/issues/1832")]
+    [Fact]
     public override Task CanSendMessageAsync()
     {
         return base.CanSendMessageAsync();
@@ -33,7 +33,7 @@ public override Task CanHandleNullMessageAsync()
         return base.CanHandleNullMessageAsync();
     }
 
-    [Fact(Skip = "https://github.com/confluentinc/confluent-kafka-dotnet/issues/1832")]
+    [Fact]
     public override Task CanSendDerivedMessageAsync()
     {
         return base.CanSendDerivedMessageAsync();
@@ -45,13 +45,13 @@ public override Task CanSendDelayedMessageAsync()
         return base.CanSendDelayedMessageAsync();
     }
 
-    [Fact(Skip = "https://github.com/confluentinc/confluent-kafka-dotnet/issues/1832")]
+    [Fact]
     public override Task WillReceiveDerivedMessageTypesAsync()
     {
         return base.WillReceiveDerivedMessageTypesAsync();
     }
 
-    [Fact(Skip = "https://github.com/confluentinc/confluent-kafka-dotnet/issues/1832")]
+    [Fact]
     public override Task CanSubscribeToAllMessageTypesAsync()
     {
         return base.CanSubscribeToAllMessageTypesAsync();
@@ -63,7 +63,7 @@ public override Task CanSubscribeToRawMessagesAsync()
         return base.CanSubscribeToRawMessagesAsync();
     }
 
-    [Fact(Skip = "https://github.com/confluentinc/confluent-kafka-dotnet/issues/1832")]
+    [Fact]
     public override Task CanTolerateSubscriberFailureAsync()
     {
         return base.CanTolerateSubscriberFailureAsync();
@@ -75,13 +75,13 @@ public override Task CanSendMessageToMultipleSubscribersAsync()
         return base.CanSendMessageToMultipleSubscribersAsync();
     }
 
-    [Fact(Skip = "https://github.com/confluentinc/confluent-kafka-dotnet/issues/1832")]
+    [Fact]
     public override Task CanSubscribeConcurrentlyAsync()
     {
         return base.CanSubscribeConcurrentlyAsync();
     }
 
-    [Fact(Skip = "https://github.com/confluentinc/confluent-kafka-dotnet/issues/1832")]
+    [Fact]
     public override Task WillOnlyReceiveSubscribedMessageTypeAsync()
     {
         return base.WillOnlyReceiveSubscribedMessageTypeAsync();