Added new IProducer.ProduceAsync overload for produce with minimal allocations

src/Confluent.Kafka/IProducer.cs

@@ -21,12 +21,33 @@
 
 
 namespace Confluent.Kafka
-{
+{
+    /// <summary>
+    /// Defines a high-level Apache Kafka producer client without serialization capable of producing pre-serialized messages. 
+    /// </summary>
+    public interface IProducer : IClient
+    {
+        /// <summary>
+        /// Asynchronously send a single <b>preserialized</b> message to a Kafka topic. 
+        /// </summary>
+        /// <remarks>
+        /// Use this method to produce with minimal allocations.
+        /// </remarks>
+        /// <param name="topic">The topic to produce message to.</param>
+        /// <param name="partition">The partition to produce to or <c>Partition.Any</c> to use configured partitioner.</param>
+        /// <param name="key">Serialized message key or <c>null</c>.</param>
+        /// <param name="value">Serialized message value or <c>null</c>.</param>
+        /// <param name="headers">Message headers or <c>null</c> to produce message without headers.</param>
+        /// <param name="timestamp"></param>
+        /// <returns>Result of produce.</returns>
+        Task<ProduceResult> ProduceAsync(string topic, Partition partition, ArraySegment<byte>? key, ArraySegment<byte>? value, IReadOnlyList<IHeader> headers, Timestamp timestamp);
+    }
+
     /// <summary>
     ///     Defines a high-level Apache Kafka producer client
     ///     that provides key and value serialization.
     /// </summary>
-    public interface IProducer<TKey, TValue> : IClient
+    public interface IProducer<TKey, TValue> : IProducer
     {
         /// <summary>
         ///     Asynchronously send a single message to a
@@ -99,9 +120,8 @@ Task<DeliveryResult<TKey, TValue>> ProduceAsync(
         Task<DeliveryResult<TKey, TValue>> ProduceAsync(
             TopicPartition topicPartition,
             Message<TKey, TValue> message,
-            CancellationToken cancellationToken = default(CancellationToken));
-
-
+            CancellationToken cancellationToken = default(CancellationToken));
+
         /// <summary>
         ///     Asynchronously send a single message to a
         ///     Kafka topic. The partition the message is sent
@@ -540,6 +560,6 @@ void Produce(
         /// <exception cref="Confluent.Kafka.KafkaException">
         ///     Thrown on all other errors.
         /// </exception>
-        void SendOffsetsToTransaction(IEnumerable<TopicPartitionOffset> offsets, IConsumerGroupMetadata groupMetadata, TimeSpan timeout);
+        void SendOffsetsToTransaction(IEnumerable<TopicPartitionOffset> offsets, IConsumerGroupMetadata groupMetadata, TimeSpan timeout);
     }
 }

src/Confluent.Kafka/ProduceResult.cs

@@ -0,0 +1,52 @@
+// Copyright 2016-2018 Confluent Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Refer to LICENSE for more information.
+
+namespace Confluent.Kafka
+{
+    /// <summary>
+    /// Models the result of a call to ProduceAsync.
+    /// </summary>
+	public readonly record struct ProduceResult
+    {
+        /// <summary>
+        /// The partition message was produced to.
+        /// </summary>
+        public Partition Partition { get; }
+
+		/// <summary>
+		/// The offset message was persisted at.
+		/// </summary>
+		public Offset Offset { get; }
+
+		/// <summary>
+		/// The persistence status of the message.
+		/// </summary>
+		public PersistenceStatus PersistenceStatus { get; }
+
+        /// <summary>
+        /// Creates new result.
+        /// </summary>
+        /// <param name="partition"></param>
+        /// <param name="offset"></param>
+        /// <param name="persistenceStatus"></param>
+        public ProduceResult(Partition partition, Offset offset, PersistenceStatus persistenceStatus)
+        {
+            Partition = partition;
+            Offset = offset;
+            PersistenceStatus = persistenceStatus;
+		}
+    }
+}

src/Confluent.Kafka/Producer.cs

@@ -1,19 +1,19 @@
-// Copyright 2016-2018 Confluent Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-//
-// Refer to LICENSE for more information.
-
+// Copyright 2016-2018 Confluent Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Refer to LICENSE for more information.
+
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -25,11 +25,11 @@
 
 
 namespace Confluent.Kafka
-{
+{
     /// <summary>
     ///     A high level producer with serialization capability.
     /// </summary>
-    internal class Producer<TKey, TValue> : IProducer<TKey, TValue>, IClient
+    internal class Producer<TKey, TValue> : IProducer<TKey, TValue>
     {
         internal class Config
         {
@@ -56,8 +56,8 @@ internal class Config
             { typeof(float), Serializers.Single },
             { typeof(double), Serializers.Double },
             { typeof(byte[]), Serializers.ByteArray }
-        };
-
+        };
+        
         private int cancellationDelayMaxMs;
         private bool disposeHasBeenCalled = false;
         private object disposeHasBeenCalledLockObj = new object();
@@ -324,7 +324,7 @@ private void ProduceImpl(
             else
             {
                 err = KafkaHandle.Produce(
-                    topic,
+                    topic,
                     val, valOffset, valLength,
                     key, keyOffset, keyLength,
                     partition.Value,
@@ -741,7 +741,41 @@ internal Producer(ProducerBuilder<TKey, TValue> builder)
                 builder.AsyncKeySerializer, builder.AsyncValueSerializer);
         }
 
-
+        /// <inheritdoc/>
+        public async Task<ProduceResult> ProduceAsync(
+            string topic, 
+            Partition partition, 
+            ArraySegment<byte>? key, 
+            ArraySegment<byte>? value, 
+            IReadOnlyList<IHeader> headers, 
+            Timestamp timestamp)
+        {
+            // Start produce request
+            DeliveryHandlerSlim deliveryHandler = new();
+            ProduceImpl(topic, 
+                value?.Array, value?.Offset ?? 0, value?.Count ?? 0, 
+                key?.Array, key?.Offset ?? 0, key?.Count ?? 0, 
+                timestamp, 
+                partition, 
+                headers ?? Array.Empty<IHeader>(), 
+                deliveryHandler);
+
+            // Wait for response
+            DeliveryReport<Null, Null> deliveryReport = await deliveryHandler.Task.ConfigureAwait(false);
+
+            // Return response
+            return new(deliveryReport.Partition, deliveryReport.Offset, deliveryReport.Status);
+        }
+
+        /// <summary>
+        /// Implements light-weight delivery handler for produce requests.
+        /// </summary>
+        class DeliveryHandlerSlim : TaskCompletionSource<DeliveryReport<Null, Null>>, IDeliveryHandler
+        {
+            public DeliveryHandlerSlim() : base(TaskCreationOptions.RunContinuationsAsynchronously) {}
+            public void HandleDeliveryReport(DeliveryReport<Null, Null> deliveryReport) => TrySetResult(deliveryReport);
+        }
+
         /// <inheritdoc/>
         public async Task<DeliveryResult<TKey, TValue>> ProduceAsync(
             TopicPartition topicPartition,
@@ -815,7 +849,7 @@ public async Task<DeliveryResult<TKey, TValue>> ProduceAsync(
                 else
                 {
                     ProduceImpl(
-                        topicPartition.Topic, 
+                        topicPartition.Topic,
                         valBytes, 0, valBytes == null ? 0 : valBytes.Length, 
                         keyBytes, 0, keyBytes == null ? 0 : keyBytes.Length, 
                         message.Timestamp, topicPartition.Partition, headers.BackingList, 
@@ -912,7 +946,7 @@ public void Produce(
             }
 
             try
-            {
+            {
                 ProduceImpl(
                     topicPartition.Topic,
                     valBytes, 0, valBytes == null ? 0 : valBytes.Length,