all tests pass

Author: Matt Howlett

src/Confluent.Kafka.Avro/AvroSerdeProvider.cs

@@ -62,7 +62,7 @@ public SerializerGenerator<T> GetSerializerGenerator<T>()
             return (forKey) =>
             {
                 var serializer = new AvroSerializer<T>(schemaRegistryClient);
-                serializer.Configure(config, true);
+                serializer.Configure(config, forKey);
                 return (string topic, T data) => serializer.Serialize(topic, data);
             };
         }

src/Confluent.Kafka/Consumer.cs

@@ -237,8 +237,13 @@ public Consumer(
         {
             Librdkafka.Initialize(null);
 
-            KeyDeserializer = keyDeserializerGenerator(true);
-            ValueDeserializer = valueDeserializerGenerator(false);
+            KeyDeserializer = keyDeserializerGenerator == null 
+                ? null
+                : keyDeserializerGenerator(true);
+                
+            ValueDeserializer = valueDeserializerGenerator == null
+                ? null
+                : valueDeserializerGenerator(false);
 
             if (KeyDeserializer != null && (object)KeyDeserializer == (object)ValueDeserializer)
             {

src/Confluent.Kafka/Producer.cs

@@ -166,7 +166,9 @@ public Producer(
 
             this.handle = ownedClient.Handle;
             this.producer = ownedClient;
-            setAndValidateSerializers(keySerializerGenerator(true), valueSerializerGenerator(false));
+            setAndValidateSerializers(
+                keySerializerGenerator == null ? null: keySerializerGenerator(true),
+                valueSerializerGenerator == null ? null : valueSerializerGenerator(false));
         }
 
 
@@ -184,8 +186,8 @@ public Producer(
         /// </param>
         public Producer(
             Handle handle,
-            Serializer<TKey> keySerializer,
-            Serializer<TValue> valueSerializer)
+            Serializer<TKey> keySerializer = null,
+            Serializer<TValue> valueSerializer = null)
         {
             if (!(handle.Owner is Producer))
             {
@@ -224,7 +226,9 @@ public Producer(
             this.ownedClient = null;
             this.handle = handle;
             this.producer = (Producer)handle.Owner;
-            setAndValidateSerializers(keySerializerGenerator(true), valueSerializerGenerator(false));
+            setAndValidateSerializers(
+                keySerializerGenerator == null ? null : keySerializerGenerator(true),
+                valueSerializerGenerator == null ? null : valueSerializerGenerator(false));
         }
 
 

src/Confluent.SchemaRegistry/CachedSchemaRegistryClient.cs

@@ -83,7 +83,9 @@ public CachedSchemaRegistryClient(IEnumerable<KeyValuePair<string, string>> conf
             var basicAuthInfo = Convert.ToString(config.FirstOrDefault(prop => prop.Key.ToLower() == SchemaRegistryConfig.PropertyNames.SchemaRegistryBasicAuthUserInfo).Value);
             if (basicAuthInfo != "" && basicAuthSource == "")
             {
-                throw new ArgumentException($"CachedSchemaRegistryClient: {SchemaRegistryConfig.PropertyNames.SchemaRegistryBasicAuthUserInfo} was specified, but {SchemaRegistryConfig.PropertyNames.SchemaRegistryBasicAuthCredentialsSource} was not.");
+                // default to USER_INFO if no source specified, since the strongly typed config
+                // class doesn't even expose the config source property.
+                basicAuthSource = "USER_INFO";
             }
 
             string username = null;

test/Confluent.Kafka.Avro.IntegrationTests/test.parameters.json

@@ -1,4 +1,4 @@
 {
-  "bootstrapServer": "localhost:9092",
+  "bootstrapServer": "10.200.7.144:9092",
   "schemaRegistryServer": "localhost:8081"
 }

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

@@ -67,6 +67,8 @@ public static void LogDelegate(string bootstrapServers, string singlePartitionTo
                 consumer.OnLog += (_, m)
                     => logCount += 1;
 
+                consumer.Assign(new TopicPartition(singlePartitionTopic, 0));
+                
                 consumer.Consume(TimeSpan.FromMilliseconds(100));
             }
             Assert.True(logCount > 0);

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

@@ -48,7 +48,7 @@ public static void Producer_Produce_Null_DeliveryHandler(string bootstrapServers
                 Assert.Null(dr.Message.Value);
                 Assert.Equal(TimestampType.CreateTime, dr.Message.Timestamp.Type);
                 Assert.True(Math.Abs((DateTime.UtcNow - dr.Message.Timestamp.UtcDateTime).TotalMinutes) < 1.0);
-                count += 1;  
+                count += 1;
             };
 
             using (var producer = new Producer<Null, Null>(producerConfig))

test/Confluent.Kafka.IntegrationTests/kafka.parameters.json

@@ -1,5 +1,5 @@
 {
-  "bootstrapServers": "localhost:9092",
+  "bootstrapServers": "10.200.7.144:9092",
   "singlePartitionTopic": "test-topic-1",
   "partitionedTopic": "test-topic-2"
 }