Use fallbacks / default settings

Author: gliljas

src/Microsoft.Azure.WebJobs.Extensions.Kafka/Config/KafkaOptions.cs

@@ -219,6 +219,22 @@ public int ChannelFullRetryIntervalInMs
             }
         }
 
+        /// <summary>
+        /// Default compression level parameter for algorithm selected by configuration property <see cref="CompressionType"/>
+        /// Can be overridden by setting <see cref="KafkaAttribute.CompressionLevel"/>
+        /// 
+        /// compression.level in librdkafka
+        /// </summary>
+        public int? CompressionLevel { get; set; }
+
+        /// <summary>
+        /// Default compression codec to use for compressing message sets.
+        /// Can be overridden by setting <see cref="KafkaAttribute.CompressionType"/>
+        /// 
+        /// compression.codec in librdkafka
+        /// </summary>
+        public MessageCompressionType? CompressionType { get; set; }
+
         public string Format()
         {
             var serializerSettings = new JsonSerializerSettings()

src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaAttribute.cs

@@ -15,7 +15,8 @@ namespace Microsoft.Azure.WebJobs.Extensions.Kafka
     [Binding]
     public sealed class KafkaAttribute : Attribute
     {
-        private int? compressionLevel;
+        internal int? DefinedCompressionLevel;
+        internal MessageCompressionType? DefinedCompressionType;
         private int? maxMessageBytes;
         private int? batchSize;
         private bool? enableIdempotence;
@@ -152,12 +153,12 @@ public KafkaAttribute()
         /// Compression level parameter for algorithm selected by configuration property <see cref="CompressionType"/>
         /// compression.level in librdkafka
         /// </summary>
-        public int CompressionLevel { get => compressionLevel.GetValueOrDefault(-1); set => compressionLevel = value; }
+        public int CompressionLevel { set => DefinedCompressionLevel = value; }
 
         /// <summary>
         /// Compression codec to use for compressing message sets. 
         /// compression.codec in librdkafka
         /// </summary>
-        public MessageCompressionType CompressionType { get; set; } = MessageCompressionType.NotSet;
+        public MessageCompressionType CompressionType { set => DefinedCompressionType = value; }
     }
 }

src/Microsoft.Azure.WebJobs.Extensions.Kafka/Output/KafkaProducerFactory.cs

@@ -131,12 +131,12 @@ public ProducerConfig GetProducerConfig(KafkaProducerEntity entity)
                 Debug = kafkaOptions?.LibkafkaDebug,
                 MetadataMaxAgeMs = kafkaOptions?.MetadataMaxAgeMs,
                 SocketKeepaliveEnable = kafkaOptions?.SocketKeepaliveEnable,
-                CompressionLevel = entity.Attribute.CompressionLevel 
+                CompressionLevel = entity.Attribute.DefinedCompressionLevel ?? kafkaOptions.CompressionLevel ?? -1
             };
 
-            if (entity.Attribute.CompressionType != MessageCompressionType.NotSet)
+            if ((entity.Attribute.DefinedCompressionType ?? kafkaOptions.CompressionType) != MessageCompressionType.NotSet)
             {
-                conf.CompressionType = (CompressionType)entity.Attribute.CompressionType;
+                conf.CompressionType = (CompressionType)(entity.Attribute.DefinedCompressionType ?? kafkaOptions.CompressionType);
             }
 
             if (entity.Attribute.AuthenticationMode != BrokerAuthenticationMode.NotSet)