diff --git a/src/Dafda.Tests/Producing/TestOutgoingMessageFactory.cs b/src/Dafda.Tests/Producing/TestOutgoingMessageFactory.cs
index 534214df..5966680d 100644
--- a/src/Dafda.Tests/Producing/TestOutgoingMessageFactory.cs
+++ b/src/Dafda.Tests/Producing/TestOutgoingMessageFactory.cs
@@ -7,11 +7,18 @@
 namespace Dafda.Tests.Producing
 {
     [Collection("Serializing")]
-    public class TestOutgoingMessageFactory
+    public class TestOutgoingMessageFactory : IClassFixture<DafdaActivitySourceFixture>
     {
+        private readonly DafdaActivitySourceFixture _fixture;
         private const string DummyTopic = "dummy_topic";
         private const string DummyType = "dummy_type";
 
+        public TestOutgoingMessageFactory(DafdaActivitySourceFixture fixture)
+        {
+            _fixture = fixture;
+            _fixture.ResetDafdaActivitySource();
+        }
+
         [Fact]
         public async Task Can_create_outgoing_message_from_registry_with_expected_raw_message()
         {
diff --git a/src/Dafda.Tests/Producing/TestProducer.cs b/src/Dafda.Tests/Producing/TestProducer.cs
index e6abbe55..2e90ffa0 100644
--- a/src/Dafda.Tests/Producing/TestProducer.cs
+++ b/src/Dafda.Tests/Producing/TestProducer.cs
@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.Linq;
 using System.Threading.Tasks;
 using Dafda.Consuming;
 using Dafda.Diagnostics;
@@ -17,8 +18,16 @@
 namespace Dafda.Tests.Producing
 {
     [Collection("Serializing")]
-    public class TestProducer
+    public class TestProducer : IClassFixture<DafdaActivitySourceFixture>
     {
+        private readonly DafdaActivitySourceFixture _fixture;
+
+        public TestProducer(DafdaActivitySourceFixture fixture)
+        {
+            _fixture = fixture;
+            _fixture.ResetDafdaActivitySource();
+        }
+
         [Fact]
         public async Task Can_produce_message()
         {
@@ -295,5 +304,425 @@ await sut.Produce(
 
             AssertJson.Equal(expected, spy.Value);
         }
+
+        [Fact]
+        public async Task Can_produce_multiple_messages()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var sut = A.Producer
+                .With(spy)
+                .With(A.OutgoingMessageRegistry
+                    .Register<Message>("foo", "bar", @event => @event.Id)
+                    .Build()
+                )
+                .Build();
+
+            var messages = new[]
+            {
+                new Message { Id = "id1" },
+                new Message { Id = "id2" },
+                new Message { Id = "id3" }
+            };
+
+            await sut.Produce(messages);
+
+            Assert.Equal(3, spy.ProduceCallCount);
+            Assert.Equal("foo", spy.Topic);
+        }
+
+        [Fact]
+        public async Task produces_multiple_messages_to_expected_topic()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var expectedTopic = "foo";
+
+            var sut = A.Producer
+                .With(spy)
+                .With(A.OutgoingMessageRegistry
+                    .Register<Message>(expectedTopic, "bar", @event => @event.Id)
+                    .Build()
+                )
+                .Build();
+
+            var messages = new[]
+            {
+                new Message { Id = "id1" },
+                new Message { Id = "id2" }
+            };
+
+            await sut.Produce(
+                messages: messages,
+                headers: new Dictionary<string, object>
+                {
+                    { "foo-key", "foo-value" }
+                }
+            );
+
+            Assert.Equal(expectedTopic, spy.Topic);
+            Assert.Equal(2, spy.ProduceCallCount);
+        }
+
+        [Fact]
+        public async Task produces_multiple_messages_with_expected_partition_keys()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var sut = A.Producer
+                .With(spy)
+                .With(A.OutgoingMessageRegistry
+                    .Register<Message>("foo", "bar", @event => @event.Id)
+                    .Build()
+                )
+                .Build();
+
+            var messages = new[]
+            {
+                new Message { Id = "key1" },
+                new Message { Id = "key2" },
+                new Message { Id = "key3" }
+            };
+
+            await sut.Produce(
+                messages: messages,
+                headers: new Dictionary<string, object>
+                {
+                    { "foo-key", "foo-value" }
+                }
+            );
+
+            Assert.Equal(3, spy.ProduceCallCount);
+            Assert.Contains("key1", spy.AllKeys);
+            Assert.Contains("key2", spy.AllKeys);
+            Assert.Contains("key3", spy.AllKeys);
+        }
+
+        [Fact]
+        public async Task produces_multiple_messages_with_expected_serialized_values()
+        {
+            var expectedValue = "foo-value";
+
+            var spy = new KafkaProducerSpy(new PayloadSerializerStub(expectedValue));
+
+            var sut = A.Producer
+                .With(spy)
+                .With(A.OutgoingMessageRegistry
+                    .Register<Message>("foo", "bar", @event => @event.Id)
+                    .Build()
+                )
+                .Build();
+
+            var messages = new[]
+            {
+                new Message { Id = "id1" },
+                new Message { Id = "id2" }
+            };
+
+            await sut.Produce(
+                messages: messages,
+                headers: new Dictionary<string, object>
+                {
+                    { "foo-key", "foo-value" }
+                }
+            );
+
+            Assert.Equal(expectedValue, spy.Value);
+            Assert.Equal(2, spy.ProduceCallCount);
+        }
+
+        [Fact]
+        public async Task produces_expected_multiple_messages_without_headers_using_default_serializer()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var sut = A.Producer
+                .With(spy)
+                .With(new MessageIdGeneratorStub(() => "1"))
+                .With(A.OutgoingMessageRegistry
+                    .Register<Message>("foo", "bar", @event => @event.Id)
+                    .Build()
+                )
+                .Build();
+
+            var messages = new[]
+            {
+                new Message { Id = "dummyId1" },
+                new Message { Id = "dummyId2" }
+            };
+
+            await sut.Produce(
+                messages: messages,
+                headers: []
+            );
+
+            Assert.Equal(2, spy.ProduceCallCount);
+            var expected = @"{
+                                ""messageId"":""1"",
+                                ""type"":""bar"",
+                                ""data"":
+                                {
+                                    ""id"":""dummyId2""
+                                    }
+                                }";
+
+            AssertJson.Equal(expected, spy.Value);
+        }
+
+        [Fact]
+        public async Task produces_expected_multiple_messages_with_headers_using_default_serializer()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var sut = A.Producer
+                .With(spy)
+                .With(new MessageIdGeneratorStub(() => "1"))
+                .With(A.OutgoingMessageRegistry
+                    .Register<Message>("foo", "bar", @event => @event.Id)
+                    .Build()
+                )
+                .Build();
+
+            var messages = new[]
+            {
+                new Message { Id = "dummyId1" },
+                new Message { Id = "dummyId2" }
+            };
+
+            await sut.Produce(
+                messages: messages,
+                headers: new Dictionary<string, object>
+                {
+                    { "foo-key", "foo-value" }
+                }
+            );
+
+            Assert.Equal(2, spy.ProduceCallCount);
+            var expected = @"{
+                                ""messageId"":""1"",
+                                ""type"":""bar"",
+                                ""foo-key"":""foo-value"",
+                                ""data"":
+                                {
+                                    ""id"":""dummyId2""
+                                    }
+                                }";
+
+            AssertJson.Equal(expected, spy.Value);
+        }
+
+        [Fact]
+        public async Task produces_multiple_messages_using_metadata()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var sut = A.Producer
+                .With(spy)
+                .With(A.OutgoingMessageRegistry
+                    .Register<Message>("foo", "bar", @event => @event.Id)
+                    .Build()
+                )
+                .Build();
+
+            var messages = new[]
+            {
+                new Message { Id = "key1" },
+                new Message { Id = "key2" }
+            };
+
+            await sut.Produce(
+                messages: messages,
+                headers: new Metadata()
+            );
+
+            Assert.Equal(2, spy.ProduceCallCount);
+            Assert.Contains("key1", spy.AllKeys);
+            Assert.Contains("key2", spy.AllKeys);
+        }
+
+        [Fact]
+        public async Task produces_multiple_messages_using_message_handler_context()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var sut = A.Producer
+                .With(spy)
+                .With(new MessageIdGeneratorStub(() => "1"))
+                .With(A.OutgoingMessageRegistry
+                    .Register<Message>("foo", "bar", @event => @event.Id)
+                    .Build()
+                )
+                .Build();
+
+            var messages = new[]
+            {
+                new Message { Id = "0" },
+                new Message { Id = "1" }
+            };
+
+            await sut.Produce(
+                messages: messages,
+                context: new MessageHandlerContext(new Metadata())
+            );
+
+            Assert.Equal(2, spy.ProduceCallCount);
+            var expected = @"
+                            {
+                            ""messageId"":""1"",
+                            ""type"":""bar"",
+                            ""data"":
+                            {
+                                ""id"":""1""
+                                }
+                            }";
+
+            AssertJson.Equal(expected, spy.Value);
+        }
+
+        [Fact]
+        public async Task produces_multiple_messages_using_message_handler_context_with_additional_headers()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var sut = A.Producer
+                .With(spy)
+                .With(new MessageIdGeneratorStub(() => "1"))
+                .With(A.OutgoingMessageRegistry
+                    .Register<Message>("foo", "bar", @event => @event.Id)
+                    .Build()
+                )
+                .Build();
+
+            var messages = new[]
+            {
+                new Message { Id = "0" },
+                new Message { Id = "1" }
+            };
+
+            await sut.Produce(
+                messages: messages,
+                context: new MessageHandlerContext(new Metadata()),
+                headers: new Dictionary<string, string>
+                {
+                    { "additional-key", "additional-value" }
+                }
+            );
+
+            Assert.Equal(2, spy.ProduceCallCount);
+            var expected = @"
+                            {
+                            ""messageId"":""1"",
+                            ""type"":""bar"",
+                            ""additional-key"":""additional-value"",
+                            ""data"":
+                            {
+                                ""id"":""1""
+                                }
+                            }";
+
+            AssertJson.Equal(expected, spy.Value);
+        }
+
+        [Fact]
+        public async Task produces_multiple_messages_preserves_order()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var sut = A.Producer
+                .With(spy)
+                .With(A.OutgoingMessageRegistry
+                    .Register<Message>("foo", "bar", @event => @event.Id)
+                    .Build()
+                )
+                .Build();
+
+            var messages = new[]
+            {
+                new Message { Id = "first" },
+                new Message { Id = "second" },
+                new Message { Id = "third" },
+                new Message { Id = "fourth" },
+                new Message { Id = "fifth" }
+            };
+
+            await sut.Produce(messages);
+
+            Assert.Equal(5, spy.ProduceCallCount);
+            Assert.Equal("first", spy.ProducedMessages[0].Key);
+            Assert.Equal("second", spy.ProducedMessages[1].Key);
+            Assert.Equal("third", spy.ProducedMessages[2].Key);
+            Assert.Equal("fourth", spy.ProducedMessages[3].Key);
+            Assert.Equal("fifth", spy.ProducedMessages[4].Key);
+        }
+
+        [Fact]
+        public async Task produces_multiple_messages_with_headers_preserves_order()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var sut = A.Producer
+                .With(spy)
+                .With(A.OutgoingMessageRegistry
+                    .Register<Message>("foo", "bar", @event => @event.Id)
+                    .Build()
+                )
+                .Build();
+
+            var messages = new[]
+            {
+                new Message { Id = "alpha" },
+                new Message { Id = "beta" },
+                new Message { Id = "gamma" }
+            };
+
+            await sut.Produce(
+                messages: messages,
+                headers: new Dictionary<string, object>
+                {
+                    { "foo-key", "foo-value" }
+                }
+            );
+
+            Assert.Equal(3, spy.ProduceCallCount);
+            Assert.Equal("alpha", spy.ProducedMessages[0].Key);
+            Assert.Equal("beta", spy.ProducedMessages[1].Key);
+            Assert.Equal("gamma", spy.ProducedMessages[2].Key);
+            Assert.Collection(spy.ProducedMessages,
+                msg => Assert.Equal(1, msg.Order),
+                msg => Assert.Equal(2, msg.Order),
+                msg => Assert.Equal(3, msg.Order)
+            );
+        }
+
+        [Fact]
+        public async Task produces_multiple_messages_with_context_preserves_order()
+        {
+            var spy = new KafkaProducerSpy();
+
+            var sut = A.Producer
+                .With(spy)
+                .With(A.OutgoingMessageRegistry
+                    .Register<Message>("foo", "bar", @event => @event.Id)
+                    .Build()
+                )
+                .Build();
+
+            var messages = new[]
+            {
+                new Message { Id = "msg1" },
+                new Message { Id = "msg2" },
+                new Message { Id = "msg3" },
+                new Message { Id = "msg4" }
+            };
+
+            await sut.Produce(
+                messages: messages,
+                context: new MessageHandlerContext(new Metadata())
+            );
+
+            Assert.Equal(4, spy.ProduceCallCount);
+            var orderedKeys = spy.ProducedMessages.OrderBy(m => m.Order).Select(m => m.Key).ToArray();
+            Assert.Equal(new[] { "msg1", "msg2", "msg3", "msg4" }, orderedKeys);
+        }
     }
 }
\ No newline at end of file
diff --git a/src/Dafda.Tests/TestDoubles/KafkaProducerSpy.cs b/src/Dafda.Tests/TestDoubles/KafkaProducerSpy.cs
index 633ebe1e..1e636fba 100644
--- a/src/Dafda.Tests/TestDoubles/KafkaProducerSpy.cs
+++ b/src/Dafda.Tests/TestDoubles/KafkaProducerSpy.cs
@@ -2,6 +2,7 @@
 using System.Diagnostics;
 using System.Linq;
 using System.Threading.Tasks;
+using Confluent.Kafka;
 using Dafda.Producing;
 using Dafda.Serializing;
 using Microsoft.Extensions.Logging.Abstractions;
@@ -10,6 +11,9 @@ namespace Dafda.Tests.TestDoubles
 {
     internal class KafkaProducerSpy : KafkaProducer
     {
+        private readonly List<string> _keys = new List<string>();
+        private readonly List<ProducedMessage> _producedMessages = new List<ProducedMessage>();
+
         public KafkaProducerSpy()
             : this(Enumerable.Empty<KeyValuePair<string, string>>(), new DefaultPayloadSerializer())
         {
@@ -25,14 +29,35 @@ private KafkaProducerSpy(IEnumerable<KeyValuePair<string, string>> configuration
         {
         }
 
-        internal override Task InternalProduce(string topic, string key, string value)
+        internal override Task<DeliveryResult<string, string>> InternalProduce(string topic, string key, string value)
         {
             Topic = topic;
             Key = key;
             Value = value;
             ProducerActivityId = Activity.Current?.Id;
+            ProduceCallCount++;
+            _keys.Add(key);
+            _producedMessages.Add(new ProducedMessage
+            {
+                Topic = topic,
+                Key = key,
+                Value = value,
+                Order = ProduceCallCount
+            });
+
+            var deliveryResult = new DeliveryResult<string, string>
+            {
+                Topic = topic,
+                Partition = new Partition(0),
+                Offset = new Offset(ProduceCallCount - 1),
+                Message = new Message<string, string>
+                {
+                    Key = key,
+                    Value = value
+                }
+            };
 
-            return Task.CompletedTask;
+            return Task.FromResult(deliveryResult);
         }
 
         public override void Dispose()
@@ -47,5 +72,16 @@ public override void Dispose()
         public string Key { get; private set; }
         public string Value { get; private set; }
         public string ProducerActivityId { get; private set; }
+        public int ProduceCallCount { get; private set; }
+        public IReadOnlyList<string> AllKeys => _keys;
+        public IReadOnlyList<ProducedMessage> ProducedMessages => _producedMessages;
+
+        public class ProducedMessage
+        {
+            public string Topic { get; set; }
+            public string Key { get; set; }
+            public string Value { get; set; }
+            public int Order { get; set; }
+        }
     }
 }
\ No newline at end of file
diff --git a/src/Dafda/Producing/KafkaProducer.cs b/src/Dafda/Producing/KafkaProducer.cs
index ef9476a1..a81c3563 100644
--- a/src/Dafda/Producing/KafkaProducer.cs
+++ b/src/Dafda/Producing/KafkaProducer.cs
@@ -22,24 +22,24 @@ public KafkaProducer(ILoggerFactory loggerFactory, IEnumerable<KeyValuePair<stri
 
         public string ClientId => _innerKafkaProducer.Name;
 
-        public async Task Produce(PayloadDescriptor payloadDescriptor)
+        public async Task<DeliveryResult<string, string>> Produce(PayloadDescriptor payloadDescriptor)
         {
             var serializer = _payloadSerializerRegistry.Get(payloadDescriptor.TopicName);
 
-            await InternalProduce(
+            return await InternalProduce(
                 topic: payloadDescriptor.TopicName,
                 key: payloadDescriptor.PartitionKey,
                 value: await serializer.Serialize(payloadDescriptor)
             );
         }
 
-        internal virtual async Task InternalProduce(string topic, string key, string value)
+        internal virtual async Task<DeliveryResult<string, string>> InternalProduce(string topic, string key, string value)
         {
             try
             {
                 _logger.LogDebug("Producing message with {Key} on {Topic}", key, topic);
 
-                await _innerKafkaProducer.ProduceAsync(
+                return await _innerKafkaProducer.ProduceAsync(
                     topic: topic,
                     message: new Message<string, string>
                     {
diff --git a/src/Dafda/Producing/Producer.cs b/src/Dafda/Producing/Producer.cs
index 44059544..add88cbf 100644
--- a/src/Dafda/Producing/Producer.cs
+++ b/src/Dafda/Producing/Producer.cs
@@ -1,4 +1,5 @@
-using System.Threading.Tasks;
+using System;
+using System.Threading.Tasks;
 using System.Collections.Generic;
 using System.Linq;
 using Dafda.Consuming;
@@ -22,13 +23,94 @@ internal Producer(KafkaProducer kafkaProducer, OutgoingMessageRegistry outgoingM
 
         internal string Name { get; set; } = "__Default Producer__";
 
+        /// <summary>
+        /// Produce multiple <paramref name="messages"/> on Kafka. The messages will be produced in the order of the IEnumerable provided.
+        /// </summary>
+        /// <param name="messages">The messages</param>
+        /// <param name="onDelivery">Callback invoked for each message after delivery</param>
+        public async Task Produce(IEnumerable<object> messages, Action<Confluent.Kafka.DeliveryResult<string, string>> onDelivery = null)
+        {
+            var produceTasks = messages.Select(async message =>
+            {
+                var result = await Produce(message);
+                onDelivery?.Invoke(result);
+            });
+            await Task.WhenAll(produceTasks);
+        }
+
+        /// <summary>
+        /// Produce multiple <paramref name="messages"/> on Kafka including <paramref name="headers"/>. The messages will be produced in the order of the IEnumerable provided.
+        /// </summary>
+        /// <param name="messages">The messages</param>
+        /// <param name="headers">The message headers</param>
+        /// <param name="onDelivery">Callback invoked for each message after delivery</param>
+        public async Task Produce(IEnumerable<object> messages, Metadata headers, Action<Confluent.Kafka.DeliveryResult<string, string>> onDelivery = null)
+        {
+            var produceTasks = messages.Select(async message =>
+            {
+                var result = await Produce(message, headers);
+                onDelivery?.Invoke(result);
+            });
+            await Task.WhenAll(produceTasks);
+        }
+
+        /// <summary>
+        /// Produce multiple <paramref name="messages"/> on Kafka including <paramref name="headers"/>. The messages will be produced in the order of the IEnumerable provided.
+        /// </summary>
+        /// <param name="messages">The messages</param>
+        /// <param name="headers">The message headers</param>
+        /// <param name="onDelivery">Callback invoked for each message after delivery</param>
+        public async Task Produce(IEnumerable<object> messages, Dictionary<string, object> headers, Action<Confluent.Kafka.DeliveryResult<string, string>> onDelivery = null)
+        {
+            var produceTasks = messages.Select(async message =>
+            {
+                var result = await Produce(message, headers);
+                onDelivery?.Invoke(result);
+            });
+            await Task.WhenAll(produceTasks);
+        }
+
+        /// <summary>
+        /// Produce multiple <paramref name="messages"/> on Kafka. The messages will be produced in the order of the IEnumerable provided.
+        /// </summary>
+        /// <param name="messages">The messages</param>
+        /// <param name="context">Context from the consumer. Supply this to get correlation and causation id on the new messages</param>
+        /// <param name="onDelivery">Callback invoked for each message after delivery</param>
+        public async Task Produce(IEnumerable<object> messages, MessageHandlerContext context, Action<Confluent.Kafka.DeliveryResult<string, string>> onDelivery = null)
+        {
+            var produceTasks = messages.Select(async message =>
+            {
+                var result = await Produce(message, context);
+                onDelivery?.Invoke(result);
+            });
+            await Task.WhenAll(produceTasks);
+        }
+
+        /// <summary>
+        /// Produce multiple <paramref name="messages"/> on Kafka. The messages will be produced in the order of the IEnumerable provided.
+        /// </summary>
+        /// <param name="messages">The messages</param>
+        /// <param name="context">Context from the consumer. Supply this to get correlation and causation id on the new messages</param>
+        /// <param name="headers">Additional message headers</param>
+        /// <param name="onDelivery">Callback invoked for each message after delivery</param>
+        public async Task Produce(IEnumerable<object> messages, MessageHandlerContext context, Dictionary<string, string> headers, Action<Confluent.Kafka.DeliveryResult<string, string>> onDelivery = null)
+        {
+            var produceTasks = messages.Select(async message =>
+            {
+                var result = await Produce(message, context, headers);
+                onDelivery?.Invoke(result);
+            });
+            await Task.WhenAll(produceTasks);
+        }
+
         /// <summary>
         /// Produce a <paramref name="message"/> on Kafka
         /// </summary>
         /// <param name="message">The message</param>
-        public async Task Produce(object message)
+        /// <returns>The delivery result from Kafka</returns>
+        public async Task<Confluent.Kafka.DeliveryResult<string, string>> Produce(object message)
         {
-            await Produce(message, new Metadata());
+            return await Produce(message, new Metadata());
         }
 
         /// <summary>
@@ -36,13 +118,14 @@ public async Task Produce(object message)
         /// </summary>
         /// <param name="message">The message</param>
         /// <param name="headers">The message headers</param>
-        public async Task Produce(object message, Metadata headers)
+        /// <returns>The delivery result from Kafka</returns>
+        public async Task<Confluent.Kafka.DeliveryResult<string, string>> Produce(object message, Metadata headers)
         {
             var payloadDescriptor = _payloadDescriptorFactory.Create(message, headers);
             payloadDescriptor.ClientId = _kafkaProducer.ClientId;
             using var activity = DafdaActivitySource.StartPublishingActivity(payloadDescriptor);
 
-            await _kafkaProducer.Produce(payloadDescriptor);
+            return await _kafkaProducer.Produce(payloadDescriptor);
         }
 
         /// <summary>
@@ -50,10 +133,11 @@ public async Task Produce(object message, Metadata headers)
         /// </summary>
         /// <param name="message">The message</param>
         /// <param name="headers">The message headers</param>
-        public async Task Produce(object message, Dictionary<string, object> headers)
+        /// <returns>The delivery result from Kafka</returns>
+        public async Task<Confluent.Kafka.DeliveryResult<string, string>> Produce(object message, Dictionary<string, object> headers)
         {
             var dict = headers.ToDictionary( pair => pair.Key, pair => pair.Value.ToString());
-            await Produce(message, new Metadata( dict ));
+            return await Produce(message, new Metadata( dict ));
         }
 
         /// <summary>
@@ -61,9 +145,10 @@ public async Task Produce(object message, Dictionary<string, object> headers)
         /// </summary>
         /// <param name="message">The message</param>
         /// <param name="context">Context from the consumer. Supply this to get correlation and causation id on the new message</param>
-        public async Task Produce(object message, MessageHandlerContext context)
+        /// <returns>The delivery result from Kafka</returns>
+        public async Task<Confluent.Kafka.DeliveryResult<string, string>> Produce(object message, MessageHandlerContext context)
         {
-            await Produce(message, context, new Dictionary<string, string>());
+            return await Produce(message, context, new Dictionary<string, string>());
         }
 
         /// <summary>
@@ -72,13 +157,14 @@ public async Task Produce(object message, MessageHandlerContext context)
         /// <param name="message">The message</param>
         /// <param name="context">Context from the consumer. Supply this to get correlation and causation id on the new message</param>
         /// <param name="headers">Additional message headers</param>
-        public async Task Produce(object message, MessageHandlerContext context, Dictionary<string, string> headers)
+        /// <returns>The delivery result from Kafka</returns>
+        public async Task<Confluent.Kafka.DeliveryResult<string, string>> Produce(object message, MessageHandlerContext context, Dictionary<string, string> headers)
         {
             var payloadDescriptor = _payloadDescriptorFactory.Create(message, context, headers);
             payloadDescriptor.ClientId = _kafkaProducer.ClientId;
             using var activity = DafdaActivitySource.StartPublishingActivity(payloadDescriptor);
 
-            await _kafkaProducer.Produce(payloadDescriptor);
+            return await _kafkaProducer.Produce(payloadDescriptor);
         }
     }
 }
\ No newline at end of file