Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add legacy checkpoint reading support to EventHubs #17335

Merged
merged 9 commits into from
Dec 7, 2020
Merged

Add legacy checkpoint reading support to EventHubs #17335

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
a81834b
Add legacy checkpoint reading support to EventHubs
pakrym Dec 4, 2020
7c14f4c
cleanup parsing a bit
pakrym Dec 4, 2020
b35090f
Fancy mock blob client
pakrym Dec 7, 2020
364934c
implement per-parition reading
pakrym Dec 7, 2020
6ab097f
whitespace
pakrym Dec 7, 2020
485f25b
formatting
pakrym Dec 7, 2020
2dbb5f9
Merge remote-tracking branch 'azure/master' into pakrym/Add-legacy-ch…
pakrym Dec 7, 2020
5627402
formatting
pakrym Dec 7, 2020
68f916b
more formatting
pakrym Dec 7, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
147 changes: 144 additions & 3 deletions ...ub/Azure.Messaging.EventHubs.Processor/tests/CheckpointStore/BlobsCheckpointStoreTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Azure.Messaging.EventHubs.Consumer;
Expand Down Expand Up @@ -433,6 +434,135 @@ public async Task ListCheckpointsConsidersDataInvalidWithNoOffsetOrSequenceNumbe
mockLogger.Verify(log => log.InvalidCheckpointFound(partitionId, FullyQualifiedNamespace, EventHubName, ConsumerGroup));
}

/// <summary>
/// Verifies basic functionality of ListCheckpointsAsync and ensures the starting position is set correctly.
/// </summary>
///
[Test]
public async Task ListCheckpointsUsesOffsetAsTheStartingPositionWhenPresentInLegacyCheckpoint()
{
var blobList = new List<BlobItem>{
pakrym marked this conversation as resolved.
Show resolved Hide resolved
BlobsModelFactory.BlobItem($"{FullyQualifiedNamespace}/{EventHubName}/{ConsumerGroup}/0",
false,
BlobsModelFactory.BlobItemProperties(true, lastModified: DateTime.UtcNow, eTag: new ETag(MatchingEtag)),
"snapshot")
};

var containerClient = new MockBlobContainerClient() { Blobs = blobList };
containerClient.BlobContent = Encoding.UTF8.GetBytes("{" +
"\"PartitionId\":\"0\"," +
"\"Owner\":\"681d365b-de1b-4288-9733-76294e17daf0\"," +
"\"Token\":\"2d0c4276-827d-4ca4-a345-729caeca3b82\"," +
"\"Epoch\":386," +
"\"Offset\":\"13\"," +
"\"SequenceNumber\":960180}");

var target = new BlobsCheckpointStore(containerClient, new BasicRetryPolicy(new EventHubsRetryOptions()), readLegacyCheckpoints: true);
var checkpoints = await target.ListCheckpointsAsync(FullyQualifiedNamespace, EventHubName, ConsumerGroup, new CancellationToken());

Assert.That(checkpoints, Is.Not.Null, "A set of checkpoints should have been returned.");
Assert.That(checkpoints.Single().StartingPosition, Is.EqualTo(EventPosition.FromOffset(13, false)));
Assert.That(checkpoints.Single().PartitionId, Is.EqualTo("0"));
}

/// <summary>
/// Verifies basic functionality of ListCheckpointsAsync and ensures the starting position is set correctly.
/// </summary>
///
[Test]
public async Task ListCheckpointsUsesSequenceNumberAsTheStartingPositionWhenNoOffsetIsPresentInLegacyCheckpoint()
{
var blobList = new List<BlobItem>{
BlobsModelFactory.BlobItem($"{FullyQualifiedNamespace}/{EventHubName}/{ConsumerGroup}/0",
false,
BlobsModelFactory.BlobItemProperties(true, lastModified: DateTime.UtcNow, eTag: new ETag(MatchingEtag)),
"snapshot")
};

var containerClient = new MockBlobContainerClient() { Blobs = blobList };
containerClient.BlobContent = Encoding.UTF8.GetBytes("{" +
"\"PartitionId\":\"0\"," +
"\"Owner\":\"681d365b-de1b-4288-9733-76294e17daf0\"," +
"\"Token\":\"2d0c4276-827d-4ca4-a345-729caeca3b82\"," +
"\"Epoch\":386," +
"\"SequenceNumber\":960180}");

var target = new BlobsCheckpointStore(containerClient, new BasicRetryPolicy(new EventHubsRetryOptions()), readLegacyCheckpoints: true);
var checkpoints = await target.ListCheckpointsAsync(FullyQualifiedNamespace, EventHubName, ConsumerGroup, new CancellationToken());

Assert.That(checkpoints, Is.Not.Null, "A set of checkpoints should have been returned.");
Assert.That(checkpoints.Single().StartingPosition, Is.EqualTo(EventPosition.FromSequenceNumber(960180, false)));
Assert.That(checkpoints.Single().PartitionId, Is.EqualTo("0"));
}

/// <summary>
/// Verifies basic functionality of ListCheckpointsAsync and ensures the starting position is set correctly.
/// </summary>
///
[Test]
public async Task ListCheckpointsConsidersDataInvalidWithNoOffsetOrSequenceNumberLegacyCheckpoint()
{
var blobList = new List<BlobItem>{
BlobsModelFactory.BlobItem($"{FullyQualifiedNamespace}/{EventHubName}/{ConsumerGroup}/0",
false,
BlobsModelFactory.BlobItemProperties(true, lastModified: DateTime.UtcNow, eTag: new ETag(MatchingEtag)),
"snapshot")
};

var containerClient = new MockBlobContainerClient() { Blobs = blobList };
containerClient.BlobContent = Encoding.UTF8.GetBytes("{" +
"\"PartitionId\":\"0\"," +
"\"Owner\":\"681d365b-de1b-4288-9733-76294e17daf0\"," +
"\"Token\":\"2d0c4276-827d-4ca4-a345-729caeca3b82\"," +
"\"Epoch\":386}");


pakrym marked this conversation as resolved.
Show resolved Hide resolved
var mockLogger = new Mock<BlobEventStoreEventSource>();
var target = new BlobsCheckpointStore(containerClient, new BasicRetryPolicy(new EventHubsRetryOptions()), readLegacyCheckpoints: true);

target.Logger = mockLogger.Object;

var checkpoints = await target.ListCheckpointsAsync(FullyQualifiedNamespace, EventHubName, ConsumerGroup, new CancellationToken());

Assert.That(checkpoints, Is.Not.Null, "A set of checkpoints should have been returned.");
Assert.That(checkpoints.Any(), Is.False, "No valid checkpoints should exist.");

mockLogger.Verify(log => log.InvalidCheckpointFound("0", FullyQualifiedNamespace, EventHubName, ConsumerGroup));
}

/// <summary>
/// Verifies basic functionality of ListCheckpointsAsync and ensures the starting position is set correctly.
/// </summary>
///
[TestCase("")]
[TestCase("{\"PartitionId\":\"0\",\"Owner\":\"681d365b-de1b-4288-9733-76294e17daf0\",")]
[TestCase("\0\0\0")]
public async Task ListCheckpointsConsidersDataInvalidWithLegacyCheckpointBlobContainingInvalidJson(string json)
{
var blobList = new List<BlobItem>{
BlobsModelFactory.BlobItem($"{FullyQualifiedNamespace}/{EventHubName}/{ConsumerGroup}/0",
false,
BlobsModelFactory.BlobItemProperties(true, lastModified: DateTime.UtcNow, eTag: new ETag(MatchingEtag)),
"snapshot")
};

var containerClient = new MockBlobContainerClient() { Blobs = blobList };
containerClient.BlobContent = Encoding.UTF8.GetBytes(json);

pakrym marked this conversation as resolved.
Show resolved Hide resolved

var mockLogger = new Mock<BlobEventStoreEventSource>();
var target = new BlobsCheckpointStore(containerClient, new BasicRetryPolicy(new EventHubsRetryOptions()), readLegacyCheckpoints: true);

target.Logger = mockLogger.Object;

var checkpoints = await target.ListCheckpointsAsync(FullyQualifiedNamespace, EventHubName, ConsumerGroup, new CancellationToken());

Assert.That(checkpoints, Is.Not.Null, "A set of checkpoints should have been returned.");
Assert.That(checkpoints.Any(), Is.False, "No valid checkpoints should exist.");

mockLogger.Verify(log => log.InvalidCheckpointFound("0", FullyQualifiedNamespace, EventHubName, ConsumerGroup));
}

/// <summary>
/// Verifies basic functionality of ListCheckpointsAsync and ensures the appropriate events are emitted when errors occur.
/// </summary>
Expand Down Expand Up @@ -1438,6 +1568,7 @@ private class MockBlobContainerClient : BlobContainerClient
public override string Name { get; }
internal IEnumerable<BlobItem> Blobs;
internal BlobInfo BlobInfo;
internal byte[] BlobContent;
internal Exception BlobClientUploadBlobException;
internal Exception BlobClientSetMetadataException;
internal Exception GetBlobsAsyncException;
Expand Down Expand Up @@ -1466,12 +1597,12 @@ public override AsyncPageable<BlobItem> GetBlobsAsync(BlobTraits traits = BlobTr
throw GetBlobsAsyncException;
}

return new MockAsyncPageable<BlobItem>(Blobs);
return new MockAsyncPageable<BlobItem>(Blobs.Where(b => prefix == null || b.Name.StartsWith(prefix, StringComparison.Ordinal)));
}

public override BlobClient GetBlobClient(string blobName)
{
return new MockBlobClient(blobName, BlobInfo, BlobClientUploadBlobException, BlobClientSetMetadataException, BlobClientUploadAsyncCallback, BlobClientSetMetadataAsyncCallback);
return new MockBlobClient(blobName, BlobInfo, BlobClientUploadBlobException, BlobClientSetMetadataException, BlobClientUploadAsyncCallback, BlobClientSetMetadataAsyncCallback, BlobContent);
}
}

Expand All @@ -1481,6 +1612,8 @@ private class MockBlobClient : BlobClient
internal BlobInfo BlobInfo;
internal Exception BlobClientUploadBlobException;
internal Exception BlobClientSetMetadataException;

private byte[] Content;
private Action<Stream, BlobHttpHeaders, IDictionary<string, string>, BlobRequestConditions, IProgress<long>, AccessTier?, StorageTransferOptions, CancellationToken> UploadAsyncCallback;
private Action<IDictionary<string, string>, BlobRequestConditions, CancellationToken> SetMetadataAsyncCallback;

Expand All @@ -1489,14 +1622,16 @@ public MockBlobClient(string blobName,
Exception blobClientUploadBlobException = null,
Exception blobClientSetMetadataException = null,
Action<Stream, BlobHttpHeaders, IDictionary<string, string>, BlobRequestConditions, IProgress<long>, AccessTier?, StorageTransferOptions, CancellationToken> uploadAsyncCallback = null,
Action<IDictionary<string, string>, BlobRequestConditions, CancellationToken> setMetadataAsyncCallback = null)
Action<IDictionary<string, string>, BlobRequestConditions, CancellationToken> setMetadataAsyncCallback = null,
byte[] content = null)
{
BlobClientUploadBlobException = blobClientUploadBlobException;
BlobClientSetMetadataException = blobClientSetMetadataException;
UploadAsyncCallback = uploadAsyncCallback;
SetMetadataAsyncCallback = setMetadataAsyncCallback;
Name = blobName;
BlobInfo = blobInfo;
Content = content;
}

public override Task<Response<BlobInfo>> SetMetadataAsync(IDictionary<string, string> metadata, BlobRequestConditions conditions = null, CancellationToken cancellationToken = default(CancellationToken))
Expand Down Expand Up @@ -1540,6 +1675,12 @@ public override Task<Response<BlobContentInfo>> UploadAsync(Stream content, Blob
BlobsModelFactory.BlobContentInfo(new ETag("etag"), DateTime.UtcNow, new byte[] { }, string.Empty, 0L),
Mock.Of<Response>()));
}

public override async Task<Response> DownloadToAsync(Stream destination, CancellationToken cancellationToken)
{
await destination.WriteAsync(Content, 0, Content.Length, cancellationToken);
return Mock.Of<Response>();
}
}

private class MockAsyncPageable<T> : AsyncPageable<T>
Expand Down
Loading