Make sure that we prevent downstream to propagate null completion exception (#427)

Author: Arkatufus

src/Akka.Streams.Kafka/Stages/DownstreamFinishedWithNoCauseException.cs

@@ -0,0 +1,8 @@
+using System;
+
+namespace Akka.Streams.Kafka.Stages;
+
+public class DownstreamFinishedWithNoCauseException: Exception
+{
+    public override string Message => "Downstream flow or stage completed with no cause";
+}

src/Akka.Streams.Kafka/Stages/Producers/DefaultProducerStage.cs

@@ -153,10 +153,30 @@ public DefaultProducerStageLogic(IProducerStage<K, V, P, TIn, TOut> stage, Attri
                     CheckForCompletion();
                 });
 
-            SetHandler(_stage.Out, onPull: () =>
-            {
-                TryPull(_stage.In);
-            });
+            SetHandler(
+                outlet: _stage.Out, 
+                onPull: () =>
+                {
+                    TryPull(_stage.In);
+                }, 
+                onDownstreamFinish: ex =>
+                {
+                    if (ex is null)
+                    {
+                        try
+                        {
+                            throw new DownstreamFinishedWithNoCauseException();
+                        }
+                        catch(DownstreamFinishedWithNoCauseException e)
+                        {
+                            InternalOnDownstreamFinish(e);
+                        }
+                    }
+                    else
+                    {
+                        InternalOnDownstreamFinish(ex);
+                    }
+                });
         }
 
         public override void PreStart()