Clear Event Hub Device Watermarks when source event hub changed

src/console/Startup.cs

@@ -106,8 +106,18 @@ public virtual StorageCheckpointClient ResolveCheckpointClient(IServiceProvider
             var checkpointBlobClient = factory.CreateStorageClient(checkpointContainerOptions);
             var logger = serviceProvider.GetRequiredService<ITelemetryLogger>();
 
-            storageOptions.BlobPrefix = applicationType;
-            var checkpointClient = new StorageCheckpointClient(checkpointBlobClient, storageOptions, logger);
+            var eventProcessorOptions = new EventHubClientOptions();
+            if (applicationType == _normalizationAppType)
+            {
+                Configuration.GetSection("InputEventHub").Bind(eventProcessorOptions);
+            }
+            else if (applicationType == _measurementToFhirAppType)
+            {
+                Configuration.GetSection("NormalizationEventHub").Bind(eventProcessorOptions);
+            }
+
+            storageOptions.BlobPrefix = $"{applicationType}/{storageOptions.BlobPrefix}";
+            var checkpointClient = new StorageCheckpointClient(checkpointBlobClient, storageOptions, eventProcessorOptions, logger);
             return checkpointClient;
         }
 

src/lib/Microsoft.Health.Events/EventCheckpointing/ICheckpointClient.cs

@@ -15,5 +15,7 @@ public interface ICheckpointClient
         Task PublishCheckpointAsync(string partitionId);
 
         Task<Checkpoint> GetCheckpointForPartitionAsync(string partitionId);
+
+        Task ResetCheckpointsAsync();
     }
 }

src/lib/Microsoft.Health.Events/EventCheckpointing/StorageCheckpointClient.cs

@@ -15,6 +15,7 @@
 using Azure.Storage.Blobs;
 using Azure.Storage.Blobs.Models;
 using EnsureThat;
+using Microsoft.Health.Events.Common;
 using Microsoft.Health.Events.Model;
 using Microsoft.Health.Events.Telemetry;
 using Microsoft.Health.Logging.Telemetry;
@@ -23,17 +24,25 @@ namespace Microsoft.Health.Events.EventCheckpointing
 {
     public class StorageCheckpointClient : ICheckpointClient
     {
-        private ConcurrentDictionary<string, Checkpoint> _checkpoints;
-        private ConcurrentDictionary<string, int> _lastCheckpointTracker;
-        private int _lastCheckpointMaxCount;
-        private BlobContainerClient _storageClient;
-        private ITelemetryLogger _log;
-
-        public StorageCheckpointClient(BlobContainerClient containerClient, StorageCheckpointOptions options, ITelemetryLogger log)
+        private readonly string _blobCheckpointPrefix;
+        private readonly string _blobPath;
+        private readonly ConcurrentDictionary<string, Checkpoint> _checkpoints;
+        private readonly int _lastCheckpointMaxCount;
+        private readonly ConcurrentDictionary<string, int> _lastCheckpointTracker;
+        private readonly ITelemetryLogger _log;
+        private readonly BlobContainerClient _storageClient;
+
+        public StorageCheckpointClient(BlobContainerClient containerClient, StorageCheckpointOptions options, EventHubClientOptions eventHubClientOptions, ITelemetryLogger log)
         {
             EnsureArg.IsNotNull(containerClient);
             EnsureArg.IsNotNull(options);
-            BlobPrefix = options.BlobPrefix;
+            EnsureArg.IsNotNull(eventHubClientOptions);
+
+            // Blob path for checkpoints includes the event hub name to scope the checkpoints per source event hub.
+            _blobCheckpointPrefix = $"{options.BlobPrefix}/checkpoint/";
+            _blobPath = eventHubClientOptions != default ?
+                $"{_blobCheckpointPrefix}{eventHubClientOptions.EventHubNamespaceFQDN}/{eventHubClientOptions.EventHubName}/" :
+                _blobCheckpointPrefix;
 
             _lastCheckpointMaxCount = int.Parse(options.CheckpointBatchCount);
             _checkpoints = new ConcurrentDictionary<string, Checkpoint>();
@@ -42,8 +51,6 @@ public StorageCheckpointClient(BlobContainerClient containerClient, StorageCheck
             _log = log;
         }
 
-        public string BlobPrefix { get; }
-
         public BlobContainerClient GetBlobContainerClient()
         {
             return _storageClient;
@@ -55,7 +62,7 @@ public async Task UpdateCheckpointAsync(Checkpoint checkpoint)
             EnsureArg.IsNotNullOrWhiteSpace(checkpoint.Id);
             var lastProcessed = EnsureArg.IsNotNullOrWhiteSpace(checkpoint.LastProcessed.DateTime.ToString("MM/dd/yyyy hh:mm:ss.fff tt"));
 
-            var blobName = $"{BlobPrefix}/checkpoint/{checkpoint.Id}";
+            var blobName = $"{checkpoint.Prefix}{checkpoint.Id}";
             var blobClient = _storageClient.GetBlobClient(blobName);
 
             var metadata = new Dictionary<string, string>()
@@ -87,7 +94,7 @@ public async Task UpdateCheckpointAsync(Checkpoint checkpoint)
 
         public Task<Checkpoint> GetCheckpointForPartitionAsync(string partitionIdentifier)
         {
-            var prefix = $"{BlobPrefix}/checkpoint/{partitionIdentifier}";
+            var prefix = $"{_blobPath}{partitionIdentifier}";
 
             Task<Checkpoint> GetCheckpointAsync()
             {
@@ -103,7 +110,7 @@ Task<Checkpoint> GetCheckpointAsync()
                         DateTimeOffset.TryParse(str, null, DateTimeStyles.AssumeUniversal, out lastEventTimestamp);
                     }
 
-                    checkpoint.Prefix = BlobPrefix;
+                    checkpoint.Prefix = _blobPath;
                     checkpoint.Id = partitionId;
                     checkpoint.LastProcessed = lastEventTimestamp;
                 }
@@ -136,7 +143,7 @@ public async Task SetCheckpointAsync(IEventMessage eventArgs)
                 var checkpoint = new Checkpoint();
                 checkpoint.LastProcessed = eventArgs.EnqueuedTime;
                 checkpoint.Id = partitionId;
-                checkpoint.Prefix = BlobPrefix;
+                checkpoint.Prefix = _blobPath;
 
                 _checkpoints[partitionId] = checkpoint;
                 var count = _lastCheckpointTracker.AddOrUpdate(partitionId, 1, (key, value) => value + 1);
@@ -161,5 +168,28 @@ public async Task PublishCheckpointAsync(string partitionId)
             Checkpoint checkpoint = _checkpoints[partitionId];
             await UpdateCheckpointAsync(checkpoint);
         }
+
+        /// <summary>
+        /// Deletes the previously recorded checkpoints if the current checkpoint blob path (corresponding to the input event hub) has changed.
+        /// </summary>
+        public async Task ResetCheckpointsAsync()
+        {
+            try
+            {
+                foreach (BlobItem blob in _storageClient.GetBlobs(states: BlobStates.All, prefix: _blobCheckpointPrefix, cancellationToken: CancellationToken.None))
+                {
+                    if (!blob.Name.Contains(_blobPath, StringComparison.OrdinalIgnoreCase))
+                    {
+                        await _storageClient.DeleteBlobAsync(blob.Name, cancellationToken: CancellationToken.None);
+                    }
+                }
+            }
+#pragma warning disable CA1031
+            catch (Exception ex)
+#pragma warning restore CA1031
+            {
+                _log.LogError(new Exception($"Unable to reset checkpoints. {ex.Message}"));
+            }
+        }
     }
 }

src/lib/Microsoft.Health.Events/EventHubProcessor/EventProcessor.cs

@@ -36,6 +36,9 @@ public async Task RunAsync(EventProcessorClient processor, CancellationToken ct)
         {
             EnsureArg.IsNotNull(processor);
 
+            // Reset previous checkpoints corresponding to an older source event hub (i.e. applicable if the source event hub changes)
+            await _checkpointClient.ResetCheckpointsAsync();
+
             // Processes two types of events
             // 1) Event hub events
             // 2) Maximum wait events. These are generated when we have not received an event hub

test/Microsoft.Health.Events.UnitTest/StorageCheckpointTests.cs

@@ -0,0 +1,174 @@
+using Azure;
+using Azure.Storage.Blobs;
+using Azure.Storage.Blobs.Models;
+using Microsoft.Health.Events.Common;
+using Microsoft.Health.Events.EventCheckpointing;
+using Microsoft.Health.Logging.Telemetry;
+using NSubstitute;
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Microsoft.Health.Events.UnitTest
+{
+    public class StorageCheckpointTests
+    {
+        private readonly BlobContainerClient _blobContainerClient;
+        private readonly EventHubClientOptions _eventHubClientOptions;
+        private readonly ITelemetryLogger _logger;
+        private readonly StorageCheckpointOptions _storageCheckpointOptions;
+
+        private readonly string _blobCheckpointPrefix;
+        private readonly string _blobPath;
+        private readonly string _eventHubName;
+        private readonly string _eventHubNamespaceFQDN;
+
+        public StorageCheckpointTests()
+        {
+            _blobContainerClient = Substitute.For<BlobContainerClient>();
+
+            _storageCheckpointOptions = new StorageCheckpointOptions()
+            {
+                BlobPrefix = "Normalization",
+                CheckpointBatchCount = "5"
+            };
+
+            _eventHubClientOptions = new EventHubClientOptions();
+            _eventHubNamespaceFQDN = "test.servicebus.windows.net";
+            _eventHubName = "devicedata";
+
+            // Blob path corresponds to current event hub name
+            _blobCheckpointPrefix = $"{_storageCheckpointOptions.BlobPrefix}/checkpoint/";
+            _blobPath = $"{_blobCheckpointPrefix}{_eventHubNamespaceFQDN}/{_eventHubName}/";
+
+            IReadOnlyList<BlobItem> mockBlobItems = new List<BlobItem>()
+            {
+                BlobsModelFactory.BlobItem(name: $"{_blobPath}1"),
+                BlobsModelFactory.BlobItem(name: $"{_blobPath}10"),
+                BlobsModelFactory.BlobItem(name: $"{_blobPath}20")
+            };
+
+            var mockPageBlobItems = Page<BlobItem>.FromValues(mockBlobItems, "continuationToken", Substitute.For<Response>());
+            var mockPageableBlobItems = Pageable<BlobItem>.FromPages(new[] { mockPageBlobItems });
+
+            _blobContainerClient.GetBlobs(states: BlobStates.All, prefix: _blobCheckpointPrefix, cancellationToken: CancellationToken.None)
+                .Returns(mockPageableBlobItems);
+
+            _logger = Substitute.For<ITelemetryLogger>();
+        }
+
+        [Fact]
+        public void GivenNullParameters_WhenStorageCheckpointClientCreated_Throws()
+        {
+            Assert.Throws<ArgumentNullException>(() => new StorageCheckpointClient(null, _storageCheckpointOptions, _eventHubClientOptions, _logger));
+            Assert.Throws<ArgumentNullException>(() => new StorageCheckpointClient(_blobContainerClient, null, _eventHubClientOptions, _logger));
+            Assert.Throws<ArgumentNullException>(() => new StorageCheckpointClient(_blobContainerClient, _storageCheckpointOptions, null, _logger));
+        }
+
+        [Fact]
+        public async Task GivenUnchangedEventHubOptions_WhenResetCheckpointsAsyncCalled_ThenNoCheckpointsAreDeleted()
+        {
+            _eventHubClientOptions.EventHubNamespaceFQDN = _eventHubNamespaceFQDN;
+            _eventHubClientOptions.EventHubName = _eventHubName;
+
+            var storageClient = new StorageCheckpointClient(_blobContainerClient, _storageCheckpointOptions, _eventHubClientOptions, _logger);
+            await storageClient.ResetCheckpointsAsync();
+
+            // Given that the source event hub didn't change, verify that no checkpoint deletions occured.
+            _blobContainerClient.Received(1).GetBlobs(states: BlobStates.All, prefix: _blobCheckpointPrefix, cancellationToken: CancellationToken.None);
+            await _blobContainerClient.ReceivedWithAnyArgs(0).DeleteBlobAsync(null);
+        }
+
+        [Fact]
+        public async Task GivenUpdatedEventHubNamespace_WhenResetCheckpointsAsyncCalled_ThenPreviousCheckpointsAreDeleted()
+        {
+            _eventHubClientOptions.EventHubNamespaceFQDN = "newtest.servicebus.windows.net";
+            _eventHubClientOptions.EventHubName = _eventHubName;
+
+            var storageClient = new StorageCheckpointClient(_blobContainerClient, _storageCheckpointOptions, _eventHubClientOptions, _logger);
+            await storageClient.ResetCheckpointsAsync();
+
+            // Given that the event hub namespace changed and is therefore a new source, verify that the checkpoints corresponding to the old source will be deleted.
+            _blobContainerClient.Received(1).GetBlobs(states: BlobStates.All, prefix: _blobCheckpointPrefix, cancellationToken: CancellationToken.None);
+            await _blobContainerClient.ReceivedWithAnyArgs(3).DeleteBlobAsync(null);
+            await _blobContainerClient.Received(1).DeleteBlobAsync($"{_blobPath}1");
+            await _blobContainerClient.Received(1).DeleteBlobAsync($"{_blobPath}10");
+            await _blobContainerClient.Received(1).DeleteBlobAsync($"{_blobPath}20");
+        }
+
+        [Fact]
+        public async Task GivenUpdatedEventHubName_WhenResetCheckpointsAsyncCalled_ThenPreviousCheckpointsAreDeleted()
+        {
+            _eventHubClientOptions.EventHubNamespaceFQDN = _eventHubNamespaceFQDN;
+            _eventHubClientOptions.EventHubName = "newdevicedata";
+
+            var storageClient = new StorageCheckpointClient(_blobContainerClient, _storageCheckpointOptions, _eventHubClientOptions, _logger);
+            await storageClient.ResetCheckpointsAsync();
+
+            // Given that the event hub changed and is therefore a new source, verify that the checkpoints corresponding to the old source will be deleted.
+            _blobContainerClient.Received(1).GetBlobs(states: BlobStates.All, prefix: _blobCheckpointPrefix, cancellationToken: CancellationToken.None);
+            await _blobContainerClient.ReceivedWithAnyArgs(3).DeleteBlobAsync(null);
+            await _blobContainerClient.Received(1).DeleteBlobAsync($"{_blobPath}1");
+            await _blobContainerClient.Received(1).DeleteBlobAsync($"{_blobPath}10");
+            await _blobContainerClient.Received(1).DeleteBlobAsync($"{_blobPath}20");
+        }
+
+        [Fact]
+        public async Task GivenUpdatedEventHubNamespaceAndEventHubName_WhenResetCheckpointsAsyncCalled_ThenPreviousCheckpointsAreDeleted()
+        {
+            _eventHubClientOptions.EventHubNamespaceFQDN = "newtest.servicebus.windows.net";
+            _eventHubClientOptions.EventHubName = "newdevicedata";
+
+            var storageClient = new StorageCheckpointClient(_blobContainerClient, _storageCheckpointOptions, _eventHubClientOptions, _logger);
+            await storageClient.ResetCheckpointsAsync();
+
+            // Given that the event hub namespace and the event hub name changed and is therefore a new source, verify that the checkpoints corresponding to the old source will be deleted.
+            _blobContainerClient.Received(1).GetBlobs(states: BlobStates.All, prefix: _blobCheckpointPrefix, cancellationToken: CancellationToken.None);
+            await _blobContainerClient.ReceivedWithAnyArgs(3).DeleteBlobAsync(null);
+            await _blobContainerClient.Received(1).DeleteBlobAsync($"{_blobPath}1");
+            await _blobContainerClient.Received(1).DeleteBlobAsync($"{_blobPath}10");
+            await _blobContainerClient.Received(1).DeleteBlobAsync($"{_blobPath}20");
+        }
+
+        [Fact]
+        public async Task GivenDifferentAppType_WhenResetCheckpointsAsyncCalled_ThenCheckpointsOfOtherAppsAreNotDeleted()
+        {
+            _eventHubClientOptions.EventHubNamespaceFQDN = _eventHubNamespaceFQDN;
+            _eventHubClientOptions.EventHubName = "newdevicedata";
+            _storageCheckpointOptions.BlobPrefix = "MeasurementToFhir";
+            var fhirconvBlobCheckpointPrefix = $"{_storageCheckpointOptions.BlobPrefix }/checkpoint/";
+            var fhirconvBlobPath = $"{fhirconvBlobCheckpointPrefix}{_eventHubNamespaceFQDN}/{_eventHubName}/";
+
+            IReadOnlyList<BlobItem> mockBlobItems = new List<BlobItem>()
+            {
+                BlobsModelFactory.BlobItem(name: $"{fhirconvBlobPath}1"),
+                BlobsModelFactory.BlobItem(name: $"{fhirconvBlobPath}10"),
+                BlobsModelFactory.BlobItem(name: $"{fhirconvBlobPath}20")
+            };
+
+            var mockPageBlobItems = Page<BlobItem>.FromValues(mockBlobItems, "continuationToken", Substitute.For<Response>());
+            var mockPageableBlobItems = Pageable<BlobItem>.FromPages(new[] { mockPageBlobItems });
+
+            _blobContainerClient.GetBlobs(states: BlobStates.All, prefix: fhirconvBlobCheckpointPrefix, cancellationToken: CancellationToken.None)
+                .Returns(mockPageableBlobItems);
+
+            var storageClient = new StorageCheckpointClient(_blobContainerClient, _storageCheckpointOptions, _eventHubClientOptions, _logger);
+            await storageClient.ResetCheckpointsAsync();
+
+            // Given that we are processing events for a different app type and the source changed, verify that the previous checkpoints corresponding to this app are deleted.
+            _blobContainerClient.Received(1).GetBlobs(states: BlobStates.All, prefix: fhirconvBlobCheckpointPrefix, cancellationToken: CancellationToken.None);
+            await _blobContainerClient.ReceivedWithAnyArgs(3).DeleteBlobAsync(null);
+            await _blobContainerClient.Received(1).DeleteBlobAsync($"{fhirconvBlobPath}1");
+            await _blobContainerClient.Received(1).DeleteBlobAsync($"{fhirconvBlobPath}10");
+            await _blobContainerClient.Received(1).DeleteBlobAsync($"{fhirconvBlobPath}20");
+
+            // Given that we are processing events for a different app type, verify that the checkpoints corresponding to the other apps are not deleted.
+            _blobContainerClient.Received(0).GetBlobs(states: BlobStates.All, prefix: _blobCheckpointPrefix, cancellationToken: CancellationToken.None);
+            await _blobContainerClient.Received(0).DeleteBlobAsync($"{_blobPath}1");
+            await _blobContainerClient.Received(0).DeleteBlobAsync($"{_blobPath}10");
+            await _blobContainerClient.Received(0).DeleteBlobAsync($"{_blobPath}20");
+        }
+    }
+}