producer poll (#856)

* producer poll

* + integration test, move check outside lock

* Local -> _

* Review feedback changes

Author: mhowlett

src/Confluent.Kafka/IProducer.cs

@@ -179,9 +179,7 @@ void BeginProduce(
 
 
         /// <summary>
-        ///     Poll for callback events. Typically, you should not 
-        ///     call this method. Only call on producer instances 
-        ///     where background polling has been disabled.
+        ///     Poll for callback events.
         /// </summary>
         /// <param name="timeout">
         ///     The maximum period of time to block if
@@ -190,7 +188,10 @@ void BeginProduce(
         ///     because this operation cannot be cancelled.
         /// </param>
         /// <returns>
-        ///     Returns the number of events served.
+        ///     Returns the number of events served since
+        ///     the last call to this method or if this 
+        ///     method has not yet been called, over the
+        ///     lifetime of the producer.
         /// </returns>
         int Poll(TimeSpan timeout);
 

src/Confluent.Kafka/Producer.cs

@@ -78,6 +78,9 @@ private SafeKafkaHandle KafkaHandle
         private Task callbackTask;
         private CancellationTokenSource callbackCts;
 
+        private int eventsServedCount = 0;
+        private object pollSyncObj = new object();
+
         private Task StartPollTask(CancellationToken ct)
             => Task.Factory.StartNew(() =>
                 {
@@ -86,7 +89,17 @@ private Task StartPollTask(CancellationToken ct)
                         while (true)
                         {
                             ct.ThrowIfCancellationRequested();
-                            ownedKafkaHandle.Poll((IntPtr)cancellationDelayMaxMs);
+                            int eventsServedCount_ = ownedKafkaHandle.Poll((IntPtr)cancellationDelayMaxMs);
+
+                            // note: lock {} is equivalent to Monitor.Enter then Monitor.Exit 
+                            if (eventsServedCount_ > 0)
+                            {
+                                lock (pollSyncObj)
+                                {
+                                    this.eventsServedCount += eventsServedCount_;
+                                    Monitor.Pulse(pollSyncObj);
+                                }
+                            }
                         }
                     }
                     catch (OperationCanceledException) {}
@@ -218,6 +231,7 @@ private void ProduceImpl(
                 }
             }
 
+            ErrorCode err;
             if (this.enableDeliveryReports && deliveryHandler != null)
             {
                 // Passes the TaskCompletionSource to the delivery report callback via the msg_opaque pointer
@@ -229,7 +243,7 @@ private void ProduceImpl(
                 var gch = GCHandle.Alloc(deliveryHandler);
                 var ptr = GCHandle.ToIntPtr(gch);
 
-                var err = KafkaHandle.Produce(
+                err = KafkaHandle.Produce(
                     topic,
                     val, valOffset, valLength,
                     key, keyOffset, keyLength,
@@ -240,25 +254,25 @@ private void ProduceImpl(
 
                 if (err != ErrorCode.NoError)
                 {
+                    // note: freed in the delivery handler callback otherwise.
                     gch.Free();
-                    throw new KafkaException(KafkaHandle.CreatePossiblyFatalError(err, null));
                 }
             }
             else
             {
-                var err = KafkaHandle.Produce(
+                err = KafkaHandle.Produce(
                     topic,
                     val, valOffset, valLength,
                     key, keyOffset, keyLength,
                     partition.Value,
                     timestamp.UnixTimestampMs,
                     headers,
                     IntPtr.Zero);
+            }
 
-                if (err != ErrorCode.NoError)
-                {
-                    throw new KafkaException(KafkaHandle.CreatePossiblyFatalError(err, null));
-                }
+            if (err != ErrorCode.NoError)
+            {
+                throw new KafkaException(KafkaHandle.CreatePossiblyFatalError(err, null));
             }
         }
 
@@ -268,12 +282,22 @@ private void ProduceImpl(
         /// </summary>
         public int Poll(TimeSpan timeout)
         {
-            if (!manualPoll)
+            if (manualPoll)
             {
-                throw new InvalidOperationException("Poll method called, but manual polling is not enabled.");
+                return this.KafkaHandle.Poll((IntPtr)timeout.TotalMillisecondsAsInt());
             }
 
-            return this.KafkaHandle.Poll((IntPtr)timeout.TotalMillisecondsAsInt());
+            lock (pollSyncObj)
+            {
+                if (eventsServedCount == 0)
+                {
+                    Monitor.Wait(pollSyncObj, timeout);
+                }
+
+                var result = eventsServedCount;
+                eventsServedCount = 0;
+                return result;
+            }
         }
 
 

test/Confluent.Kafka.Benchmark/BenchmarkProducer.cs

@@ -18,7 +18,6 @@
 using System.Threading.Tasks;
 using System.Threading;
 using System.Linq;
-using System.Collections.Generic;
 
 
 namespace Confluent.Kafka.Benchmark
@@ -76,6 +75,13 @@ private static long BenchmarkProducerImpl(
                         var msgCount = nMessages;
                         Action<DeliveryReport<Null, byte[]>> deliveryHandler = (DeliveryReport<Null, byte[]> deliveryReport) => 
                         {
+                            if (deliveryReport.Error.IsError)
+                            {
+                                // Not interested in benchmark results in the (unlikely) event there is an error.
+                                Console.WriteLine($"A error occured producing a message: {deliveryReport.Error.Reason}");
+                                Environment.Exit(1); // note: exceptions do not currently propagate to calling code from a deliveryHandler method.
+                            }
+
                             if (--msgCount == 0)
                             {
                                 autoEvent.Set();
@@ -84,19 +90,62 @@ private static long BenchmarkProducerImpl(
 
                         for (int i = 0; i < nMessages; i += 1)
                         {
-                            producer.BeginProduce(topic, new Message<Null, byte[]> { Value = val, Headers = headers }, deliveryHandler);
+                            try
+                            {
+                                producer.BeginProduce(topic, new Message<Null, byte[]> { Value = val, Headers = headers }, deliveryHandler);
+                            }
+                            catch (ProduceException<Null, byte[]> ex)
+                            {
+                                if (ex.Error.Code == ErrorCode.Local_QueueFull)
+                                {
+                                    producer.Poll(TimeSpan.FromSeconds(1));
+                                    i -= 1;
+                                }
+                                else
+                                {
+                                    throw;
+                                }
+                            }
                         }
 
-                        autoEvent.WaitOne();
+                        while (true)
+                        {
+                            if (autoEvent.WaitOne(TimeSpan.FromSeconds(1)))
+                            {
+                                break;
+                            }
+                            Console.WriteLine(msgCount);
+                        }
                     }
                     else
                     {
-                        var tasks = new Task[nMessages];
-                        for (int i = 0; i < nMessages; i += 1)
+                        try
+                        {
+                            var tasks = new Task[nMessages];
+                            for (int i = 0; i < nMessages; i += 1)
+                            {
+                                tasks[i] = producer.ProduceAsync(topic, new Message<Null, byte[]> { Value = val, Headers = headers });
+                                if (tasks[i].IsFaulted)
+                                {
+                                    if (((ProduceException<Null, byte[]>)tasks[i].Exception.InnerException).Error.Code == ErrorCode.Local_QueueFull)
+                                    {
+                                        producer.Poll(TimeSpan.FromSeconds(1));
+                                        i -= 1;
+                                    }
+                                    else
+                                    {
+                                        // unexpected, abort benchmark test.
+                                        throw tasks[i].Exception;
+                                    }
+                                }
+                            }
+
+                            Task.WaitAll(tasks);
+                        }
+                        catch (AggregateException ex)
                         {
-                            tasks[i] = producer.ProduceAsync(topic, new Message<Null, byte[]> { Value = val, Headers = headers });
+                            Console.WriteLine(ex.Message);
                         }
-                        Task.WaitAll(tasks);
                     }
 
                     var duration = DateTime.Now.Ticks - startTime;

test/Confluent.Kafka.IntegrationTests/Tests/Producer_Poll.cs

@@ -0,0 +1,67 @@
+// Copyright 2019 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.
+
+#pragma warning disable xUnit1026
+
+using System;
+using System.Diagnostics;
+using System.Threading.Tasks;
+using Xunit;
+
+
+namespace Confluent.Kafka.IntegrationTests
+{
+    /// <summary>
+    ///     Test Producer.Poll when producer in automatic poll mode.
+    /// </summary>
+    public partial class Tests
+    {
+        [Theory, MemberData(nameof(KafkaParameters))]
+        public void Producer_Poll(string bootstrapServers)
+        {
+            LogToFile("start Producer_Poll");
+
+            using (var tempTopic = new TemporaryTopic(bootstrapServers, 1))
+            using (var producer = new ProducerBuilder<Null, string>(new ProducerConfig { BootstrapServers = bootstrapServers }).Build())
+            {
+                var r = producer.ProduceAsync(tempTopic.Name, new Message<Null, string> { Value = "a message" }).Result;
+                Assert.True(r.Status == PersistenceStatus.Persisted);
+
+                // should be no events to serve and this should block for 500ms.
+                var sw = new Stopwatch();
+                sw.Start();
+                // Note: Poll returns the number of events served since the last
+                // call to Poll (or if the poll method hasn't beeen called, over
+                // the lifetime of the producer).
+                Assert.True(producer.Poll(TimeSpan.FromMilliseconds(500)) >= 1);
+                Assert.True(sw.ElapsedMilliseconds < 1);
+                Assert.Equal(0, producer.Poll(TimeSpan.FromMilliseconds(500)));
+                Assert.True(sw.ElapsedMilliseconds >= 500);
+
+                sw.Reset();
+                sw.Start();
+                producer.BeginProduce(tempTopic.Name, new Message<Null, string> { Value = "a message 2" }, dr => Assert.False(dr.Error.IsError));
+                // should block until the callback for the BeginProduce call executes.
+                Assert.Equal(1, producer.Poll(TimeSpan.FromSeconds(4)));
+                Assert.True(sw.ElapsedMilliseconds > 0);
+                Assert.True(sw.ElapsedMilliseconds < 3500);
+            }
+
+            Assert.Equal(0, Library.HandleCount);
+            LogToFile("end   Producer_Poll");
+        }
+    }
+}

test/Confluent.Kafka.IntegrationTests/Tests/Producer_Poll_Backoff.cs

@@ -0,0 +1,78 @@
+// Copyright 2019 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.
+
+#pragma warning disable xUnit1026
+
+using System;
+using System.Diagnostics;
+using System.Threading.Tasks;
+using Xunit;
+
+
+namespace Confluent.Kafka.IntegrationTests
+{
+    /// <summary>
+    ///     
+    /// </summary>
+    public partial class Tests
+    {
+        [Theory, MemberData(nameof(KafkaParameters))]
+        public void Producer_Poll_Backoff(string bootstrapServers)
+        {
+            LogToFile("start Producer_Poll_Backoff");
+
+            var pConfig = new ProducerConfig
+            {
+                BootstrapServers = bootstrapServers,
+                QueueBufferingMaxMessages = 10,
+                LingerMs = 100
+            };
+
+            using (var tempTopic = new TemporaryTopic(bootstrapServers, 1))
+            using (var producer = new ProducerBuilder<Null, string>(pConfig).Build())
+            {
+                // test timing around producer.Poll.
+                Stopwatch sw = new Stopwatch();
+                sw.Start();
+                var exceptionCount = 0;
+                for (int i=0; i<11; ++i)
+                {
+                    try
+                    {
+                        producer.BeginProduce(tempTopic.Name, new Message<Null, string> { Value = "a message" });
+                    }
+                    catch (ProduceException<Null, string> ex)
+                    {
+                        exceptionCount += 1;
+                        Assert.Equal(ErrorCode.Local_QueueFull, ex.Error.Code);
+                        var served = producer.Poll(TimeSpan.FromSeconds(4));
+                        Assert.True(served >= 1);
+                        var elapsed = sw.ElapsedMilliseconds;
+                        Assert.True(elapsed > 100); // linger.ms
+                        Assert.True(elapsed < 4000);
+                    }
+                }
+                Assert.Equal(1, exceptionCount);
+
+                producer.Flush();
+            }
+
+            Assert.Equal(0, Library.HandleCount);
+            LogToFile("end   Producer_Poll_Backoff");
+        }
+
+    }
+}