Use PostConfigure for InitialOffsetOptions (Azure#17778)

PostConfigure ensures that the InitialOffsetOptions will be propagated to the EventProcessorOptions regardless of the ordering used when doing configuration.

Author: JoshLove-msft

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Config/EventHubWebJobsBuilderExtensions.cs

@@ -86,7 +86,7 @@ public static IWebJobsBuilder AddEventHubs(this IWebJobsBuilder builder, Action<
             builder.Services.AddAzureClientsCore();
             builder.Services.AddSingleton<EventHubClientFactory>();
             builder.Services.Configure<EventHubOptions>(configure);
-            builder.Services.Configure<EventHubOptions>(ConfigureInitialOffsetOptions);
+            builder.Services.PostConfigure<EventHubOptions>(ConfigureInitialOffsetOptions);
 
             return builder;
         }

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/EventHubTests.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Globalization;
 using System.Text;
 using Azure.Messaging.EventHubs;
 using Azure.Messaging.EventHubs.Consumer;
@@ -202,7 +203,8 @@ public void InitializeFromHostMetadata()
                         { "AzureWebJobs:extensions:EventHubs:EventProcessorOptions:PrefetchCount", "200" },
                         { "AzureWebJobs:extensions:EventHubs:BatchCheckpointFrequency", "5" },
                         { "AzureWebJobs:extensions:EventHubs:PartitionManagerOptions:LeaseDuration", "00:00:31" },
-                        { "AzureWebJobs:extensions:EventHubs:PartitionManagerOptions:RenewInterval", "00:00:21" }
+                        { "AzureWebJobs:extensions:EventHubs:PartitionManagerOptions:RenewInterval", "00:00:21" },
+                        { "AzureWebJobs:extensions:EventHubs:InitialOffsetOptions:Type", "FromEnd" },
                     });
                 })
                 .Build();
@@ -217,6 +219,85 @@ public void InitializeFromHostMetadata()
             Assert.AreEqual(100, options.MaxBatchSize);
             Assert.AreEqual(31, options.EventProcessorOptions.PartitionOwnershipExpirationInterval.TotalSeconds);
             Assert.AreEqual(21, options.EventProcessorOptions.LoadBalancingUpdateInterval.TotalSeconds);
+            Assert.AreEqual(EventPosition.Latest, eventProcessorOptions.DefaultStartingPosition);
+        }
+
+        [Test]
+        public void InitializeFromCodeRespectsFinalOffsetOptions_FromStart()
+        {
+            IHost host = new HostBuilder()
+                .ConfigureDefaultTestHost(builder =>
+                {
+                    builder.AddEventHubs(options => options.InitialOffsetOptions.Type = "FromEnd");
+                })
+                .ConfigureServices(services =>
+                    {
+                        services.Configure<EventHubOptions>(options =>
+                        {
+                            options.InitialOffsetOptions.Type = "FromStart";
+                        });
+                    })
+                .Build();
+
+            // Force the ExtensionRegistryFactory to run, which will initialize the EventHubConfiguration.
+            var extensionRegistry = host.Services.GetService<IExtensionRegistry>();
+            var options = host.Services.GetService<IOptions<EventHubOptions>>().Value;
+
+            var eventProcessorOptions = options.EventProcessorOptions;
+            Assert.AreEqual(EventPosition.Earliest, eventProcessorOptions.DefaultStartingPosition);
+        }
+
+        [Test]
+        public void InitializeFromCodeRespectsFinalOffsetOptions_FromEnd()
+        {
+            var host = new HostBuilder()
+                 .ConfigureDefaultTestHost(builder =>
+                 {
+                     builder.AddEventHubs(options => options.InitialOffsetOptions.Type = "FromStart");
+                 })
+                 .ConfigureServices(services =>
+                 {
+                     services.Configure<EventHubOptions>(options =>
+                     {
+                         options.InitialOffsetOptions.Type = "FromEnd";
+                     });
+                 })
+                 .Build();
+
+            // Force the ExtensionRegistryFactory to run, which will initialize the EventHubConfiguration.
+            var extensionRegistry = host.Services.GetService<IExtensionRegistry>();
+            var options = host.Services.GetService<IOptions<EventHubOptions>>().Value;
+
+            var eventProcessorOptions = options.EventProcessorOptions;
+            Assert.AreEqual(EventPosition.Latest, eventProcessorOptions.DefaultStartingPosition);
+        }
+
+        [Test]
+        public void InitializeFromCodeRespectsFinalOffsetOptions_FromEnqueuedTime()
+        {
+            var host = new HostBuilder()
+                 .ConfigureDefaultTestHost(builder =>
+                 {
+                     builder.AddEventHubs(options => options.InitialOffsetOptions.Type = "FromStart");
+                 })
+                 .ConfigureServices(services =>
+                 {
+                     services.Configure<EventHubOptions>(options =>
+                     {
+                         options.InitialOffsetOptions.Type = "FromEnqueuedTime";
+                         options.InitialOffsetOptions.EnqueuedTimeUTC = DateTimeOffset.UtcNow.ToString();
+                     });
+                 })
+                 .Build();
+
+            // Force the ExtensionRegistryFactory to run, which will initialize the EventHubConfiguration.
+            var extensionRegistry = host.Services.GetService<IExtensionRegistry>();
+            var options = host.Services.GetService<IOptions<EventHubOptions>>().Value;
+
+            var eventProcessorOptions = options.EventProcessorOptions;
+            Assert.AreEqual(
+                EventPosition.FromEnqueuedTime(DateTime.Parse(options.InitialOffsetOptions.EnqueuedTimeUTC,
+                CultureInfo.InvariantCulture).ToUniversalTime()), eventProcessorOptions.DefaultStartingPosition);
         }
 
         internal static EventProcessorHostPartition GetPartitionContext(string partitionId = "0", string eventHubPath = "path",

sdk/eventhub/test-resources-post.ps1

@@ -0,0 +1,16 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+# The purpose of this script is to add a small delay between the creation of the live test resources
+# and the execution of the live tests. This allows RBAC to replicate and avoids flakiness in the first set 
+# of live tests that might otherwise start running before RBAC has replicated.
+
+param (
+    [hashtable] $DeploymentOutputs,
+    [string] $TenantId,
+    [string] $TestApplicationId,
+    [string] $TestApplicationSecret
+)
+
+Write-Verbose "Sleeping for 60 seconds to let RBAC replicate"
+Start-Sleep -s 60