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

[Event Hubs Processor] Validate at Startup #20861

Merged
merged 1 commit into from
May 6, 2021
Merged

[Event Hubs Processor] Validate at Startup #20861

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
7 changes: 7 additions & 0 deletions sdk/eventhub/Azure.Messaging.EventHubs.Processor/Azure.Messaging.EventHubs.Processor.sln
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Azure.Core.TestFramework",
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "External", "External", "{797FF941-76FD-45FD-AC17-A73DFE2BA621}"
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Azure.Messaging.EventHubs", "..\Azure.Messaging.EventHubs\src\Azure.Messaging.EventHubs.csproj", "{D82CC3DD-E6CC-4688-BDA8-16E7A303C35B}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
Expand All @@ -29,12 +31,17 @@ Global
{7DFF0E65-DC9A-410D-9A11-AD6A06860FE1}.Debug|Any CPU.Build.0 = Debug|Any CPU
{7DFF0E65-DC9A-410D-9A11-AD6A06860FE1}.Release|Any CPU.ActiveCfg = Release|Any CPU
{7DFF0E65-DC9A-410D-9A11-AD6A06860FE1}.Release|Any CPU.Build.0 = Release|Any CPU
{D82CC3DD-E6CC-4688-BDA8-16E7A303C35B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{D82CC3DD-E6CC-4688-BDA8-16E7A303C35B}.Debug|Any CPU.Build.0 = Debug|Any CPU
{D82CC3DD-E6CC-4688-BDA8-16E7A303C35B}.Release|Any CPU.ActiveCfg = Release|Any CPU
{D82CC3DD-E6CC-4688-BDA8-16E7A303C35B}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
EndGlobalSection
GlobalSection(NestedProjects) = preSolution
{7DFF0E65-DC9A-410D-9A11-AD6A06860FE1} = {797FF941-76FD-45FD-AC17-A73DFE2BA621}
{D82CC3DD-E6CC-4688-BDA8-16E7A303C35B} = {797FF941-76FD-45FD-AC17-A73DFE2BA621}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {44BD3BD5-61DF-464D-8627-E00B0BC4B3A3}
Expand Down
5 changes: 4 additions & 1 deletion ...enthub/Azure.Messaging.EventHubs.Processor/src/Azure.Messaging.EventHubs.Processor.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,10 @@
</ItemGroup>

<ItemGroup>
<PackageReference Include="Azure.Messaging.EventHubs" />
<!-- RETURN TO PACAKGE REFERENCE -->
<ProjectReference Include="$(MSBuildThisFileDirectory)..\..\Azure.Messaging.EventHubs\src\Azure.Messaging.EventHubs.csproj" />
<!-- PackageReference Include="Azure.Messaging.EventHubs" /-->
<!-- END RETURN TO PACKAGE REFERENCE -->
<PackageReference Include="Azure.Storage.Blobs" />
<PackageReference Include="Microsoft.Azure.Amqp" />
<PackageReference Include="Microsoft.Bcl.AsyncInterfaces" />
Expand Down
21 changes: 21 additions & 0 deletions ...ub/Azure.Messaging.EventHubs.Processor/src/Diagnostics/EventProcessorClientEventSource.cs
100755 → 100644
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,5 +178,26 @@ public virtual void UpdateCheckpointError(string partitionId,
WriteEvent(25, partitionId ?? string.Empty, identifier ?? string.Empty, eventHubName ?? string.Empty, consumerGroup ?? string.Empty, errorMessage ?? string.Empty);
}
}

/// <summary>
/// Indicates that the process of cleaning up after startup validation has experienced an exception.
/// </summary>
///
/// <param name="identifier">A unique name used to identify the event processor.</param>
/// <param name="eventHubName">The name of the Event Hub that the processor is associated with.</param>
/// <param name="consumerGroup">The name of the consumer group that the processor is associated with.</param>
/// <param name="errorMessage">The message for the exception that occurred.</param>
///
[Event(26, Level = EventLevel.Error, Message = "An exception occurred while attempting to perform cleanup after validating the processor configuration and permissions during startup for processor instance with identifier '{1}' for Event Hub: {2} and Consumer Group: {3}. Error Message: '{4}'")]
public virtual void ValidationCleanupError(string identifier,
string eventHubName,
string consumerGroup,
string errorMessage)
{
if (IsEnabled())
{
WriteEvent(26, identifier ?? string.Empty, eventHubName ?? string.Empty, consumerGroup ?? string.Empty, errorMessage ?? string.Empty);
}
}
}
}
325 changes: 214 additions & 111 deletions sdk/eventhub/Azure.Messaging.EventHubs.Processor/src/EventProcessorClient.cs
View file
Open in desktop

Large diffs are not rendered by default.

3 changes: 1 addition & 2 deletions sdk/eventhub/Azure.Messaging.EventHubs.Processor/tests/Diagnostics/DiagnosticsTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,12 +11,11 @@
using Azure.Messaging.EventHubs.Diagnostics;
using Azure.Messaging.EventHubs.Primitives;
using Azure.Messaging.EventHubs.Processor.Diagnostics;
using Azure.Messaging.EventHubs.Tests;
using Moq;
using Moq.Protected;
using NUnit.Framework;

namespace Azure.Messaging.EventHubs.Processor.Tests
namespace Azure.Messaging.EventHubs.Tests
{
/// <summary>
/// The suite of tests for validating the diagnostics instrumentation
Expand Down
179 changes: 179 additions & 0 deletions ...thub/Azure.Messaging.EventHubs.Processor/tests/Processor/EventProcessorClientLiveTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
using Azure.Messaging.EventHubs.Primitives;
using Azure.Messaging.EventHubs.Processor;
using Azure.Messaging.EventHubs.Producer;
using Azure.Storage.Blobs;
using Moq;
using NUnit.Framework;

Expand Down Expand Up @@ -531,6 +532,184 @@ public async Task ProcessorClientBeginsWithTheNextEventAfterCheckpointing()
}
}

/// <summary>
/// Verifies that the <see cref="EventProcessorClient" /> can read a set of published events.
/// </summary>
///
[Test]
[TestCase(true)]
[TestCase(false)]
public async Task ProcessorClientDetectsAnInvalidEventHubsConnectionString(bool async)
{
// Setup the environment.

await using EventHubScope scope = await EventHubScope.CreateAsync(2);
var connectionString = "Endpoint=sb://fake.servicebus.windows.net/;SharedAccessKeyName=FakeSharedAccessKey;SharedAccessKey=<< FAKE >>";

using var cancellationSource = new CancellationTokenSource();
cancellationSource.CancelAfter(EventHubsTestEnvironment.Instance.TestExecutionTimeLimit);

// Create the processor and attempt to start.

var processor = new EventProcessorClient(Mock.Of<BlobContainerClient>(), EventHubConsumerClient.DefaultConsumerGroupName, connectionString, scope.EventHubName);
processor.ProcessErrorAsync += _ => Task.CompletedTask;
processor.ProcessEventAsync += _ => Task.CompletedTask;

if (async)
{
Assert.That(async () => await processor.StartProcessingAsync(cancellationSource.Token), Throws.InstanceOf<AggregateException>());
}
else
{
Assert.That(() => processor.StartProcessing(cancellationSource.Token), Throws.InstanceOf<AggregateException>());
}

await processor.StopProcessingAsync(cancellationSource.Token);
cancellationSource.Cancel();
}

/// <summary>
/// Verifies that the <see cref="EventProcessorClient" /> can read a set of published events.
/// </summary>
///
[Test]
[TestCase(true)]
[TestCase(false)]
public async Task ProcessorClientDetectsAnInvalidEventHubName(bool async)
{
// Setup the environment.

using var cancellationSource = new CancellationTokenSource();
cancellationSource.CancelAfter(EventHubsTestEnvironment.Instance.TestExecutionTimeLimit);

// Create the processor and attempt to start.

var processor = new EventProcessorClient(Mock.Of<BlobContainerClient>(), EventHubConsumerClient.DefaultConsumerGroupName, EventHubsTestEnvironment.Instance.EventHubsConnectionString, "fake");
processor.ProcessErrorAsync += _ => Task.CompletedTask;
processor.ProcessEventAsync += _ => Task.CompletedTask;

if (async)
{
Assert.That(async () => await processor.StartProcessingAsync(cancellationSource.Token), Throws.InstanceOf<AggregateException>());
}
else
{
Assert.That(() => processor.StartProcessing(cancellationSource.Token), Throws.InstanceOf<AggregateException>());
}

await processor.StopProcessingAsync(cancellationSource.Token);
cancellationSource.Cancel();
}

/// <summary>
/// Verifies that the <see cref="EventProcessorClient" /> can read a set of published events.
/// </summary>
///
[Test]
[TestCase(true)]
[TestCase(false)]
public async Task ProcessorClientDetectsAnInvalidConsumerGroup(bool async)
{
// Setup the environment.

await using EventHubScope scope = await EventHubScope.CreateAsync(2);

using var cancellationSource = new CancellationTokenSource();
cancellationSource.CancelAfter(EventHubsTestEnvironment.Instance.TestExecutionTimeLimit);

// Create the processor and attempt to start.

var processor = new EventProcessorClient(Mock.Of<BlobContainerClient>(), "fake", EventHubsTestEnvironment.Instance.EventHubsConnectionString, scope.EventHubName);
processor.ProcessErrorAsync += _ => Task.CompletedTask;
processor.ProcessEventAsync += _ => Task.CompletedTask;

if (async)
{
Assert.That(async () => await processor.StartProcessingAsync(cancellationSource.Token), Throws.InstanceOf<AggregateException>());
}
else
{
Assert.That(() => processor.StartProcessing(cancellationSource.Token), Throws.InstanceOf<AggregateException>());
}

await processor.StopProcessingAsync(cancellationSource.Token);
cancellationSource.Cancel();
}

/// <summary>
/// Verifies that the <see cref="EventProcessorClient" /> can read a set of published events.
/// </summary>
///
[Test]
[TestCase(true)]
[TestCase(false)]
public async Task ProcessorClientDetectsAnInvalidStorageConnectionString(bool async)
{
// Setup the environment.

await using EventHubScope eventHubScope = await EventHubScope.CreateAsync(2);
await using StorageScope storageScope = await StorageScope.CreateAsync();

using var cancellationSource = new CancellationTokenSource();
cancellationSource.CancelAfter(EventHubsTestEnvironment.Instance.TestExecutionTimeLimit);

// Create the processor and attempt to start.

var storageConnectionString = StorageTestEnvironment.Instance.StorageConnectionString.Replace(StorageTestEnvironment.Instance.StorageEndpointSuffix, "fake.com");
var containerClient = new BlobContainerClient(storageConnectionString, storageScope.ContainerName);
var processor = new EventProcessorClient(containerClient, eventHubScope.ConsumerGroups[0], EventHubsTestEnvironment.Instance.EventHubsConnectionString, eventHubScope.EventHubName);
processor.ProcessErrorAsync += _ => Task.CompletedTask;
processor.ProcessEventAsync += _ => Task.CompletedTask;

if (async)
{
Assert.That(async () => await processor.StartProcessingAsync(cancellationSource.Token), Throws.InstanceOf<AggregateException>());
}
else
{
Assert.That(() => processor.StartProcessing(cancellationSource.Token), Throws.InstanceOf<AggregateException>());
}

await processor.StopProcessingAsync(cancellationSource.Token);
cancellationSource.Cancel();
}

/// <summary>
/// Verifies that the <see cref="EventProcessorClient" /> can read a set of published events.
/// </summary>
///
[Test]
[TestCase(true)]
[TestCase(false)]
public async Task ProcessorClientDetectsAnInvalidStorageContainer(bool async)
{
// Setup the environment.

await using EventHubScope eventHubScope = await EventHubScope.CreateAsync(2);

using var cancellationSource = new CancellationTokenSource();
cancellationSource.CancelAfter(EventHubsTestEnvironment.Instance.TestExecutionTimeLimit);

// Create the processor and attempt to start.

var containerClient = new BlobContainerClient(StorageTestEnvironment.Instance.StorageConnectionString, "fake");
var processor = new EventProcessorClient(containerClient, eventHubScope.ConsumerGroups[0], EventHubsTestEnvironment.Instance.EventHubsConnectionString, eventHubScope.EventHubName);
processor.ProcessErrorAsync += _ => Task.CompletedTask;
processor.ProcessEventAsync += _ => Task.CompletedTask;

if (async)
{
Assert.That(async () => await processor.StartProcessingAsync(cancellationSource.Token), Throws.InstanceOf<AggregateException>());
}
else
{
Assert.That(() => processor.StartProcessing(cancellationSource.Token), Throws.InstanceOf<AggregateException>());
}

await processor.StopProcessingAsync(cancellationSource.Token);
cancellationSource.Cancel();
}

/// <summary>
/// Creates an <see cref="EventProcessorClient" /> that uses mock storage and
/// a connection based on a connection string.
Expand Down
Loading