Merge branch '1.0.x'

Author: mhowlett

CHANGELOG.md

@@ -26,6 +26,7 @@
 - Builder classes now return interfaces, not concrete classes.
 - Removed the dependency on `CompilerServices.Unsafe` which was causing `ProduceAsync` to hang in some scenarios.
 - Fixed a deadlock-on-dispose issue in `AdminClient`.
+- Made `Producer.ProduceAsync` async.
 
 
 # 1.0.0-beta3

src/Confluent.Kafka/Producer.cs

@@ -658,7 +658,7 @@ internal Producer(ProducerBuilder<TKey, TValue> builder)
         ///     A Task which will complete with a delivery report corresponding to
         ///     the produce request, or an exception if an error occured.
         /// </returns>
-        public Task<DeliveryResult<TKey, TValue>> ProduceAsync(
+        public async Task<DeliveryResult<TKey, TValue>> ProduceAsync(
             TopicPartition topicPartition,
             Message<TKey, TValue> message)
         {
@@ -667,10 +667,7 @@ public Task<DeliveryResult<TKey, TValue>> ProduceAsync(
             {
                 keyBytes = (keySerializer != null)
                     ? keySerializer.Serialize(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic))
-                    : asyncKeySerializer.SerializeAsync(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic))
-                        .ConfigureAwait(continueOnCapturedContext: false)
-                        .GetAwaiter()
-                        .GetResult();
+                    : await asyncKeySerializer.SerializeAsync(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic));
             }
             catch (Exception exception)
             {
@@ -689,10 +686,7 @@ public Task<DeliveryResult<TKey, TValue>> ProduceAsync(
             {
                 valBytes = (valueSerializer != null)
                     ? valueSerializer.Serialize(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic))
-                    : asyncValueSerializer.SerializeAsync(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic))
-                        .ConfigureAwait(continueOnCapturedContext: false)
-                        .GetAwaiter()
-                        .GetResult();
+                    : await asyncValueSerializer.SerializeAsync(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic));
             }
             catch (Exception exception)
             {
@@ -722,7 +716,7 @@ public Task<DeliveryResult<TKey, TValue>> ProduceAsync(
                         message.Timestamp, topicPartition.Partition, message.Headers,
                         handler);
 
-                    return handler.Task;
+                    return await handler.Task;
                 }
                 else
                 {
@@ -739,7 +733,7 @@ public Task<DeliveryResult<TKey, TValue>> ProduceAsync(
                         Message = message
                     };
 
-                    return Task.FromResult(result);
+                    return result;
                 }
             }
             catch (KafkaException ex)

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

@@ -58,7 +58,7 @@ public void AddBrokers(string bootstrapServers)
                 Assert.True(brokersAdded > 0, "Should have added one broker or more (duplicates considered added)");
 
                 var newMetadata = adminClient.GetMetadata(TimeSpan.FromSeconds(3));
-                Assert.True(newMetadata.Brokers.Count == 1);
+                Assert.True(newMetadata.Brokers.Count >= 1);
 
                 brokersAdded = adminClient.AddBrokers("");
                 Assert.True(brokersAdded == 0, "Should not have added brokers");

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

@@ -58,9 +58,10 @@ public void AdminClient_CreatePartitions(string bootstrapServers)
                     producer.ProduceAsync(new TopicPartition(topicName1, 2), new Message<Null, Null>()).Wait();
                     Assert.True(false, "expecting exception");
                 }
-                catch (KafkaException ex)
+                catch (AggregateException ex)
                 {
-                    Assert.True(ex.Error.IsError);
+                    Assert.IsType<ProduceException<Null,Null>>(ex.InnerException);
+                    Assert.True(((ProduceException<Null,Null>)ex.InnerException).Error.IsError);
                 }
             }
 
@@ -79,9 +80,10 @@ public void AdminClient_CreatePartitions(string bootstrapServers)
                     var dr2 = producer.ProduceAsync(new TopicPartition(topicName2, 1), new Message<Null, Null>()).Result;
                     Assert.True(false, "expecting exception");
                 }
-                catch (KafkaException ex)
+                catch (AggregateException ex)
                 {
-                    Assert.True(ex.Error.IsError);
+                    Assert.IsType<ProduceException<Null,Null>>(ex.InnerException);
+                    Assert.True(((ProduceException<Null,Null>)ex.InnerException).Error.IsError);
                 }
             }
 

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

@@ -70,7 +70,7 @@ public void Timestamps(string bootstrapServers)
                     new Message<Null, string> { Value = "test-value" }).Result);
 
                 // TimestampType: LogAppendTime
-                Assert.Throws<ArgumentException>(() =>
+                Assert.Throws<AggregateException>(() =>
                     producer.ProduceAsync(
                         new TopicPartition(singlePartitionTopic, 0),
                         new Message<Null, string>
@@ -80,7 +80,7 @@ public void Timestamps(string bootstrapServers)
                         }).Result);
 
                 // TimestampType: NotAvailable
-                Assert.Throws<ArgumentException>(() =>
+                Assert.Throws<AggregateException>(() =>
                     producer.ProduceAsync(
                         new TopicPartition(singlePartitionTopic, 0),
                         new Message<Null, string> 
@@ -159,13 +159,13 @@ Action<DeliveryReport<Null, string>> dh
                     new Message<byte[], byte[]> { Timestamp = Timestamp.Default }).Result);
 
                 // TimestampType: LogAppendTime
-                Assert.Throws<ArgumentException>(() =>
+                Assert.Throws<AggregateException>(() =>
                     producer.ProduceAsync(
                         singlePartitionTopic,
                         new Message<byte[], byte[]> { Timestamp = new Timestamp(DateTime.Now, TimestampType.LogAppendTime) }).Result);
 
                 // TimestampType: NotAvailable
-                Assert.Throws<ArgumentException>(() =>
+                Assert.Throws<AggregateException>(() =>
                     producer.ProduceAsync(
                         singlePartitionTopic,
                         new Message<byte[], byte[]> { Timestamp = new Timestamp(10, TimestampType.NotAvailable) }).Result);