Dispose ProduceAsync's CancellationTokenRegistration (#1289)

Author: mhowlett

src/Confluent.Kafka/Producer.cs

@@ -716,18 +716,19 @@ public async Task<DeliveryResult<TKey, TValue>> ProduceAsync(
                         enableDeliveryReportKey ? message.Key : default(TKey),
                         enableDeliveryReportValue ? message.Value : default(TValue));
 
+                    if (cancellationToken != null && cancellationToken.CanBeCanceled)
+                    {
+                        handler.CancellationTokenRegistration
+                            = cancellationToken.Register(() => handler.TrySetCanceled());
+                    }
+
                     ProduceImpl(
                         topicPartition.Topic,
                         valBytes, 0, valBytes == null ? 0 : valBytes.Length,
                         keyBytes, 0, keyBytes == null ? 0 : keyBytes.Length,
                         message.Timestamp, topicPartition.Partition, headers,
                         handler);
 
-                    if (cancellationToken != null && cancellationToken.CanBeCanceled)
-                    {
-                        cancellationToken.Register(() => handler.TrySetCanceled());
-                    }
-
                     return await handler.Task.ConfigureAwait(false);
                 }
                 else
@@ -867,6 +868,8 @@ public TypedTaskDeliveryHandlerShim(string topic, TKey key, TValue val)
                 Value = val;
             }
 
+            public CancellationTokenRegistration CancellationTokenRegistration;
+
             public string Topic;
 
             public TKey Key;
@@ -875,6 +878,11 @@ public TypedTaskDeliveryHandlerShim(string topic, TKey key, TValue val)
 
             public void HandleDeliveryReport(DeliveryReport<Null, Null> deliveryReport)
             {
+                if (CancellationTokenRegistration != null)
+                {
+                    CancellationTokenRegistration.Dispose();
+                }
+
                 if (deliveryReport == null)
                 {
 #if NET45