Merge branch 'dev' into SetStorageServerWaitTime

src/Microsoft.Azure.EventHubs.Processor/Microsoft.Azure.EventHubs.Processor.csproj

@@ -30,6 +30,15 @@
     <Copyright>© Microsoft Corporation. All rights reserved.</Copyright>
   </PropertyGroup>
 
+  <ItemGroup>
+    <ProjectReference Include="..\Microsoft.Azure.EventHubs\Microsoft.Azure.EventHubs.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Azure.KeyVault.Core" Version="3.0.3" />
+    <PackageReference Include="Newtonsoft.Json" Version="10.0.3" />
+  </ItemGroup>
+
   <PropertyGroup Condition="'$(TargetFramework)' == 'uap10.0'">
     <NugetTargetMoniker>UAP,Version=v10.0</NugetTargetMoniker>
     <TargetPlatformIdentifier>UAP</TargetPlatformIdentifier>
@@ -39,22 +48,19 @@
     <TargetFrameworkVersion>v5.0</TargetFrameworkVersion>
   </PropertyGroup>
 
-  <ItemGroup>
-    <ProjectReference Include="..\Microsoft.Azure.EventHubs\Microsoft.Azure.EventHubs.csproj" />
-  </ItemGroup>
-
-  <ItemGroup>
-    <PackageReference Include="Newtonsoft.Json" Version="10.0.3" />
-    <PackageReference Include="WindowsAzure.Storage" Version="9.3.3" />
-  </ItemGroup>
-
   <ItemGroup Condition=" '$(TargetFramework)' == 'net461' ">
     <Reference Include="System" />
     <Reference Include="Microsoft.CSharp" />
+    <PackageReference Include="WindowsAzure.Storage" Version="9.3.3" />
+  </ItemGroup>
+
+  <ItemGroup Condition=" '$(TargetFramework)' == 'netstandard2.0' ">
+    <PackageReference Include="Microsoft.Azure.Storage.Blob" Version="9.4.2" />
   </ItemGroup>
 
   <ItemGroup Condition=" '$(TargetFramework)' == 'uap10.0' ">
     <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform " Version="5.2.3" />
+    <PackageReference Include="Microsoft.Azure.Storage.Blob" Version="9.4.2" />
   </ItemGroup>
 
 </Project>

src/Microsoft.Azure.EventHubs.Processor/PartitionManager.cs

@@ -296,7 +296,7 @@ namespace Microsoft.Azure.EventHubs.Processor
 
                     // Check any expired leases that we can grab here.
                     var checkLeaseTasks = new List<Task>();
-                    foreach (var possibleLease in allLeases.Values.Where(lease => lease.Owner != this.host.HostName))
+                    foreach (var possibleLease in allLeases.Values)
                     {
                         var subjectLease = possibleLease;
 

test/Microsoft.Azure.EventHubs.Tests/Processor/ProcessorTestBase.cs

@@ -955,7 +955,7 @@ namespace Microsoft.Azure.EventHubs.Tests.Processor
 
         [Fact]
         [DisplayTestMethodName]
-        async Task ReRegisterEventProcessor()
+        async Task ReRegister()
         {
             var eventProcessorHost = new EventProcessorHost(
                 null, // Entity path will be picked from connection string.
@@ -978,6 +978,35 @@ namespace Microsoft.Azure.EventHubs.Tests.Processor
             await RunGenericScenario(eventProcessorHost, epo);
         }
 
+        [Fact]
+        [DisplayTestMethodName]
+        async Task ReRegisterAfterLeaseExpiry()
+        {
+            var hostName = Guid.NewGuid().ToString();
+
+            var processorOptions = new EventProcessorOptions
+            {
+                InitialOffsetProvider = pId => EventPosition.FromEnd()
+            };
+
+            var eventProcessorHost = new EventProcessorHost(
+                hostName,
+                null, // Entity path will be picked from connection string.
+                PartitionReceiver.DefaultConsumerGroupName,
+                TestUtility.EventHubsConnectionString,
+                TestUtility.StorageConnectionString,
+                Guid.NewGuid().ToString());
+
+            var runResult = await RunGenericScenario(eventProcessorHost, processorOptions);
+            Assert.False(runResult.ReceivedEvents.Any(kvp => kvp.Value.Count != 1), "First host: One of the partitions didn't return exactly 1 event");
+
+            // Allow sometime so that leases can expire.
+            await Task.Delay(60);
+
+            runResult = await RunGenericScenario(eventProcessorHost);
+            Assert.False(runResult.ReceivedEvents.Any(kvp => kvp.Value.Count != 1), "Second host: One of the partitions didn't return exactly 1 event");
+        }
+
         async Task<Dictionary<string, Tuple<string, DateTime>>> DiscoverEndOfStream()
         {
             var ehClient = EventHubClient.CreateFromConnectionString(TestUtility.EventHubsConnectionString);