Exposing AutoOffsetReset property in KafkaOptions (#364)

* Exposing AutoOffsetReset property in KafkaOptions

* Adding internal autoOffsetReset field

* Reverting Changes to sln file

Author: jainharsh98

README.md

@@ -319,11 +319,13 @@ The settings exposed here are targeted to more advanced users that want to custo
 |MaxPartitionFetchBytes|max.partition.fetch.bytes|Trigger
 |FetchMaxBytes|fetch.max.bytes|Trigger
 |AutoCommitIntervalMs|auto.commit.interval.ms|Trigger
+|AutoOffsetReset|auto.offset.reset|Trigger
 |LibkafkaDebug|debug|Both
 |MetadataMaxAgeMs|metadata.max.age.ms|Both
 |SocketKeepaliveEnable|socket.keepalive.enable|Both
 
 **NOTE:** `MetadataMaxAgeMs` default is `180000` `SocketKeepaliveEnable` default is `true` otherwise, the default value is the same as the [Configuration properties](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). The reason of the default settings, refer to this [issue](https://github.com/Azure/azure-functions-kafka-extension/issues/187).
+**NOTE:** `AutoOffsetReset` default is Earliest. Allowed Values are `Earliest` and `Latest`.
 
 If you are missing an configuration setting please create an issue and describe why you need it.
 

docs/public-documentation/Trigger.md

@@ -691,6 +691,7 @@ The settings exposed here are to customize how librdkafka works. [Librdkafka Doc
 |MaxPartitionFetchBytes|max.partition.fetch.bytes|
 |FetchMaxBytes|fetch.max.bytes|
 |AutoCommitIntervalMs|auto.commit.interval.ms|
+|AutoOffsetReset|auto.offset.reset|
 |LibkafkaDebug|debug|
 |MetadataMaxAgeMs|metadata.max.age.ms|
 |SocketKeepaliveEnable|socket.keepalive.enable|

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

@@ -136,6 +136,19 @@ public int MaxBatchSize
         /// <value>The auto commit interval ms.</value>
         public int AutoCommitIntervalMs { get; set; } = 200;
 
+        AutoOffsetReset autoOffsetReset = AutoOffsetReset.Earliest;
+
+        /// <summary>
+        /// Ges or sets the AutoOffsetReset option for librdkafka library.
+        /// default: Earliest
+        /// Allowed Values: Latest, Earliest
+        /// Librdkafka: auto.commit.offset
+        /// </summary>
+        public AutoOffsetReset AutoOffsetReset {
+            get => this.autoOffsetReset;
+            set => this.autoOffsetReset = AutoOffsetReset.Latest.Equals(value) ? AutoOffsetReset.Latest : AutoOffsetReset.Earliest; 
+        }
+
         /// <summary>
         /// Gets or sets the debug option for librdkafka library.
         /// Default = "" (disable)

src/Microsoft.Azure.WebJobs.Extensions.Kafka/Listeners/KafkaListener.cs

@@ -164,7 +164,7 @@ private ConsumerConfig GetConsumerConfiguration()
                 Debug = this.options.LibkafkaDebug,
 
                 // start from earliest if no checkpoint has been committed
-                AutoOffsetReset = AutoOffsetReset.Earliest,
+                AutoOffsetReset = this.options.AutoOffsetReset,
 
                 // Secure communication/authentication
                 SaslMechanism = this.listenerConfiguration.SaslMechanism,

test/Microsoft.Azure.WebJobs.Extensions.Kafka.UnitTests/KafkaListenerTest.cs

@@ -350,11 +350,65 @@ public async Task When_Options_Are_Set_Should_Be_Set_In_Consumer_Config()
             Assert.Equal("path/to/key", target.ConsumerConfig.SslKeyLocation);
             Assert.Equal("path/to/cacert", target.ConsumerConfig.SslCaLocation);
             Assert.Equal(kafkaOptions.AutoCommitIntervalMs, target.ConsumerConfig.AutoCommitIntervalMs);
+            Assert.Equal(AutoOffsetReset.Earliest, target.ConsumerConfig.AutoOffsetReset);
+            Assert.Equal(true, target.ConsumerConfig.EnableAutoCommit);
+            Assert.Equal(false, target.ConsumerConfig.EnableAutoOffsetStore);
+            Assert.Equal(180000, target.ConsumerConfig.MetadataMaxAgeMs);
+            Assert.Equal(true, target.ConsumerConfig.SocketKeepaliveEnable);
+
+            await target.StopAsync(default);
+        }
+
+        [Fact]
+        public async Task When_Options_With_AutoOffsetReset_Latest_Are_Set_Should_Be_Set_In_Consumer_Config()
+        {
+            var executor = new Mock<ITriggeredFunctionExecutor>();
+            var consumer = new Mock<IConsumer<Ignore, string>>();
+
+            var listenerConfig = new KafkaListenerConfiguration()
+            {
+                BrokerList = "testBroker",
+                Topic = "topic",
+                ConsumerGroup = "group1",
+                SslKeyPassword = "password1",
+                SslCertificateLocation = "path/to/cert",
+                SslKeyLocation = "path/to/key",
+                SslCaLocation = "path/to/cacert"
+            };
+
+            var kafkaOptions = new KafkaOptions()
+            {
+                AutoOffsetReset = AutoOffsetReset.Latest
+            };
+
+            var target = new KafkaListenerForTest<Ignore, string>(
+                executor.Object,
+                true,
+                kafkaOptions,
+                listenerConfig,
+                requiresKey: true,
+                valueDeserializer: null,
+                NullLogger.Instance,
+                functionId: "testId"
+                );
+
+            target.SetConsumer(consumer.Object);
+
+            await target.StartAsync(default);
+
+            Assert.Equal(12, target.ConsumerConfig.Count());
+            Assert.Equal("testBroker", target.ConsumerConfig.BootstrapServers);
+            Assert.Equal("group1", target.ConsumerConfig.GroupId);
+            Assert.Equal("password1", target.ConsumerConfig.SslKeyPassword);
+            Assert.Equal("path/to/cert", target.ConsumerConfig.SslCertificateLocation);
+            Assert.Equal("path/to/key", target.ConsumerConfig.SslKeyLocation);
+            Assert.Equal("path/to/cacert", target.ConsumerConfig.SslCaLocation);
+            Assert.Equal(kafkaOptions.AutoCommitIntervalMs, target.ConsumerConfig.AutoCommitIntervalMs);
+            Assert.Equal(AutoOffsetReset.Latest, target.ConsumerConfig.AutoOffsetReset);
             Assert.Equal(true, target.ConsumerConfig.EnableAutoCommit);
             Assert.Equal(false, target.ConsumerConfig.EnableAutoOffsetStore);
             Assert.Equal(180000, target.ConsumerConfig.MetadataMaxAgeMs);
             Assert.Equal(true, target.ConsumerConfig.SocketKeepaliveEnable);
-            Assert.Equal(AutoOffsetReset.Earliest, target.ConsumerConfig.AutoOffsetReset);
 
             await target.StopAsync(default);
         }