| | | 1 | | // Copyright (c) Microsoft. All rights reserved. |
| | | 2 | | // Licensed under the MIT license. See LICENSE file in the project root for full license information. |
| | | 3 | | |
| | | 4 | | namespace Microsoft.Azure.EventHubs.Processor |
| | | 5 | | { |
| | | 6 | | using System; |
| | | 7 | | using System.Collections.Generic; |
| | | 8 | | using System.Linq; |
| | | 9 | | using System.Threading; |
| | | 10 | | using System.Threading.Tasks; |
| | | 11 | | using Microsoft.Azure.EventHubs.Primitives; |
| | | 12 | | using Microsoft.Azure.Storage; |
| | | 13 | | using Microsoft.Azure.Storage.Blob; |
| | | 14 | | using Newtonsoft.Json; |
| | | 15 | | |
| | | 16 | | class AzureStorageCheckpointLeaseManager : ICheckpointManager, ILeaseManager |
| | | 17 | | { |
| | 0 | 18 | | static string MetaDataOwnerName = "OWNINGHOST"; |
| | | 19 | | |
| | | 20 | | EventProcessorHost host; |
| | | 21 | | TimeSpan leaseDuration; |
| | | 22 | | TimeSpan leaseRenewInterval; |
| | | 23 | | |
| | 0 | 24 | | static readonly TimeSpan storageMaximumExecutionTime = TimeSpan.FromMinutes(2); |
| | | 25 | | readonly CloudStorageAccount cloudStorageAccount; |
| | | 26 | | readonly string leaseContainerName; |
| | | 27 | | readonly string storageBlobPrefix; |
| | | 28 | | BlobRequestOptions defaultRequestOptions; |
| | | 29 | | OperationContext operationContext = null; |
| | | 30 | | CloudBlobContainer eventHubContainer; |
| | | 31 | | CloudBlobDirectory consumerGroupDirectory; |
| | | 32 | | |
| | | 33 | | internal AzureStorageCheckpointLeaseManager(string storageConnectionString, string leaseContainerName, string st |
| | 0 | 34 | | : this(CloudStorageAccount.Parse(storageConnectionString), leaseContainerName, storageBlobPrefix) |
| | | 35 | | { |
| | 0 | 36 | | } |
| | | 37 | | |
| | 0 | 38 | | internal AzureStorageCheckpointLeaseManager(CloudStorageAccount cloudStorageAccount, string leaseContainerName, |
| | | 39 | | { |
| | 0 | 40 | | Guard.ArgumentNotNull(nameof(cloudStorageAccount), cloudStorageAccount); |
| | | 41 | | |
| | | 42 | | try |
| | | 43 | | { |
| | 0 | 44 | | NameValidator.ValidateContainerName(leaseContainerName); |
| | 0 | 45 | | } |
| | 0 | 46 | | catch (ArgumentException) |
| | | 47 | | { |
| | 0 | 48 | | throw new ArgumentException( |
| | 0 | 49 | | "Azure Storage lease container name is invalid. Please check naming conventions at https://msdn.micr |
| | 0 | 50 | | nameof(leaseContainerName)); |
| | | 51 | | } |
| | | 52 | | |
| | 0 | 53 | | this.cloudStorageAccount = cloudStorageAccount; |
| | 0 | 54 | | this.leaseContainerName = leaseContainerName; |
| | | 55 | | |
| | | 56 | | // Convert all-whitespace prefix to empty string. Convert null prefix to empty string. |
| | | 57 | | // Then the rest of the code only has one case to worry about. |
| | 0 | 58 | | this.storageBlobPrefix = (storageBlobPrefix != null) ? storageBlobPrefix.Trim() : ""; |
| | 0 | 59 | | } |
| | | 60 | | |
| | | 61 | | // The EventProcessorHost can't pass itself to the AzureStorageCheckpointLeaseManager constructor |
| | | 62 | | // because it is still being constructed. Do other initialization here also because it might throw and |
| | | 63 | | // hence we don't want it in the constructor. |
| | | 64 | | internal void Initialize(EventProcessorHost host) |
| | | 65 | | { |
| | 0 | 66 | | this.host = host; |
| | | 67 | | |
| | | 68 | | // Assign partition manager options. |
| | 0 | 69 | | this.leaseDuration = host.PartitionManagerOptions.LeaseDuration; |
| | 0 | 70 | | this.leaseRenewInterval = host.PartitionManagerOptions.RenewInterval; |
| | | 71 | | |
| | | 72 | | // Create storage default request options. |
| | 0 | 73 | | this.defaultRequestOptions = new BlobRequestOptions() |
| | 0 | 74 | | { |
| | 0 | 75 | | // Gets or sets the server timeout interval for a single HTTP request. |
| | 0 | 76 | | ServerTimeout = TimeSpan.FromSeconds(this.LeaseRenewInterval.TotalSeconds / 2), |
| | 0 | 77 | | |
| | 0 | 78 | | // Gets or sets the maximum execution time across all potential retries for the request. |
| | 0 | 79 | | MaximumExecutionTime = TimeSpan.FromSeconds(this.LeaseRenewInterval.TotalSeconds) |
| | 0 | 80 | | }; |
| | | 81 | | |
| | | 82 | | #if FullNetFx |
| | | 83 | | // Proxy enabled? |
| | | 84 | | if (this.host.EventProcessorOptions != null && this.host.EventProcessorOptions.WebProxy != null) |
| | | 85 | | { |
| | | 86 | | this.operationContext = new OperationContext |
| | | 87 | | { |
| | | 88 | | Proxy = this.host.EventProcessorOptions.WebProxy |
| | | 89 | | }; |
| | | 90 | | } |
| | | 91 | | #endif |
| | | 92 | | |
| | | 93 | | // Create storage client and configure max execution time. |
| | | 94 | | // Max execution time will apply to any storage call unless otherwise specified by custom request options. |
| | 0 | 95 | | var storageClient = this.cloudStorageAccount.CreateCloudBlobClient(); |
| | 0 | 96 | | storageClient.DefaultRequestOptions = new BlobRequestOptions |
| | 0 | 97 | | { |
| | 0 | 98 | | MaximumExecutionTime = AzureStorageCheckpointLeaseManager.storageMaximumExecutionTime |
| | 0 | 99 | | }; |
| | | 100 | | |
| | 0 | 101 | | this.eventHubContainer = storageClient.GetContainerReference(this.leaseContainerName); |
| | 0 | 102 | | this.consumerGroupDirectory = this.eventHubContainer.GetDirectoryReference(this.storageBlobPrefix + this.hos |
| | 0 | 103 | | } |
| | | 104 | | |
| | | 105 | | // |
| | | 106 | | // In this implementation, checkpoints are data that's actually in the lease blob, so checkpoint operations |
| | | 107 | | // turn into lease operations under the covers. |
| | | 108 | | // |
| | | 109 | | public Task<bool> CheckpointStoreExistsAsync() |
| | | 110 | | { |
| | 0 | 111 | | return LeaseStoreExistsAsync(); |
| | | 112 | | } |
| | | 113 | | |
| | | 114 | | public Task<bool> CreateCheckpointStoreIfNotExistsAsync() |
| | | 115 | | { |
| | | 116 | | // Because we control the caller, we know that this method will only be called after createLeaseStoreIfNotEx |
| | | 117 | | // In this implementation, it's the same store, so the store will always exist if execution reaches here. |
| | 0 | 118 | | return Task.FromResult(true); |
| | | 119 | | } |
| | | 120 | | |
| | | 121 | | public async Task<Checkpoint> GetCheckpointAsync(string partitionId) |
| | | 122 | | { |
| | 0 | 123 | | AzureBlobLease lease = (AzureBlobLease)await GetLeaseAsync(partitionId).ConfigureAwait(false); |
| | 0 | 124 | | Checkpoint checkpoint = null; |
| | 0 | 125 | | if (lease != null && !string.IsNullOrEmpty(lease.Offset)) |
| | | 126 | | { |
| | 0 | 127 | | checkpoint = new Checkpoint(partitionId) |
| | 0 | 128 | | { |
| | 0 | 129 | | Offset = lease.Offset, |
| | 0 | 130 | | SequenceNumber = lease.SequenceNumber |
| | 0 | 131 | | }; |
| | | 132 | | } |
| | | 133 | | |
| | 0 | 134 | | return checkpoint; |
| | 0 | 135 | | } |
| | | 136 | | |
| | | 137 | | public Task<Checkpoint> CreateCheckpointIfNotExistsAsync(string partitionId) |
| | | 138 | | { |
| | | 139 | | // Normally the lease will already be created, checkpoint store is initialized after lease store. |
| | 0 | 140 | | return Task.FromResult<Checkpoint>(null); |
| | | 141 | | } |
| | | 142 | | |
| | | 143 | | public async Task UpdateCheckpointAsync(Lease lease, Checkpoint checkpoint) |
| | | 144 | | { |
| | 0 | 145 | | AzureBlobLease newLease = new AzureBlobLease((AzureBlobLease) lease) |
| | 0 | 146 | | { |
| | 0 | 147 | | Offset = checkpoint.Offset, |
| | 0 | 148 | | SequenceNumber = checkpoint.SequenceNumber |
| | 0 | 149 | | }; |
| | | 150 | | |
| | 0 | 151 | | await this.UpdateLeaseAsync(newLease).ConfigureAwait(false); |
| | 0 | 152 | | } |
| | | 153 | | |
| | | 154 | | public Task DeleteCheckpointAsync(string partitionId) |
| | | 155 | | { |
| | | 156 | | // Make this a no-op to avoid deleting leases by accident. |
| | 0 | 157 | | return Task.FromResult(0); |
| | | 158 | | } |
| | | 159 | | |
| | | 160 | | // |
| | | 161 | | // Lease operations. |
| | | 162 | | // |
| | 0 | 163 | | public TimeSpan LeaseRenewInterval => this.leaseRenewInterval; |
| | | 164 | | |
| | 0 | 165 | | public TimeSpan LeaseDuration => this.leaseDuration; |
| | | 166 | | |
| | | 167 | | public Task<bool> LeaseStoreExistsAsync() |
| | | 168 | | { |
| | 0 | 169 | | return this.eventHubContainer.ExistsAsync(null, this.operationContext); |
| | | 170 | | } |
| | | 171 | | |
| | | 172 | | public Task<bool> CreateLeaseStoreIfNotExistsAsync() |
| | | 173 | | { |
| | 0 | 174 | | return this.eventHubContainer.CreateIfNotExistsAsync(null, this.operationContext); |
| | | 175 | | } |
| | | 176 | | |
| | | 177 | | public async Task<bool> DeleteLeaseStoreAsync() |
| | | 178 | | { |
| | 0 | 179 | | bool retval = true; |
| | | 180 | | |
| | 0 | 181 | | BlobContinuationToken outerContinuationToken = null; |
| | | 182 | | do |
| | | 183 | | { |
| | 0 | 184 | | BlobResultSegment outerResultSegment = await this.eventHubContainer.ListBlobsSegmentedAsync(outerContinu |
| | 0 | 185 | | outerContinuationToken = outerResultSegment.ContinuationToken; |
| | 0 | 186 | | foreach (IListBlobItem blob in outerResultSegment.Results) |
| | | 187 | | { |
| | 0 | 188 | | if (blob is CloudBlobDirectory) |
| | | 189 | | { |
| | 0 | 190 | | BlobContinuationToken innerContinuationToken = null; |
| | | 191 | | do |
| | | 192 | | { |
| | 0 | 193 | | BlobResultSegment innerResultSegment = await ((CloudBlobDirectory)blob).ListBlobsSegmentedAs |
| | 0 | 194 | | innerContinuationToken = innerResultSegment.ContinuationToken; |
| | 0 | 195 | | foreach (IListBlobItem subBlob in innerResultSegment.Results) |
| | | 196 | | { |
| | | 197 | | try |
| | | 198 | | { |
| | 0 | 199 | | await ((CloudBlockBlob)subBlob).DeleteIfExistsAsync().ConfigureAwait(false); |
| | 0 | 200 | | } |
| | 0 | 201 | | catch (StorageException e) |
| | | 202 | | { |
| | 0 | 203 | | ProcessorEventSource.Log.AzureStorageManagerWarning(this.host.HostName, "N/A", "Fail |
| | 0 | 204 | | retval = false; |
| | 0 | 205 | | } |
| | | 206 | | } |
| | | 207 | | } |
| | 0 | 208 | | while (innerContinuationToken != null); |
| | 0 | 209 | | } |
| | 0 | 210 | | else if (blob is CloudBlockBlob) |
| | | 211 | | { |
| | | 212 | | try |
| | | 213 | | { |
| | 0 | 214 | | await ((CloudBlockBlob)blob).DeleteIfExistsAsync().ConfigureAwait(false); |
| | 0 | 215 | | } |
| | 0 | 216 | | catch (StorageException e) |
| | | 217 | | { |
| | 0 | 218 | | ProcessorEventSource.Log.AzureStorageManagerWarning(this.host.HostName, "N/A", "Failure whil |
| | 0 | 219 | | retval = false; |
| | 0 | 220 | | } |
| | | 221 | | } |
| | 0 | 222 | | } |
| | | 223 | | } |
| | 0 | 224 | | while (outerContinuationToken != null); |
| | | 225 | | |
| | 0 | 226 | | return retval; |
| | 0 | 227 | | } |
| | | 228 | | |
| | | 229 | | public async Task<Lease> GetLeaseAsync(string partitionId) // throws URISyntaxException, IOException, StorageExc |
| | | 230 | | { |
| | 0 | 231 | | CloudBlockBlob leaseBlob = GetBlockBlobReference(partitionId); |
| | | 232 | | |
| | 0 | 233 | | await leaseBlob.FetchAttributesAsync(null, defaultRequestOptions, this.operationContext).ConfigureAwait(fals |
| | | 234 | | |
| | 0 | 235 | | return await DownloadLeaseAsync(partitionId, leaseBlob).ConfigureAwait(false); |
| | 0 | 236 | | } |
| | | 237 | | |
| | | 238 | | public async Task<IEnumerable<Lease>> GetAllLeasesAsync() |
| | | 239 | | { |
| | 0 | 240 | | var leaseList = new List<Lease>(); |
| | 0 | 241 | | BlobContinuationToken continuationToken = null; |
| | | 242 | | |
| | | 243 | | do |
| | | 244 | | { |
| | 0 | 245 | | var listBlobsTask = this.consumerGroupDirectory.ListBlobsSegmentedAsync( |
| | 0 | 246 | | true, |
| | 0 | 247 | | BlobListingDetails.Metadata, |
| | 0 | 248 | | null, |
| | 0 | 249 | | continuationToken, |
| | 0 | 250 | | this.defaultRequestOptions, |
| | 0 | 251 | | this.operationContext); |
| | | 252 | | |
| | | 253 | | // ListBlobsSegmentedAsync honors neither timeout settings in request options nor cancellation token and |
| | | 254 | | // This provides a workaround until we have storage.blob library fixed. |
| | | 255 | | BlobResultSegment leaseBlobsResult; |
| | 0 | 256 | | using (var cts = new CancellationTokenSource()) |
| | | 257 | | { |
| | 0 | 258 | | var delayTask = Task.Delay(this.LeaseRenewInterval, cts.Token); |
| | 0 | 259 | | var completedTask = await Task.WhenAny(listBlobsTask, delayTask).ConfigureAwait(false); |
| | | 260 | | |
| | 0 | 261 | | if (completedTask == listBlobsTask) |
| | | 262 | | { |
| | 0 | 263 | | cts.Cancel(); |
| | 0 | 264 | | leaseBlobsResult = await listBlobsTask; |
| | | 265 | | } |
| | | 266 | | else |
| | | 267 | | { |
| | | 268 | | // Throw OperationCanceledException, caller will log the failures appropriately. |
| | 0 | 269 | | throw new OperationCanceledException(); |
| | | 270 | | } |
| | 0 | 271 | | } |
| | | 272 | | |
| | 0 | 273 | | foreach (CloudBlockBlob leaseBlob in leaseBlobsResult.Results) |
| | | 274 | | { |
| | | 275 | | // Try getting owner name from existing blob. |
| | | 276 | | // This might return null when run on the existing lease after SDK upgrade. |
| | 0 | 277 | | leaseBlob.Metadata.TryGetValue(MetaDataOwnerName, out var owner); |
| | | 278 | | |
| | | 279 | | // Discover partition id from URI path of the blob. |
| | 0 | 280 | | var partitionId = leaseBlob.Uri.AbsolutePath.Split('/').Last(); |
| | | 281 | | |
| | 0 | 282 | | leaseList.Add(new AzureBlobLease(partitionId, owner, leaseBlob)); |
| | | 283 | | } |
| | | 284 | | |
| | 0 | 285 | | continuationToken = leaseBlobsResult.ContinuationToken; |
| | | 286 | | |
| | 0 | 287 | | } while (continuationToken != null); |
| | | 288 | | |
| | 0 | 289 | | return leaseList; |
| | 0 | 290 | | } |
| | | 291 | | |
| | | 292 | | public async Task<Lease> CreateLeaseIfNotExistsAsync(string partitionId) // throws URISyntaxException, IOExcepti |
| | | 293 | | { |
| | | 294 | | AzureBlobLease returnLease; |
| | | 295 | | try |
| | | 296 | | { |
| | 0 | 297 | | CloudBlockBlob leaseBlob = GetBlockBlobReference(partitionId); |
| | 0 | 298 | | returnLease = new AzureBlobLease(partitionId, leaseBlob); |
| | 0 | 299 | | string jsonLease = JsonConvert.SerializeObject(returnLease); |
| | | 300 | | |
| | 0 | 301 | | ProcessorEventSource.Log.AzureStorageManagerInfo( |
| | 0 | 302 | | this.host.HostName, |
| | 0 | 303 | | partitionId, |
| | 0 | 304 | | "CreateLeaseIfNotExist - leaseContainerName: " + this.leaseContainerName + |
| | 0 | 305 | | " consumerGroupName: " + this.host.ConsumerGroupName + " storageBlobPrefix: " + this.storageBlobPref |
| | | 306 | | |
| | | 307 | | // Don't provide default request options for upload call. |
| | | 308 | | // This request will respect client's default options. |
| | 0 | 309 | | await leaseBlob.UploadTextAsync( |
| | 0 | 310 | | jsonLease, |
| | 0 | 311 | | null, |
| | 0 | 312 | | AccessCondition.GenerateIfNoneMatchCondition("*"), |
| | 0 | 313 | | null, |
| | 0 | 314 | | this.operationContext).ConfigureAwait(false); |
| | 0 | 315 | | } |
| | 0 | 316 | | catch (StorageException se) |
| | | 317 | | { |
| | 0 | 318 | | if (se.RequestInformation.ErrorCode == BlobErrorCodeStrings.BlobAlreadyExists || |
| | 0 | 319 | | se.RequestInformation.ErrorCode == BlobErrorCodeStrings.LeaseIdMissing) // occurs when somebody els |
| | | 320 | | { |
| | | 321 | | // The blob already exists. |
| | 0 | 322 | | ProcessorEventSource.Log.AzureStorageManagerInfo(this.host.HostName, partitionId, "Lease already exi |
| | 0 | 323 | | returnLease = (AzureBlobLease)await GetLeaseAsync(partitionId).ConfigureAwait(false); |
| | | 324 | | } |
| | | 325 | | else |
| | | 326 | | { |
| | 0 | 327 | | ProcessorEventSource.Log.AzureStorageManagerError( |
| | 0 | 328 | | this.host.HostName, |
| | 0 | 329 | | partitionId, |
| | 0 | 330 | | "CreateLeaseIfNotExist StorageException - leaseContainerName: " + this.leaseContainerName + |
| | 0 | 331 | | " consumerGroupName: " + this.host.ConsumerGroupName + " storageBlobPrefix: " + this.storageBlob |
| | 0 | 332 | | se.ToString()); |
| | 0 | 333 | | throw; |
| | | 334 | | } |
| | | 335 | | } |
| | | 336 | | |
| | 0 | 337 | | return returnLease; |
| | 0 | 338 | | } |
| | | 339 | | |
| | | 340 | | public Task DeleteLeaseAsync(Lease lease) |
| | | 341 | | { |
| | 0 | 342 | | var azureBlobLease = (AzureBlobLease)lease; |
| | 0 | 343 | | ProcessorEventSource.Log.AzureStorageManagerInfo(this.host.HostName, azureBlobLease.PartitionId, "Deleting l |
| | 0 | 344 | | return azureBlobLease.Blob.DeleteIfExistsAsync(); |
| | | 345 | | } |
| | | 346 | | |
| | | 347 | | public Task<bool> AcquireLeaseAsync(Lease lease) |
| | | 348 | | { |
| | 0 | 349 | | return AcquireLeaseCoreAsync((AzureBlobLease)lease); |
| | | 350 | | } |
| | | 351 | | |
| | | 352 | | async Task<bool> AcquireLeaseCoreAsync(AzureBlobLease lease) |
| | | 353 | | { |
| | 0 | 354 | | CloudBlockBlob leaseBlob = lease.Blob; |
| | 0 | 355 | | bool retval = true; |
| | 0 | 356 | | string newLeaseId = Guid.NewGuid().ToString(); |
| | 0 | 357 | | string partitionId = lease.PartitionId; |
| | | 358 | | try |
| | | 359 | | { |
| | 0 | 360 | | bool renewLease = false; |
| | | 361 | | string newToken; |
| | | 362 | | |
| | 0 | 363 | | await leaseBlob.FetchAttributesAsync(null, this.defaultRequestOptions, this.operationContext).ConfigureA |
| | | 364 | | |
| | 0 | 365 | | if (leaseBlob.Properties.LeaseState == LeaseState.Leased) |
| | | 366 | | { |
| | 0 | 367 | | if (string.IsNullOrEmpty(lease.Token)) |
| | | 368 | | { |
| | | 369 | | // We reach here in a race condition: when this instance of EventProcessorHost scanned the |
| | | 370 | | // lease blobs, this partition was unowned (token is empty) but between then and now, another |
| | | 371 | | // instance of EPH has established a lease (getLeaseState() is LEASED). We normally enforce |
| | | 372 | | // that we only steal the lease if it is still owned by the instance which owned it when we |
| | | 373 | | // scanned, but we can't do that when we don't know who owns it. The safest thing to do is just |
| | | 374 | | // fail the acquisition. If that means that one EPH instance gets more partitions than it should |
| | | 375 | | // rebalancing will take care of that quickly enough. |
| | 0 | 376 | | return false; |
| | | 377 | | } |
| | | 378 | | |
| | 0 | 379 | | ProcessorEventSource.Log.AzureStorageManagerInfo(this.host.HostName, lease.PartitionId, "Need to Cha |
| | 0 | 380 | | renewLease = true; |
| | 0 | 381 | | newToken = await leaseBlob.ChangeLeaseAsync( |
| | 0 | 382 | | newLeaseId, |
| | 0 | 383 | | AccessCondition.GenerateLeaseCondition(lease.Token), |
| | 0 | 384 | | this.defaultRequestOptions, |
| | 0 | 385 | | this.operationContext).ConfigureAwait(false); |
| | | 386 | | } |
| | | 387 | | else |
| | | 388 | | { |
| | 0 | 389 | | ProcessorEventSource.Log.AzureStorageManagerInfo(this.host.HostName, lease.PartitionId, "Need to Acq |
| | 0 | 390 | | newToken = await leaseBlob.AcquireLeaseAsync( |
| | 0 | 391 | | leaseDuration, |
| | 0 | 392 | | newLeaseId, |
| | 0 | 393 | | null, |
| | 0 | 394 | | this.defaultRequestOptions, |
| | 0 | 395 | | this.operationContext).ConfigureAwait(false); |
| | | 396 | | } |
| | | 397 | | |
| | 0 | 398 | | lease.Token = newToken; |
| | 0 | 399 | | lease.Owner = this.host.HostName; |
| | 0 | 400 | | lease.IncrementEpoch(); // Increment epoch each time lease is acquired or stolen by a new host |
| | | 401 | | |
| | | 402 | | // Renew lease here if needed? |
| | | 403 | | // ChangeLease doesn't renew so we should avoid lease expiring before next renew interval. |
| | 0 | 404 | | if (renewLease) |
| | | 405 | | { |
| | 0 | 406 | | await this.RenewLeaseCoreAsync(lease).ConfigureAwait(false); |
| | | 407 | | } |
| | | 408 | | |
| | | 409 | | // Update owner in the metadata first since clients get ownership information by looking at metadata. |
| | 0 | 410 | | lease.Blob.Metadata[MetaDataOwnerName] = lease.Owner; |
| | 0 | 411 | | await lease.Blob.SetMetadataAsync( |
| | 0 | 412 | | AccessCondition.GenerateLeaseCondition(lease.Token), |
| | 0 | 413 | | this.defaultRequestOptions, |
| | 0 | 414 | | this.operationContext).ConfigureAwait(false); |
| | | 415 | | |
| | | 416 | | // Then update deserialized lease content. |
| | 0 | 417 | | await leaseBlob.UploadTextAsync( |
| | 0 | 418 | | JsonConvert.SerializeObject(lease), |
| | 0 | 419 | | null, |
| | 0 | 420 | | AccessCondition.GenerateLeaseCondition(lease.Token), |
| | 0 | 421 | | this.defaultRequestOptions, |
| | 0 | 422 | | this.operationContext).ConfigureAwait(false); |
| | 0 | 423 | | } |
| | 0 | 424 | | catch (StorageException se) |
| | | 425 | | { |
| | 0 | 426 | | throw HandleStorageException(partitionId, se); |
| | | 427 | | } |
| | | 428 | | |
| | 0 | 429 | | return retval; |
| | 0 | 430 | | } |
| | | 431 | | |
| | | 432 | | public Task<bool> RenewLeaseAsync(Lease lease) |
| | | 433 | | { |
| | 0 | 434 | | return RenewLeaseCoreAsync((AzureBlobLease)lease); |
| | | 435 | | } |
| | | 436 | | |
| | | 437 | | async Task<bool> RenewLeaseCoreAsync(AzureBlobLease lease) |
| | | 438 | | { |
| | 0 | 439 | | CloudBlockBlob leaseBlob = lease.Blob; |
| | 0 | 440 | | string partitionId = lease.PartitionId; |
| | | 441 | | |
| | | 442 | | try |
| | | 443 | | { |
| | 0 | 444 | | await leaseBlob.RenewLeaseAsync( |
| | 0 | 445 | | AccessCondition.GenerateLeaseCondition(lease.Token), |
| | 0 | 446 | | this.defaultRequestOptions, |
| | 0 | 447 | | this.operationContext).ConfigureAwait(false); |
| | 0 | 448 | | } |
| | 0 | 449 | | catch (StorageException se) |
| | | 450 | | { |
| | 0 | 451 | | throw HandleStorageException(partitionId, se); |
| | | 452 | | } |
| | | 453 | | |
| | 0 | 454 | | return true; |
| | 0 | 455 | | } |
| | | 456 | | |
| | | 457 | | public Task<bool> ReleaseLeaseAsync(Lease lease) |
| | | 458 | | { |
| | 0 | 459 | | return ReleaseLeaseCoreAsync((AzureBlobLease)lease); |
| | | 460 | | } |
| | | 461 | | |
| | | 462 | | async Task<bool> ReleaseLeaseCoreAsync(AzureBlobLease lease) |
| | | 463 | | { |
| | 0 | 464 | | ProcessorEventSource.Log.AzureStorageManagerInfo(this.host.HostName, lease.PartitionId, "Releasing lease"); |
| | | 465 | | |
| | 0 | 466 | | CloudBlockBlob leaseBlob = lease.Blob; |
| | 0 | 467 | | string partitionId = lease.PartitionId; |
| | | 468 | | |
| | | 469 | | try |
| | | 470 | | { |
| | 0 | 471 | | string leaseId = lease.Token; |
| | 0 | 472 | | AzureBlobLease releasedCopy = new AzureBlobLease(lease) |
| | 0 | 473 | | { |
| | 0 | 474 | | Token = string.Empty, |
| | 0 | 475 | | Owner = string.Empty |
| | 0 | 476 | | }; |
| | | 477 | | |
| | | 478 | | // Remove owner in the metadata. |
| | 0 | 479 | | leaseBlob.Metadata.Remove(MetaDataOwnerName); |
| | | 480 | | |
| | 0 | 481 | | await leaseBlob.SetMetadataAsync( |
| | 0 | 482 | | AccessCondition.GenerateLeaseCondition(leaseId), |
| | 0 | 483 | | this.defaultRequestOptions, |
| | 0 | 484 | | this.operationContext).ConfigureAwait(false); |
| | | 485 | | |
| | 0 | 486 | | await leaseBlob.UploadTextAsync( |
| | 0 | 487 | | JsonConvert.SerializeObject(releasedCopy), |
| | 0 | 488 | | null, |
| | 0 | 489 | | AccessCondition.GenerateLeaseCondition(leaseId), |
| | 0 | 490 | | this.defaultRequestOptions, |
| | 0 | 491 | | this.operationContext).ConfigureAwait(false); |
| | | 492 | | |
| | 0 | 493 | | await leaseBlob.ReleaseLeaseAsync(AccessCondition.GenerateLeaseCondition(leaseId), this.defaultRequestOp |
| | 0 | 494 | | } |
| | 0 | 495 | | catch (StorageException se) |
| | | 496 | | { |
| | 0 | 497 | | throw HandleStorageException(partitionId, se); |
| | | 498 | | } |
| | | 499 | | |
| | 0 | 500 | | return true; |
| | 0 | 501 | | } |
| | | 502 | | |
| | | 503 | | public Task<bool> UpdateLeaseAsync(Lease lease) |
| | | 504 | | { |
| | 0 | 505 | | return UpdateLeaseCoreAsync((AzureBlobLease)lease); |
| | | 506 | | } |
| | | 507 | | |
| | | 508 | | async Task<bool> UpdateLeaseCoreAsync(AzureBlobLease lease) |
| | | 509 | | { |
| | 0 | 510 | | if (lease == null) |
| | | 511 | | { |
| | 0 | 512 | | return false; |
| | | 513 | | } |
| | | 514 | | |
| | 0 | 515 | | string partitionId = lease.PartitionId; |
| | 0 | 516 | | ProcessorEventSource.Log.AzureStorageManagerInfo(this.host.HostName, partitionId, "Updating lease"); |
| | | 517 | | |
| | 0 | 518 | | string token = lease.Token; |
| | 0 | 519 | | if (string.IsNullOrEmpty(token)) |
| | | 520 | | { |
| | 0 | 521 | | return false; |
| | | 522 | | } |
| | | 523 | | |
| | 0 | 524 | | CloudBlockBlob leaseBlob = lease.Blob; |
| | | 525 | | try |
| | | 526 | | { |
| | 0 | 527 | | string jsonToUpload = JsonConvert.SerializeObject(lease); |
| | 0 | 528 | | ProcessorEventSource.Log.AzureStorageManagerInfo(this.host.HostName, lease.PartitionId, $"Raw JSON uploa |
| | | 529 | | |
| | | 530 | | // This is on the code path of checkpoint call thus don't provide default request options for upload cal |
| | | 531 | | // This request will respect client's default options. |
| | 0 | 532 | | await leaseBlob.UploadTextAsync( |
| | 0 | 533 | | jsonToUpload, |
| | 0 | 534 | | null, |
| | 0 | 535 | | AccessCondition.GenerateLeaseCondition(token), |
| | 0 | 536 | | null, |
| | 0 | 537 | | this.operationContext).ConfigureAwait(false); |
| | 0 | 538 | | } |
| | 0 | 539 | | catch (StorageException se) |
| | | 540 | | { |
| | 0 | 541 | | throw HandleStorageException(partitionId, se); |
| | | 542 | | } |
| | | 543 | | |
| | 0 | 544 | | return true; |
| | 0 | 545 | | } |
| | | 546 | | |
| | | 547 | | async Task<Lease> DownloadLeaseAsync(string partitionId, CloudBlockBlob blob) // throws StorageException, IOExce |
| | | 548 | | { |
| | 0 | 549 | | string jsonLease = await blob.DownloadTextAsync(null, null, this.defaultRequestOptions, this.operationContex |
| | | 550 | | |
| | 0 | 551 | | ProcessorEventSource.Log.AzureStorageManagerInfo(this.host.HostName, partitionId, "Raw JSON downloaded: " + |
| | 0 | 552 | | AzureBlobLease rehydrated = (AzureBlobLease)JsonConvert.DeserializeObject(jsonLease, typeof(AzureBlobLease)) |
| | 0 | 553 | | AzureBlobLease blobLease = new AzureBlobLease(rehydrated, blob); |
| | | 554 | | |
| | 0 | 555 | | return blobLease; |
| | 0 | 556 | | } |
| | | 557 | | |
| | | 558 | | Exception HandleStorageException(string partitionId, StorageException se) |
| | | 559 | | { |
| | 0 | 560 | | ProcessorEventSource.Log.AzureStorageManagerInfo(this.host.HostName, partitionId, "HandleStorageException - |
| | 0 | 561 | | if (se.RequestInformation.HttpStatusCode == 409 || // conflict |
| | 0 | 562 | | se.RequestInformation.HttpStatusCode == 412) // precondition failed |
| | | 563 | | { |
| | 0 | 564 | | ProcessorEventSource.Log.AzureStorageManagerInfo(this.host.HostName, partitionId, |
| | 0 | 565 | | "HandleStorageException - Error code: " + se.RequestInformation.ErrorCode); |
| | 0 | 566 | | ProcessorEventSource.Log.AzureStorageManagerInfo(this.host.HostName, partitionId, |
| | 0 | 567 | | "HandleStorageException - Error message: " + se.RequestInformation.ExtendedErrorInformation?.ErrorMe |
| | | 568 | | |
| | 0 | 569 | | if (se.RequestInformation.ErrorCode == null || |
| | 0 | 570 | | se.RequestInformation.ErrorCode == BlobErrorCodeStrings.LeaseLost || |
| | 0 | 571 | | se.RequestInformation.ErrorCode == BlobErrorCodeStrings.LeaseIdMismatchWithLeaseOperation || |
| | 0 | 572 | | se.RequestInformation.ErrorCode == BlobErrorCodeStrings.LeaseIdMismatchWithBlobOperation) |
| | | 573 | | { |
| | 0 | 574 | | return new LeaseLostException(partitionId, se); |
| | | 575 | | } |
| | | 576 | | } |
| | | 577 | | |
| | 0 | 578 | | return se; |
| | | 579 | | } |
| | | 580 | | |
| | | 581 | | CloudBlockBlob GetBlockBlobReference(string partitionId) |
| | | 582 | | { |
| | 0 | 583 | | return this.consumerGroupDirectory.GetBlockBlobReference(partitionId); |
| | | 584 | | } |
| | | 585 | | } |
| | | 586 | | } |