Merge branch 'feat/include-open-telemetry' of https://github.com/Farf…

…etch/kafkaflow into feat/include-open-telemetry
Farfetch · Oct 17, 2023 · 5a90dce · 5a90dce
2 parents dd8759d + 9ef12a4
commit 5a90dce
Showing 1 changed file with 17 additions and 18 deletions.
diff --git a/src/KafkaFlow.IntegrationTests/GlobalEventsTest.cs b/src/KafkaFlow.IntegrationTests/GlobalEventsTest.cs
@@ -289,30 +289,29 @@ await Policy
                 .WaitAndRetryAsync(Enumerable.Range(0, 6).Select(i => TimeSpan.FromSeconds(Math.Pow(i, 2))))
                 .ExecuteAsync(() => Task.FromResult(this.isPartitionAssigned));
         }
-    }
 
-    internal class TriggerErrorMessageMiddleware : IMessageMiddleware
-    {
-        public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
+        private class TriggerErrorMessageMiddleware : IMessageMiddleware
         {
-            MessageStorage.Add((byte[])context.Message.Value);
-            throw new ErrorExecutingMiddlewareException(nameof(TriggerErrorMessageMiddleware));
-            await next(context);
+            public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
+            {
+                MessageStorage.Add((byte[])context.Message.Value);
+                throw new ErrorExecutingMiddlewareException(nameof(TriggerErrorMessageMiddleware));
+            }
         }
-    }
 
-    public class TriggerErrorSerializerMiddleware : ISerializer
-    {
-        public Task SerializeAsync(object message, Stream output, ISerializerContext context)
+        private class TriggerErrorSerializerMiddleware : ISerializer
         {
-            var error = new Error(ErrorCode.BrokerNotAvailable);
-            throw new ProduceException<byte[], byte[]>(error, null);
-        }
+            public Task SerializeAsync(object message, Stream output, ISerializerContext context)
+            {
+                var error = new Error(ErrorCode.BrokerNotAvailable);
+                throw new ProduceException<byte[], byte[]>(error, null);
+            }
 
-        public Task<object> DeserializeAsync(Stream input, Type type, ISerializerContext context)
-        {
-            var error = new Error(ErrorCode.BrokerNotAvailable);
-            throw new ProduceException<byte[], byte[]>(error, null);
+            public Task<object> DeserializeAsync(Stream input, Type type, ISerializerContext context)
+            {
+                var error = new Error(ErrorCode.BrokerNotAvailable);
+                throw new ProduceException<byte[], byte[]>(error, null);
+            }
         }
     }
 }