Merge branch '1.0.x'

Author: mhowlett

CHANGELOG.md

@@ -1,3 +1,22 @@
+# 1.0.0-RC6
+
+## New Features
+
+- Added DependentAdminClientBuilder class.
+
+
+## Changes
+
+- Reverted RC4 changes.
+- Renamed AvroSerializer to AsyncAvroSerializer and AvroDeserializer to AsyncAvroDeserializer
+- Added SyncOverAsyncSerializer and SyncOverAsyncDeserializer adapter classes.
+- Added AsSyncOverAsync factory method to AsyncAvroSerializer and AsyncAvroDeserializer.
+- Removed IAsyncDeserializer setter overloads from the ConsumerBuilder class.
+- Renamed Producer.BeginProduce to Producer.Produce.
+- Produce throws an exception if used when async serializers are configured.
+- Made AdminClient, Producer, and Consumer classes internal.
+
+
 # 1.0.0-RC4
 
 ## Changes

Confluent.Kafka.sln

@@ -49,6 +49,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ConfigGen", "src\ConfigGen\
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Protobuf", "examples\Protobuf\Protobuf.csproj", "{DBAE4CDC-7775-4D51-B457-ACEE5D1D9275}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Confluent.Kafka.SyncOverAsync", "test\Confluent.Kafka.SyncOverAsync\Confluent.Kafka.SyncOverAsync.csproj", "{07E3455B-9CDF-4124-854E-8AC9B8950B1A}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -302,6 +304,18 @@ Global
 		{DBAE4CDC-7775-4D51-B457-ACEE5D1D9275}.Release|x64.Build.0 = Release|Any CPU
 		{DBAE4CDC-7775-4D51-B457-ACEE5D1D9275}.Release|x86.ActiveCfg = Release|Any CPU
 		{DBAE4CDC-7775-4D51-B457-ACEE5D1D9275}.Release|x86.Build.0 = Release|Any CPU
+		{07E3455B-9CDF-4124-854E-8AC9B8950B1A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{07E3455B-9CDF-4124-854E-8AC9B8950B1A}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{07E3455B-9CDF-4124-854E-8AC9B8950B1A}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{07E3455B-9CDF-4124-854E-8AC9B8950B1A}.Debug|x64.Build.0 = Debug|Any CPU
+		{07E3455B-9CDF-4124-854E-8AC9B8950B1A}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{07E3455B-9CDF-4124-854E-8AC9B8950B1A}.Debug|x86.Build.0 = Debug|Any CPU
+		{07E3455B-9CDF-4124-854E-8AC9B8950B1A}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{07E3455B-9CDF-4124-854E-8AC9B8950B1A}.Release|Any CPU.Build.0 = Release|Any CPU
+		{07E3455B-9CDF-4124-854E-8AC9B8950B1A}.Release|x64.ActiveCfg = Release|Any CPU
+		{07E3455B-9CDF-4124-854E-8AC9B8950B1A}.Release|x64.Build.0 = Release|Any CPU
+		{07E3455B-9CDF-4124-854E-8AC9B8950B1A}.Release|x86.ActiveCfg = Release|Any CPU
+		{07E3455B-9CDF-4124-854E-8AC9B8950B1A}.Release|x86.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(NestedProjects) = preSolution
 		{09C3255B-1972-4EB8-91D0-FB9F5CD82BCB} = {1EFCD839-0726-4BCE-B745-1E829991B1BC}
@@ -324,5 +338,6 @@ Global
 		{C5114860-FBED-491E-8DB5-E8B51F2A2E6C} = {9CE4B5F7-9251-4340-BACB-207066A5DBE8}
 		{5642B638-F4AA-4C77-BE6E-C0FFF7CC103F} = {1EFCD839-0726-4BCE-B745-1E829991B1BC}
 		{DBAE4CDC-7775-4D51-B457-ACEE5D1D9275} = {9CE4B5F7-9251-4340-BACB-207066A5DBE8}
+		{07E3455B-9CDF-4124-854E-8AC9B8950B1A} = {90058283-1F8F-465D-89E4-D4374A27E612}
 	EndGlobalSection
 EndGlobal

README.md

@@ -42,16 +42,16 @@ confluent-kafka-dotnet is distributed via NuGet. We provide three packages:
 To install Confluent.Kafka from within Visual Studio, search for Confluent.Kafka in the NuGet Package Manager UI, or run the following command in the Package Manager Console:
 
 ```
-Install-Package Confluent.Kafka -Version 1.0.0-RC4
+Install-Package Confluent.Kafka -Version 1.0.0-RC6
 ```
 
 To add a reference to a dotnet core project, execute the following at the command line:
 
 ```
-dotnet add package -v 1.0.0-RC4 Confluent.Kafka
+dotnet add package -v 1.0.0-RC6 Confluent.Kafka
 ```
 
-**Note:** We recommend using the `1.0.0-RC4` version of Confluent.Kafka for new projects in preference to the most recent stable release (0.11.6).
+**Note:** We recommend using the `1.0.0-RC6` version of Confluent.Kafka for new projects in preference to the most recent stable release (0.11.6).
 The 1.0 API provides more features, is considerably improved and is more performant than 0.11.x releases.
 
 ### Branch builds
@@ -110,7 +110,7 @@ Note that a server round-trip is slow (3ms at a minimum; actual latency depends
 In highly concurrent scenarios you will achieve high overall throughput out of the producer using 
 the above approach, but there will be a delay on each `await` call. In stream processing 
 applications, where you would like to process many messages in rapid succession, you would typically
-make use the `BeginProduce` method instead:
+make use the `Produce` method instead:
 
 ```csharp
 using System;
@@ -131,7 +131,7 @@ class Program
         {
             for (int i=0; i<100; ++i)
             {
-                p.BeginProduce("my-topic", new Message<Null, string> { Value = i.ToString() }, handler);
+                p.Produce("my-topic", new Message<Null, string> { Value = i.ToString() }, handler);
             }
 
             // wait for up to 10 seconds for any inflight messages to be delivered.
@@ -234,7 +234,7 @@ this scenario there.
 
 #### Producer
 
-When using `BeginProduce`, to determine whether a particular message has been successfully delivered to a cluster,
+When using `Produce`, to determine whether a particular message has been successfully delivered to a cluster,
 check the `Error` field of the `DeliveryReport` during the delivery handler callback.
 
 When using `ProduceAsync`, any delivery result other than `NoError` will cause the returned `Task` to be in the

examples/AdminClient/AdminClient.csproj

@@ -9,7 +9,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <!-- nuget package reference: <PackageReference Include="Confluent.Kafka" Version="1.0.0-RC4" /> -->
+    <!-- nuget package reference: <PackageReference Include="Confluent.Kafka" Version="1.0.0-RC6" /> -->
     <ProjectReference Include="../../src/Confluent.Kafka/Confluent.Kafka.csproj" />
   </ItemGroup>
 

examples/AvroBlogExamples/AvroBlogExamples.csproj

@@ -8,7 +8,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <!-- nuget package reference: <PackageReference Include="Confluent.SchemaRegistry.Serdes" Version="1.0.0-RC4" /> -->
+    <!-- nuget package reference: <PackageReference Include="Confluent.SchemaRegistry.Serdes" Version="1.0.0-RC6" /> -->
     <ProjectReference Include="../../src/Confluent.SchemaRegistry.Serdes/Confluent.SchemaRegistry.Serdes.csproj" />
   </ItemGroup>
 

examples/AvroBlogExamples/Program.cs

@@ -39,7 +39,7 @@ async static Task ProduceGeneric(string bootstrapServers, string schemaRegistryU
             using (var schemaRegistry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { SchemaRegistryUrl = schemaRegistryUrl }))
             using (var producer =
                 new ProducerBuilder<Null, GenericRecord>(new ProducerConfig { BootstrapServers = bootstrapServers })
-                    .SetValueSerializer(new AvroSerializer<GenericRecord>(schemaRegistry))
+                    .SetValueSerializer(new AsyncAvroSerializer<GenericRecord>(schemaRegistry))
                     .Build())
             {   
                 var logLevelSchema = (Avro.EnumSchema)Avro.Schema.Parse(
@@ -71,7 +71,7 @@ async static Task ProduceSpecific(string bootstrapServers, string schemaRegistry
             using (var schemaRegistry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { SchemaRegistryUrl = schemaRegistryUrl }))
             using (var producer =
                 new ProducerBuilder<Null, MessageTypes.LogMessage>(new ProducerConfig { BootstrapServers = bootstrapServers })
-                    .SetValueSerializer(new AvroSerializer<MessageTypes.LogMessage>(schemaRegistry))
+                    .SetValueSerializer(new AsyncAvroSerializer<MessageTypes.LogMessage>(schemaRegistry))
                     .Build())
             {
                 await producer.ProduceAsync("log-messages",
@@ -105,10 +105,10 @@ static void ConsumeSpecific(string bootstrapServers, string schemaRegistryUrl)
                 AutoOffsetReset = AutoOffsetReset.Earliest
             };
 
-            using (var schemaRegistry = new CachedSchemaRegistryClient( new SchemaRegistryConfig { SchemaRegistryUrl = schemaRegistryUrl }))
+            using (var schemaRegistry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { SchemaRegistryUrl = schemaRegistryUrl }))
             using (var consumer =
                 new ConsumerBuilder<Null, MessageTypes.LogMessage>(consumerConfig)
-                    .SetValueDeserializer(new AvroDeserializer<MessageTypes.LogMessage>(schemaRegistry))
+                    .SetValueDeserializer(new AsyncAvroDeserializer<MessageTypes.LogMessage>(schemaRegistry).AsSyncOverAsync())
                     .Build())
             {
                 consumer.Subscribe("log-messages");

examples/AvroGeneric/AvroGeneric.csproj

@@ -9,7 +9,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <!-- nuget package reference: <PackageReference Include="Confluent.SchemaRegistry.Serdes" Version="1.0.0-RC4" /> -->
+    <!-- nuget package reference: <PackageReference Include="Confluent.SchemaRegistry.Serdes" Version="1.0.0-RC6" /> -->
     <ProjectReference Include="../../src/Confluent.SchemaRegistry.Serdes/Confluent.SchemaRegistry.Serdes.csproj" />
   </ItemGroup>
 

examples/AvroGeneric/Program.cs

@@ -61,8 +61,8 @@ static async Task Main(string[] args)
                 using (var schemaRegistry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { SchemaRegistryUrl = schemaRegistryUrl }))
                 using (var consumer =
                     new ConsumerBuilder<string, GenericRecord>(new ConsumerConfig { BootstrapServers = bootstrapServers, GroupId = groupName })
-                        .SetKeyDeserializer(new AvroDeserializer<string>(schemaRegistry))
-                        .SetValueDeserializer(new AvroDeserializer<GenericRecord>(schemaRegistry))
+                        .SetKeyDeserializer(new AsyncAvroDeserializer<string>(schemaRegistry).AsSyncOverAsync())
+                        .SetValueDeserializer(new AsyncAvroDeserializer<GenericRecord>(schemaRegistry).AsSyncOverAsync())
                         .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
                         .Build())
                 {
@@ -95,8 +95,8 @@ static async Task Main(string[] args)
             using (var schemaRegistry = new CachedSchemaRegistryClient(new SchemaRegistryConfig { SchemaRegistryUrl = schemaRegistryUrl }))
             using (var producer =
                 new ProducerBuilder<string, GenericRecord>(new ProducerConfig { BootstrapServers = bootstrapServers })
-                    .SetKeySerializer(new AvroSerializer<string>(schemaRegistry))
-                    .SetValueSerializer(new AvroSerializer<GenericRecord>(schemaRegistry))
+                    .SetKeySerializer(new AsyncAvroSerializer<string>(schemaRegistry))
+                    .SetValueSerializer(new AsyncAvroSerializer<GenericRecord>(schemaRegistry))
                     .Build())
             {
                 Console.WriteLine($"{producer.Name} producing on {topicName}. Enter user names, q to exit.");

examples/AvroSpecific/AvroSpecific.csproj

@@ -9,7 +9,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <!-- nuget package reference: <PackageReference Include="Confluent.SchemaRegistry.Serdes" Version="1.0.0-RC4" /> -->
+    <!-- nuget package reference: <PackageReference Include="Confluent.SchemaRegistry.Serdes" Version="1.0.0-RC6" /> -->
     <ProjectReference Include="../../src/Confluent.SchemaRegistry.Serdes/Confluent.SchemaRegistry.Serdes.csproj" />
   </ItemGroup>
 

examples/AvroSpecific/Program.cs

@@ -80,8 +80,8 @@ static async Task Main(string[] args)
                 using (var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig))
                 using (var consumer =
                     new ConsumerBuilder<string, User>(consumerConfig)
-                        .SetKeyDeserializer(new AvroDeserializer<string>(schemaRegistry))
-                        .SetValueDeserializer(new AvroDeserializer<User>(schemaRegistry))
+                        .SetKeyDeserializer(new AsyncAvroDeserializer<string>(schemaRegistry).AsSyncOverAsync())
+                        .SetValueDeserializer(new AsyncAvroDeserializer<User>(schemaRegistry).AsSyncOverAsync())
                         .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
                         .Build())
                 {
@@ -113,8 +113,8 @@ static async Task Main(string[] args)
             using (var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig))
             using (var producer =
                 new ProducerBuilder<string, User>(producerConfig)
-                    .SetKeySerializer(new AvroSerializer<string>(schemaRegistry))
-                    .SetValueSerializer(new AvroSerializer<User>(schemaRegistry))
+                    .SetKeySerializer(new AsyncAvroSerializer<string>(schemaRegistry))
+                    .SetValueSerializer(new AsyncAvroSerializer<User>(schemaRegistry))
                     .Build())
             {
                 Console.WriteLine($"{producer.Name} producing on {topicName}. Enter user names, q to exit.");