feat: added new topic configuration properties

Author: gokerakc

src/LEGO.AsyncAPI.Bindings/Kafka/KafkaChannelBinding.cs

@@ -51,6 +51,10 @@ public class KafkaChannelBinding : ChannelBinding<KafkaChannelBinding>
             { "retention.bytes", (a, n) => { a.RetentionBytes = n.GetIntegerValue(); } },
             { "delete.retention.ms", (a, n) => { a.DeleteRetentionMiliseconds = n.GetIntegerValue(); } },
             { "max.message.bytes", (a, n) => { a.MaxMessageBytes = n.GetIntegerValue(); } },
+            { "confluent.key.schema.validation", (a, n) => { a.ConfluentKeySchemaValidation = n.GetBooleanValue(); } },
+            { "confluent.key.subject.name.strategy", (a, n) => { a.ConfluentKeySubjectName = n.GetScalarValue(); } },
+            { "confluent.value.schema.validation", (a, n) => { a.ConfluentValueSchemaValidation = n.GetBooleanValue(); } },
+            { "confluent.value.subject.name.strategy", (a, n) => { a.ConfluentValueSubjectName = n.GetScalarValue(); } },
         };
 
         /// <summary>

src/LEGO.AsyncAPI.Bindings/Kafka/TopicConfigurationObject.cs

@@ -33,6 +33,26 @@ public class TopicConfigurationObject : IAsyncApiElement
         /// The max.message.bytes configuration option.
         /// </summary>
         public int? MaxMessageBytes { get; set; }
+        
+        /// <summary>
+        /// The confluent.key.schema.validation configuration option.
+        /// </summary>
+        public bool? ConfluentKeySchemaValidation { get; set; }
+        
+        /// <summary>
+        /// The confluent.key.subject.name.strategy configuration option.
+        /// </summary>
+        public string ConfluentKeySubjectName { get; set; }
+        
+        /// <summary>
+        /// The confluent.value.schema.validation configuration option.
+        /// </summary>
+        public bool? ConfluentValueSchemaValidation { get; set; }
+        
+        /// <summary>
+        /// The confluent.value.subject.name.strategy configuration option.
+        /// </summary>
+        public string ConfluentValueSubjectName { get; set; }
 
         public void Serialize(IAsyncApiWriter writer)
         {
@@ -47,6 +67,10 @@ public void Serialize(IAsyncApiWriter writer)
             writer.WriteOptionalProperty<int>(AsyncApiConstants.RetentionBytes, this.RetentionBytes);
             writer.WriteOptionalProperty<int>(AsyncApiConstants.DeleteRetentionMiliseconds, this.DeleteRetentionMiliseconds);
             writer.WriteOptionalProperty<int>(AsyncApiConstants.MaxMessageBytes, this.MaxMessageBytes);
+            writer.WriteOptionalProperty<bool>(AsyncApiConstants.ConfluentKeySchemaValidation, this.ConfluentKeySchemaValidation);
+            writer.WriteOptionalProperty(AsyncApiConstants.ConfluentKeySubjectName, this.ConfluentKeySubjectName);
+            writer.WriteOptionalProperty<bool>(AsyncApiConstants.ConfluentValueSchemaValidation, this.ConfluentValueSchemaValidation);
+            writer.WriteOptionalProperty(AsyncApiConstants.ConfluentValueSubjectName, this.ConfluentValueSubjectName);
             writer.WriteEndObject();
         }
     }

src/LEGO.AsyncAPI/Models/AsyncApiConstants.cs

@@ -136,6 +136,10 @@ public static class AsyncApiConstants
         public const string RetentionBytes = "retention.bytes";
         public const string DeleteRetentionMiliseconds = "delete.retention.ms";
         public const string MaxMessageBytes = "max.message.bytes";
+        public const string ConfluentKeySchemaValidation = "confluent.key.schema.validation";
+        public const string ConfluentKeySubjectName = "confluent.key.subject.name.strategy";
+        public const string ConfluentValueSchemaValidation = "confluent.value.schema.validation";
+        public const string ConfluentValueSubjectName = "confluent.value.subject.name.strategy";
         public const string TopicConfiguration = "topicConfiguration";
         public const string GeoReplication = "geo-replication";
         public const string AdditionalItems = "additionalItems";

test/LEGO.AsyncAPI.Tests/Bindings/Kafka/KafkaBindings_Should.cs

@@ -30,7 +30,11 @@ public void KafkaChannelBinding_WithFilledObject_SerializesAndDeserializes()
       retention.ms: 1
       retention.bytes: 2
       delete.retention.ms: 3
-      max.message.bytes: 4";
+      max.message.bytes: 4
+      confluent.key.schema.validation: true
+      confluent.key.subject.name.strategy: TopicNameStrategy
+      confluent.value.schema.validation: true
+      confluent.value.subject.name.strategy: TopicNameStrategy";
 
             var channel = new AsyncApiChannel();
             channel.Bindings.Add(new KafkaChannelBinding
@@ -45,6 +49,10 @@ public void KafkaChannelBinding_WithFilledObject_SerializesAndDeserializes()
                     RetentionBytes = 2,
                     DeleteRetentionMiliseconds = 3,
                     MaxMessageBytes = 4,
+                    ConfluentKeySchemaValidation = true,
+                    ConfluentKeySubjectName = "TopicNameStrategy",
+                    ConfluentValueSchemaValidation = true,
+                    ConfluentValueSubjectName = "TopicNameStrategy",
                 },
             });