DGS-23025 Add AssociatedNameStrategy

src/Confluent.SchemaRegistry.Serdes.Avro/AvroDeserializer.cs

@@ -1,4 +1,4 @@
-// Copyright 2016-2019 Confluent Inc.
+// Copyright 2016-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.
@@ -73,7 +73,8 @@ public AvroDeserializer(ISchemaRegistryClient schemaRegistryClient, AvroDeserial
             if (config == null) { return; }
 
             var nonAvroConfig = config
-                .Where(item => !item.Key.StartsWith("avro.") && !item.Key.StartsWith("rules."));
+                .Where(item => !item.Key.StartsWith("avro.") && !item.Key.StartsWith("rules.")
+                    && !item.Key.StartsWith("subject.name.strategy."));
             if (nonAvroConfig.Count() > 0)
             {
                 throw new ArgumentException($"AvroDeserializer: unknown configuration parameter {nonAvroConfig.First().Key}.");

src/Confluent.SchemaRegistry.Serdes.Avro/AvroSerializer.cs

@@ -1,4 +1,4 @@
-// Copyright 2016-2018 Confluent Inc.
+// Copyright 2016-2018 Confluent Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -91,7 +91,8 @@ public AvroSerializer(ISchemaRegistryClient schemaRegistryClient, AvroSerializer
             if (config == null) { return; }
 
             var nonAvroConfig = config
-                .Where(item => !item.Key.StartsWith("avro.") && !item.Key.StartsWith("rules."));
+                .Where(item => !item.Key.StartsWith("avro.") && !item.Key.StartsWith("rules.")
+                    && !item.Key.StartsWith("subject.name.strategy."));
             if (nonAvroConfig.Count() > 0)
             {
                 throw new ArgumentException($"AvroSerializer: unknown configuration parameter {nonAvroConfig.First().Key}");

src/Confluent.SchemaRegistry.Serdes.Avro/GenericDeserializerImpl.cs

@@ -47,7 +47,7 @@ public GenericDeserializerImpl(
 
             if (config.UseLatestVersion != null) { this.useLatestVersion = config.UseLatestVersion.Value; }
             if (config.UseLatestWithMetadata != null) { this.useLatestWithMetadata = config.UseLatestWithMetadata; }
-            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToDelegate(); }
+            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToAsyncDelegate(schemaRegistryClient, config); }
             if (config.SchemaIdStrategy != null) { this.schemaIdDecoder = config.SchemaIdStrategy.Value.ToDeserializer(); }
         }
 
@@ -67,7 +67,7 @@ private async Task<GenericRecord> Deserialize(string topic, Headers headers, Rea
                 // Note: topic is not necessary for deserialization (or knowing if it's a key 
                 // or value) only the schema id is needed.
 
-                string subject = GetSubjectName(topic, isKey, null);
+                string subject = await GetSubjectName(topic, isKey, null).ConfigureAwait(false);
                 RegisteredSchema latestSchema = null;
                 if (subject != null)
                 {
@@ -89,7 +89,7 @@ private async Task<GenericRecord> Deserialize(string topic, Headers headers, Rea
                 (writerSchemaJson, writerSchema) = await GetWriterSchema(subject, writerId).ConfigureAwait(false);
                 if (subject == null)
                 {
-                    subject = GetSubjectName(topic, isKey, writerSchema.Fullname);
+                    subject = await GetSubjectName(topic, isKey, writerSchema.Fullname).ConfigureAwait(false);
                     if (subject != null)
                     {
                         latestSchema = await GetReaderSchema(subject)

src/Confluent.SchemaRegistry.Serdes.Avro/GenericSerializerImpl.cs

@@ -49,7 +49,7 @@ public GenericSerializerImpl(
             if (config.UseSchemaId != null) { this.useSchemaId = config.UseSchemaId.Value; }
             if (config.UseLatestVersion != null) { this.useLatestVersion = config.UseLatestVersion.Value; }
             if (config.UseLatestWithMetadata != null) { this.useLatestWithMetadata = config.UseLatestWithMetadata; }
-            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToDelegate(); }
+            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToAsyncDelegate(schemaRegistryClient, config); }
             if (config.SchemaIdStrategy != null) { this.schemaIdEncoder = config.SchemaIdStrategy.Value.ToEncoder(); }
 
             if (this.useLatestVersion && this.autoRegisterSchema)
@@ -132,7 +132,7 @@ public async Task<byte[]> Serialize(string topic, Headers headers, GenericRecord
                     // better to use hash functions based on the writerSchemaString 
                     // object reference, not value.
 
-                    subject = GetSubjectName(topic, isKey, data.Schema.Fullname);
+                    subject = await GetSubjectName(topic, isKey, data.Schema.Fullname).ConfigureAwait(false);
                     var subjectSchemaPair = new KeyValuePair<string, string>(subject, writerSchemaString);
                     latestSchema = await GetReaderSchema(subject)
                         .ConfigureAwait(continueOnCapturedContext: false);

src/Confluent.SchemaRegistry.Serdes.Avro/SpecificDeserializerImpl.cs

@@ -98,7 +98,7 @@ public SpecificDeserializerImpl(
 
             if (config.UseLatestVersion != null) { this.useLatestVersion = config.UseLatestVersion.Value; }
             if (config.UseLatestWithMetadata != null) { this.useLatestWithMetadata = config.UseLatestWithMetadata; }
-            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToDelegate(); }
+            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToAsyncDelegate(schemaRegistryClient, config); }
             if (config.SchemaIdStrategy != null) { this.schemaIdDecoder = config.SchemaIdStrategy.Value.ToDeserializer(); }
         }
 
@@ -118,7 +118,7 @@ public async Task<T> Deserialize(string topic, Headers headers, ReadOnlyMemory<b
                 // Note: topic is not necessary for deserialization (or knowing if it's a key 
                 // or value) only the schema id is needed.
 
-                string subject = GetSubjectName(topic, isKey, null);
+                string subject = await GetSubjectName(topic, isKey, null).ConfigureAwait(false);
                 RegisteredSchema latestSchema = null;
                 if (subject != null)
                 {
@@ -138,7 +138,7 @@ public async Task<T> Deserialize(string topic, Headers headers, ReadOnlyMemory<b
                 (writerSchemaJson, writerSchema) = await GetWriterSchema(subject, writerId).ConfigureAwait(false);
                 if (subject == null)
                 {
-                    subject = GetSubjectName(topic, isKey, writerSchema.Fullname);
+                    subject = await GetSubjectName(topic, isKey, writerSchema.Fullname).ConfigureAwait(false);
                     if (subject != null)
                     {
                         latestSchema = await GetReaderSchema(subject)

src/Confluent.SchemaRegistry.Serdes.Avro/SpecificSerializerImpl.cs

@@ -75,7 +75,7 @@ public SpecificSerializerImpl(
             if (config.UseSchemaId != null) { this.useSchemaId = config.UseSchemaId.Value; }
             if (config.UseLatestVersion != null) { this.useLatestVersion = config.UseLatestVersion.Value; }
             if (config.UseLatestWithMetadata != null) { this.useLatestWithMetadata = config.UseLatestWithMetadata; }
-            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToDelegate(); }
+            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToAsyncDelegate(schemaRegistryClient, config); }
             if (config.SchemaIdStrategy != null) { this.schemaIdEncoder = config.SchemaIdStrategy.Value.ToEncoder(); }
 
             if (this.useLatestVersion && this.autoRegisterSchema)
@@ -173,7 +173,7 @@ public async Task<byte[]> Serialize(string topic, Headers headers, T data, bool
                         fullname = ((Avro.RecordSchema)((ISpecificRecord)data).Schema).Fullname;
                     }
 
-                    subject = GetSubjectName(topic, isKey, fullname);
+                    subject = await GetSubjectName(topic, isKey, fullname).ConfigureAwait(false);
                     var subjectSchemaPair = new KeyValuePair<string, string>(subject, currentSchemaData.WriterSchemaString);
                     latestSchema = await GetReaderSchema(subject)
                         .ConfigureAwait(continueOnCapturedContext: false);

src/Confluent.SchemaRegistry.Serdes.Json/JsonDeserializer.cs

@@ -105,15 +105,16 @@ public JsonDeserializer(ISchemaRegistryClient schemaRegistryClient, JsonDeserial
             if (config == null) { return; }
 
             var nonJsonConfig = config
-                .Where(item => !item.Key.StartsWith("json.") && !item.Key.StartsWith("rules."));
+                .Where(item => !item.Key.StartsWith("json.") && !item.Key.StartsWith("rules.")
+                    && !item.Key.StartsWith("subject.name.strategy."));
             if (nonJsonConfig.Count() > 0)
             {
                 throw new ArgumentException($"JsonDeserializer: unknown configuration parameter {nonJsonConfig.First().Key}.");
             }
 
             if (config.UseLatestVersion != null) { this.useLatestVersion = config.UseLatestVersion.Value; }
             if (config.UseLatestWithMetadata != null) { this.useLatestWithMetadata = config.UseLatestWithMetadata; }
-            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToDelegate(); }
+            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToAsyncDelegate(schemaRegistryClient, config); }
             if (config.SchemaIdStrategy != null) { this.schemaIdDecoder = config.SchemaIdStrategy.Value.ToDeserializer(); }
             if (config.Validate != null) { this.validate = config.Validate.Value; }
         }
@@ -172,7 +173,7 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i
 
             bool isKey = context.Component == MessageComponentType.Key;
             string topic = context.Topic;
-            string subject = GetSubjectName(topic, isKey, null);
+            string subject = await GetSubjectName(topic, isKey, null).ConfigureAwait(false);
             RegisteredSchema latestSchema = null;
             if (subject != null)
             {
@@ -197,7 +198,7 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i
                         .ConfigureAwait(false);
                     if (subject == null)
                     {
-                        subject = GetSubjectName(topic, isKey, writerSchema.Title);
+                        subject = await GetSubjectName(topic, isKey, writerSchema.Title).ConfigureAwait(false);
                         if (subject != null)
                         {
                             latestSchema = await GetReaderSchema(subject)

src/Confluent.SchemaRegistry.Serdes.Json/JsonSerializer.cs

@@ -117,7 +117,8 @@ public JsonSerializer(ISchemaRegistryClient schemaRegistryClient, JsonSerializer
             if (config == null) { return; }
 
             var nonJsonConfig = config
-                .Where(item => !item.Key.StartsWith("json.") && !item.Key.StartsWith("rules."));
+                .Where(item => !item.Key.StartsWith("json.") && !item.Key.StartsWith("rules.")
+                    && !item.Key.StartsWith("subject.name.strategy."));
             if (nonJsonConfig.Count() > 0)
             {
                 throw new ArgumentException($"JsonSerializer: unknown configuration parameter {nonJsonConfig.First().Key}");
@@ -130,7 +131,7 @@ public JsonSerializer(ISchemaRegistryClient schemaRegistryClient, JsonSerializer
             if (config.UseLatestVersion != null) { this.useLatestVersion = config.UseLatestVersion.Value; }
             if (config.LatestCompatibilityStrict != null) { this.latestCompatibilityStrict = config.LatestCompatibilityStrict.Value; }
             if (config.UseLatestWithMetadata != null) { this.useLatestWithMetadata = config.UseLatestWithMetadata; }
-            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToDelegate(); }
+            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToAsyncDelegate(schemaRegistryClient, config); }
             if (config.SchemaIdStrategy != null) { this.schemaIdEncoder = config.SchemaIdStrategy.Value.ToEncoder(); }
             if (config.Validate != null) { this.validate = config.Validate.Value; }
 
@@ -207,7 +208,8 @@ public override async Task<byte[]> SerializeAsync(T value, SerializationContext
                 await serdeMutex.WaitAsync().ConfigureAwait(continueOnCapturedContext: false);
                 try
                 {
-                    subject = GetSubjectName(context.Topic, context.Component == MessageComponentType.Key, this.schemaFullname);
+                    subject = await GetSubjectName(context.Topic, context.Component == MessageComponentType.Key, this.schemaFullname)
+                        .ConfigureAwait(false);
                     latestSchema = await GetReaderSchema(subject, new Schema(schemaText, ReferenceList, SchemaType.Json))
                         .ConfigureAwait(continueOnCapturedContext: false);
 

src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufDeserializer.cs

@@ -73,7 +73,8 @@ public ProtobufDeserializer(ISchemaRegistryClient schemaRegistryClient, Protobuf
             if (config == null) { return; }
 
             var nonProtobufConfig = config
-                .Where(item => !item.Key.StartsWith("protobuf.") && !item.Key.StartsWith("rules."));
+                .Where(item => !item.Key.StartsWith("protobuf.") && !item.Key.StartsWith("rules.")
+                    && !item.Key.StartsWith("subject.name.strategy."));
             if (nonProtobufConfig.Count() > 0)
             {
                 throw new ArgumentException($"ProtobufDeserializer: unknown configuration parameter {nonProtobufConfig.First().Key}");
@@ -87,7 +88,7 @@ public ProtobufDeserializer(ISchemaRegistryClient schemaRegistryClient, Protobuf
 
             if (config.UseLatestVersion != null) { this.useLatestVersion = config.UseLatestVersion.Value; }
             if (config.UseLatestWithMetadata != null) { this.useLatestWithMetadata = config.UseLatestWithMetadata; }
-            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToDelegate(); }
+            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToAsyncDelegate(schemaRegistryClient, config); }
             if (config.SchemaIdStrategy != null) { this.schemaIdDecoder = config.SchemaIdStrategy.Value.ToDeserializer(); }
         }
 
@@ -114,7 +115,7 @@ public override async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool i
 
             bool isKey = context.Component == MessageComponentType.Key;
             string topic = context.Topic;
-            string subject = GetSubjectName(topic, isKey, null);
+            string subject = await GetSubjectName(topic, isKey, null).ConfigureAwait(false);
 
             // Currently Protobuf does not support migration rules because of lack of support for DynamicMessage
             // See https://github.com/protocolbuffers/protobuf/issues/658

src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufSerializer.cs

@@ -77,7 +77,8 @@ public ProtobufSerializer(ISchemaRegistryClient schemaRegistryClient, ProtobufSe
             }
 
             var nonProtobufConfig = config
-                .Where(item => !item.Key.StartsWith("protobuf.") && !item.Key.StartsWith("rules."));
+                .Where(item => !item.Key.StartsWith("protobuf.") && !item.Key.StartsWith("rules.")
+                    && !item.Key.StartsWith("subject.name.strategy."));
             if (nonProtobufConfig.Count() > 0)
             {
                 throw new ArgumentException($"ProtobufSerializer: unknown configuration parameter {nonProtobufConfig.First().Key}");
@@ -94,7 +95,7 @@ public ProtobufSerializer(ISchemaRegistryClient schemaRegistryClient, ProtobufSe
             {
                 throw new NotSupportedException("ProtobufSerializer: UseDeprecatedFormat is no longer supported");
             }
-            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToDelegate(); }
+            if (config.SubjectNameStrategy != null) { this.subjectNameStrategy = config.SubjectNameStrategy.Value.ToAsyncDelegate(schemaRegistryClient, config); }
             if (config.SchemaIdStrategy != null) { this.schemaIdEncoder = config.SchemaIdStrategy.Value.ToEncoder(); }
             this.referenceSubjectNameStrategy = config.ReferenceSubjectNameStrategy == null
                 ? ReferenceSubjectNameStrategy.ReferenceName.ToDelegate()
@@ -231,7 +232,8 @@ public override async Task<byte[]> SerializeAsync(T value, SerializationContext
                 await serdeMutex.WaitAsync().ConfigureAwait(continueOnCapturedContext: false);
                 try
                 {
-                    subject = GetSubjectName(context.Topic, context.Component == MessageComponentType.Key, fullname);
+                    subject = await GetSubjectName(context.Topic, context.Component == MessageComponentType.Key, fullname)
+                        .ConfigureAwait(false);
                     latestSchema = await GetReaderSchema(subject)
                         .ConfigureAwait(continueOnCapturedContext: false);
 

src/Confluent.SchemaRegistry/AsyncSerde.cs

@@ -39,7 +39,7 @@ public abstract class AsyncSerde<TParsedSchema>
         protected bool useLatestVersion = false;
         protected bool latestCompatibilityStrict = false;
         protected IDictionary<string, string> useLatestWithMetadata = null;
-        protected SubjectNameStrategyDelegate subjectNameStrategy = null;
+        protected AsyncSubjectNameStrategyDelegate subjectNameStrategy = null;
 
         protected SemaphoreSlim serdeMutex = new SemaphoreSlim(1);
 
@@ -65,29 +65,22 @@ protected AsyncSerde(ISchemaRegistryClient schemaRegistryClient, SerdeConfig con
             }
         }
 
-        protected string GetSubjectName(string topic, bool isKey, string recordType)
+        protected async Task<string> GetSubjectName(string topic, bool isKey, string recordType)
         {
-            try
-            {
-                string subject = this.subjectNameStrategy != null
-                    // use the subject name strategy specified in the serializer config if available.
-                    ? this.subjectNameStrategy(
-                        new SerializationContext(
-                            isKey ? MessageComponentType.Key : MessageComponentType.Value,
-                            topic),
-                        recordType)
-                    // else fall back to the deprecated config from (or default as currently supplied by) SchemaRegistry.
-                    : schemaRegistryClient == null
-                        ? null
-                        : isKey
-                            ? schemaRegistryClient.ConstructKeySubjectName(topic, recordType)
-                            : schemaRegistryClient.ConstructValueSubjectName(topic, recordType);
-                return subject;
-            }
-            catch (Exception e)
-            {
-                return null;
-            }
+            string subject = this.subjectNameStrategy != null
+                // use the subject name strategy specified in the serializer config if available.
+                ? await this.subjectNameStrategy(
+                    new SerializationContext(
+                        isKey ? MessageComponentType.Key : MessageComponentType.Value,
+                        topic),
+                    recordType).ConfigureAwait(false)
+                // else fall back to the deprecated config from (or default as currently supplied by) SchemaRegistry.
+                : schemaRegistryClient == null
+                    ? null
+                    : isKey
+                        ? schemaRegistryClient.ConstructKeySubjectName(topic, recordType)
+                        : schemaRegistryClient.ConstructValueSubjectName(topic, recordType);
+            return subject;
         }
 
         protected async Task<(Schema, TParsedSchema)> GetWriterSchema(string subject, SchemaId writerId, string format = null)