Scaling Azure Event Hubs Processing with Worker Roles

This post will show how to host an Azure Event Hubs EventProcessorHost in a cloud service worker role.

Background

Imagine that you have a scenario where you have a solution that needs to ingest millions of events per second, and you need to process these events in near real time.  There are many scenarios where this might occur.  Perhaps you built a solution for a consumer device such as Microsoft Band using the recently announced Microsoft Band SDK preview, you have a popular game where you are processing telemetry events, or you might have an IoT solution where you are processing events from many devices.  Architecting for this type of solution poses many challenges.  How do you enable ingest of messages at this scale?  More importantly, how do you enable processing of the messages in a manner which does not create a performance bottleneck?  How do you independently scale the message ingest and message processing capabilities?  Most importantly, how do you achieve this for very little cost?

Event Hubs and EventProcessorHost

Azure Event Hubs enables massive scale telemetry processing for solutions that require logging millions of events per second in near real time.  It does this by creating a gigantic write-ahead stream where the ingest and processing of data is handled separately.  One way to think of this is as a massively scalable queue implementation.  However, unlike queues, there is no concept of queue length in Azure Event Hubs because the data is processed as a stream.  I wrote a demonstration of the sender side in the post Use JMS and Azure Event Hubs with Eclipse, where many senders send information to an Event Hub named “devicereadings”.  The Event Hub then distributes the messages across partitions.

The question is now how to process the events in a scalable manner.  That’s exactly what Event Hubs enables, scalable event processing.  The Azure Event Hubs team made this processing easy by introducing the EventProcessorHost that enables a partitioned consumer pattern.  Consider the following scenario where I have a worker role that is processing all of the stream data from the Event Hub.  The EventProcessorHost will automatically create an instance for each partition. 

We might see that a single worker role is not sufficient for our solution, that the worker role becomes CPU bound.  As we saw in my previous post, Autoscaling Azure–Cloud Services, scaling a worker role creates a new instance of the worker role.  The benefit of using the EventProcessorHost is that we can add more worker role instances, and the partitions will be balanced across them.  We don’t have to manage the number of processor instances, the EventProcessorHost handles that for us.

Using the EventProcessorHost is simple because you need to implement one interface, IEventProcessor, the bulk of the work will be in the ProcessEventsAsync class.

Let’s work through building the solution. 

Show Me Some Code

I am going to use the Service Bus Event Hubs Getting Started sample as the basis for this post, I made some changes to the code and I’ll explain along the way.  First, I created a class library called EventHubDemo.Common.  It contains NuGet references to Json.NET, Microsoft Azure Service Bus, and Windows Azure Configuration Manager.

It contains two classes:  MetricEvent.cs and EventHubManager.cs.

MetricEvent is just a class used to (de)serialize JSON data.

MetricEvent.cs

1. using System.Runtime.Serialization;
2.  
3. namespace EventHubDemo.Common.Contracts
4. {
5.     [DataContract]
6.     public class MetricEvent
7.     {
8.         [DataMember]
9.         public int DeviceId { get; set; }
10.         [DataMember]
11.         public int Temperature { get; set; }
12.     }
13. }

The next class, EventHubManager.cs, is just a helper class for working with Event Hubs.  It centralizes common logic that is used across multiple projects to work with the Event Hub connection string and to create an Event Hub.

Code Snippet

1. using Microsoft.ServiceBus;
2. using Microsoft.ServiceBus.Messaging;
3. using System;
4. using System.Diagnostics;
5.  
6. namespace EventHubDemo.Common.Utility
7. {
8.     public class EventHubManager
9.     {
10.         public static string GetServiceBusConnectionString()
11.         {
12.             string connectionString = Microsoft.WindowsAzure.CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");
13.             
14.             if (string.IsNullOrEmpty(connectionString))
15.             {
16.                 Trace.WriteLine("Did not find Service Bus connections string in appsettings (app.config)");
17.                 return string.Empty;
18.             }
19.             ServiceBusConnectionStringBuilder builder = new ServiceBusConnectionStringBuilder(connectionString);
20.             builder.TransportType = TransportType.Amqp;
21.             return builder.ToString();
22.         }
23.  
24.         public static NamespaceManager GetNamespaceManager()
25.         {
26.             return NamespaceManager.CreateFromConnectionString(GetServiceBusConnectionString());
27.         }
28.  
29.         public static NamespaceManager GetNamespaceManager(string connectionString)
30.         {
31.             return NamespaceManager.CreateFromConnectionString(connectionString);
32.         }
33.  
34.  
35.         public static void CreateEventHubIfNotExists(string eventHubName, int numberOfPartitions, NamespaceManager manager)
36.         {
37.             try
38.             {
39.                 // Create the Event Hub
40.                 Trace.WriteLine("Creating Event Hub...");
41.                 EventHubDescription ehd = new EventHubDescription(eventHubName);
42.                 ehd.PartitionCount = numberOfPartitions;
43.                 manager.CreateEventHubIfNotExistsAsync(ehd).Wait();
44.             }
45.             catch (AggregateException agexp)
46.             {
47.                 Trace.WriteLine(agexp.Flatten());
48.             }
49.         }
50.  
51.     }
52. }

This makes it easy to reuse the logic in a Console application that sends messages or in an Azure worker role that receives them.

EventProcessorHost in an Azure Worker Role

The next step is to create a Cloud Service.  I create a new Azure Cloud Service project named “EventProcessor” in Visual Studio.

Click OK, and you are prompted for the type of role you want to create.  Choose a worker role, and rename it to something like “ReceiverRole”.

Click OK, and you now have two projects: the worker role code and the deployment project.

The bulk of the work will be in the ReceiverRole project.  Add a reference to the EventHubDemo.Common library that we created previously.  Next add the NuGet package “Microsoft.Azure.ServiceBus.EventProcessorHost”, which will add the required dependencies.

Next we add a class called SimpleEventProcessor.cs.  This class will implement the IEventProcessor interface that we mentioned previously.  When the OpenAsync method is called by the EventProcessorHost, we will write out the partition that it corresponds to.  When a message is received in the ProcessEventsAsync method, we write the message out to Trace output.  When the CloseAsync method is called, we write that out to trace as well.

SimpleEventProcessor.cs

1. using EventHubDemo.Common.Contracts;
2. using Microsoft.ServiceBus.Messaging;
3. using Newtonsoft.Json;
4. using System;
5. using System.Collections.Generic;
6. using System.Diagnostics;
7. using System.Linq;
8. using System.Text;
9. using System.Threading.Tasks;
10.  
11. namespace ReceiverRole
12. {
13.     class SimpleEventProcessor : IEventProcessor
14.     {
15.         PartitionContext partitionContext;
16.  
17.         public Task OpenAsync(PartitionContext context)
18.         {
19.             Trace.TraceInformation(string.Format("SimpleEventProcessor OpenAsync.  Partition: '{0}', Offset: '{1}'", context.Lease.PartitionId, context.Lease.Offset));
20.             this.partitionContext = context;
21.             return Task.FromResult<object>(null);
22.         }
23.  
24.         public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> events)
25.         {
26.             try
27.             {
28.                 foreach (EventData eventData in events)
29.                 {
30.                     try
31.                     {
32.                         var newData = this.DeserializeEventData(eventData);
33.  
34.                         Trace.TraceInformation(string.Format("Message received.  Partition: '{0}', Device: '{1}', Data: '{2}'",
35.                             this.partitionContext.Lease.PartitionId, newData.DeviceId, newData.Temperature));
36.                     }
37.                     catch (Exception oops)
38.                     {
39.                         Trace.TraceError(oops.Message);
40.                     }
41.                 }
42.  
43.                 await context.CheckpointAsync();
44.  
45.             }
46.             catch (Exception exp)
47.             {
48.                 Trace.TraceError("Error in processing: " + exp.Message);
49.             }
50.         }
51.  
52.         public async Task CloseAsync(PartitionContext context, CloseReason reason)
53.         {
54.             Trace.TraceWarning(string.Format("SimpleEventProcessor CloseAsync.  Partition '{0}', Reason: '{1}'.", this.partitionContext.Lease.PartitionId, reason.ToString()));
55.             if (reason == CloseReason.Shutdown)
56.             {
57.                 await context.CheckpointAsync();
58.             }
59.         }
60.  
61.         MetricEvent DeserializeEventData(EventData eventData)
62.         {
63.  
64.             string data = Encoding.UTF8.GetString(eventData.GetBytes());
65.             return JsonConvert.DeserializeObject<MetricEvent>(data);
66.         }
67.     }
68. }

Easy enough so far.  Now add a class, Receiver.cs.  This class will encapsulate the logic for registering and unregistering an IEventProcessor implementation.

Receiver.cs

1. using Microsoft.ServiceBus.Messaging;
2. using System.Diagnostics;
3.  
4. namespace ReceiverRole
5. {
6.     class Receiver
7.     {        
8.         string eventHubName;
9.         string eventHubConnectionString;
10.         EventProcessorHost eventProcessorHost;
11.         
12.         public Receiver(string eventHubName, string eventHubConnectionString)
13.         {
14.             this.eventHubConnectionString = eventHubConnectionString;
15.             this.eventHubName = eventHubName;
16.         }
17.  
18.         public void RegisterEventProcessor(ConsumerGroupDescription group, string blobConnectionString, string hostName)
19.         {
20.             EventHubClient eventHubClient = EventHubClient.CreateFromConnectionString(eventHubConnectionString, this.eventHubName);
21.  
22.             if (null == group)
23.             {
24.                 //Use default consumer group
25.                 EventHubConsumerGroup defaultConsumerGroup = eventHubClient.GetDefaultConsumerGroup();
26.                 eventProcessorHost = new EventProcessorHost(hostName, eventHubClient.Path, defaultConsumerGroup.GroupName, this.eventHubConnectionString, blobConnectionString);
27.             }
28.             else
29.             {
30.                 //Use custom consumer group
31.                 eventProcessorHost = new EventProcessorHost(hostName, eventHubClient.Path, group.Name, this.eventHubConnectionString, blobConnectionString);
32.             }
33.  
34.             
35.             Trace.TraceInformation("Registering event processor");
36.             
37.             eventProcessorHost.RegisterEventProcessorAsync<SimpleEventProcessor>().Wait();
38.         }
39.  
40.         public void UnregisterEventProcessor()
41.         {
42.             eventProcessorHost.UnregisterEventProcessorAsync().Wait();
43.         }
44.     }
45. }

The final bit is to host this in an Azure cloud service using a worker role.  I updated the WorkerRole.cs class.

Code Snippet

1. using EventHubDemo.Common.Utility;
2. using Microsoft.WindowsAzure;
3. using Microsoft.WindowsAzure.ServiceRuntime;
4. using System;
5. using System.Diagnostics;
6. using System.Net;
7. using System.Threading;
8.  
9. namespace ReceiverRole
10. {
11.     public class WorkerRole : RoleEntryPoint
12.     {                
13.         private Receiver receiver;        
14.         private readonly ManualResetEvent runCompleteEvent = new ManualResetEvent(false);
15.  
16.         public override void Run()
17.         {
18.             Trace.TraceInformation("ReceiverRole is running");
19.  
20.             //Get settings from configuration
21.             var eventHubName = CloudConfigurationManager.GetSetting("eventHubName");
22.             var consumerGroupName = CloudConfigurationManager.GetSetting("consumerGroupName");
23.             var numberOfPartitions = int.Parse(CloudConfigurationManager.GetSetting("numberOfPartitions"));
24.             var blobConnectionString = CloudConfigurationManager.GetSetting("AzureStorageConnectionString"); // Required for checkpoint/state
25.  
26.             //Get AMQP connection string
27.             var connectionString = EventHubManager.GetServiceBusConnectionString();
28.  
29.             //Create event hub if it does not exist
30.             var namespaceManager = EventHubManager.GetNamespaceManager(connectionString);
31.             EventHubManager.CreateEventHubIfNotExists(eventHubName, numberOfPartitions, namespaceManager);
32.             
33.             //Create consumer group if it does not exist
34.             var group = namespaceManager.CreateConsumerGroupIfNotExists(eventHubName, consumerGroupName);
35.  
36.             //Start processing messages
37.             receiver = new Receiver(eventHubName, connectionString);
38.  
39.             //Get host name of worker role instance.  This is used for each environment to obtain
40.             //a lease, and to renew the same lease in case of a restart.
41.             string hostName = RoleEnvironment.CurrentRoleInstance.Id;
42.             receiver.RegisterEventProcessor(group, blobConnectionString, hostName);
43.  
44.             //Wait for shutdown to be called, else the role will recycle
45.             this.runCompleteEvent.WaitOne();
46.         }
47.  
48.         public override bool OnStart()
49.         {
50.             // Set the maximum number of concurrent connections
51.             ServicePointManager.DefaultConnectionLimit = 12;
52.  
53.             // For information on handling configuration changes
54.             // see the MSDN topic at https://go.microsoft.com/fwlink/?LinkId=166357.
55.             
56.             bool result = base.OnStart();
57.  
58.             Trace.TraceInformation("ReceiverRole has been started");
59.  
60.             return result;
61.         }
62.  
63.         public override void OnStop()
64.         {
65.             Trace.TraceInformation("ReceiverRole is stopping");
66.  
67.             this.runCompleteEvent.Set();
68.             try
69.             {
70.                 //Unregister the event processor so other instances
71.                 //  will handle the partitions
72.                 receiver.UnregisterEventProcessor();
73.             }
74.             catch(Exception oops)
75.             {
76.                 Trace.TraceError(oops.Message);
77.             }
78.             
79.             base.OnStop();
80.  
81.             Trace.TraceInformation("ReceiverRole has stopped");
82.         }
83.         
84.     }
85. }

When the worker role starts, we capture the trace output.  The worker role then starts to run, and we get settings from configuration in order to connect to the Event Hub and register our event processor.  A key point to call out is line 41 where we use the host name of the worker role instance, this is needed in order to obtain a lease for the partitions across multiple instances.  Also notice line 45 where we wait for a signal.  That signal is set on line 67 when the worker role is stopped.  The processing of events is handled on a separate thread, we only need to use the ManualResetEvent to prevent the Run method from completing (which would cause the worker role to recycle).

The sample code uses CloudConfigurationManager to obtain settings.  Those settings are part of the deployment process, shown in the next section.

Role Configuration

Go to the EventProcessor project that defines the deployment configuration for the worker role.  Expand the Roles folder and double-click on the ReceiverRole node.

When you double-click, go to the Settings tab.  This is where we provide the various configuration values.

Note: It is a best practice to avoid putting secrets such as connection strings in source code. You would typically set these as part of a deployment script. For more information, see Building Real-World Cloud Apps with Azure.  

You can also configure diagnostics for the worker roles.  We will use the diagnostics capabilities when we test out the solution.  On the Configuration tab, click the Configure button.

That allows you to configure the log level for the application.  I set it to Verbose.

Now right-click the deployment project and choose Publish.  You are prompted to create a cloud service and storage account.

Once created, go through the wizard to complete publishing.

Watch the progress in the Microsoft Azure Activity Log.

Once deployed, you will have a single production instance for your worker role.

Testing It Out

Once the role is deployed, go to Visual Studio and right-click the ReceiverRole node under Cloud Services and choose “View Diagnostics Data”.

On that page, expand the Microsoft Azure application logs section, then click View all data.

You will then see the Azure Storage table called WADLogsTable, and you can see the events that happened.  When the worker role is running, we see the entry “ReceiverRole is running”.  The SimpleEventProcessor is registered once, and the EventProcessorHost takes care of creating 16 instances, one for every partition (not all log entries are shown here, but there are 16 in the log). 

Let’s scale our service.  While we could use the autoscale service to scale by CPU, we can see how things work by manually scaling as well.  Let’s manually scale by going to the Azure management portal (https://manage.windowsazure.com) and setting the scale for the cloud service to 4.

Save, and wait while the new instances are created and our packages are deployed automatically to the new instances.

Once the 3 new virtual machines are created, we go back to see what happened in the logs.  This is really cool to see how the EventProcessorHost automatically balances the processing across the available instances.  You can see here that partitions 1,10, and 12 are handled by instance 2, while partitions 3,13, and 15 are handled by instance 1.

Let’s scale the number of instances down to 2. 

This will cause our OnStop() method to be called in each worker role that is no longer needed, and our code to unregister the event processor is called.  This balances the processing back to the remaining worker role instances.  First we see the close operations:

Next we see that the closed partitions are opened using another available processor:

Testing Out the Sender

To test things out, I can use the same client that I built in the post Use JMS and Azure Event Hubs with Eclipse.  This is a simple web page that sends messages to the Event Hub using AMQP 1.0.  Alternatively, I can modify the code in the Service Bus Event Hubs Getting Started sample to send messages.  Once I send messages to the Event Hub, we see them processed in the event log, automatically balanced across partitions among the 2 available worker roles.

Summary

Event Hubs make it very easy to build highly scalable solutions that can process millions of messages per second in near real time.  This post shows how you can use an Azure worker role with EventProcessorHost to process the messages and how the load will automatically balance across the available processors.

I am not providing the sample code as a download because it is already provided in the post Service Bus Event Hubs Getting Started, and I have listed the code that I used in this post. 

For More Information

Use JMS and Azure Event Hubs with Eclipse

Building Real-World Cloud Apps with Azure – Free eBook!

Service Bus Event Hubs Getting Started – sample used as the basis of this post

Comments

• Anonymous
  February 24, 2015
  The comment has been removed

• Anonymous
  February 24, 2015
  @Jim If I am not mistaken, and if I understand the question correctly, the lease should expire, allowing another instance to lease the partition and begin processing at the last recorded offset.

• Anonymous
  February 24, 2015
  Correct, the lease will expire and the other instances will take over that workload at the last recorded offset, which is set in blob storage when CheckpointAsync() is called.  In my example, I am aggressively calling context.CheckpointAsync().  The "getting started" sample that I based this from calls CheckpointAsync() on a timed basis, something like 5 minutes.  Other examples call CheckPointAsync() once every 100 messages.  

• Anonymous
  February 24, 2015
  @Kirk, @Luke I have waited over 30 minutes, and the existing other instances never pick up the work.  I have a LeaseInterval of 30 seconds and am calling CheckpointAsync() every 5 minutes (as long as new values are coming in, which they definitely are).  However, if I start up a new instance, then it gladly picks up the work that the dead instance was handling.

• Anonymous
  February 25, 2015

1. Are you using the 2.6 or 2.6.1 ServiceBus.dll and the 1.1.0 EventProcessorHost.dll - please confirm this.
2. When you say do not gracefully shut down - it sounds like you're doing something that is not turning off / disconnecting the machines. If the app is still running it's probably still renewing leases in the background. How are you simulating this non-graceful shutdown?

• Anonymous
  February 25, 2015
  The comment has been removed

• Anonymous
  February 25, 2015
  I have many questions about the partition count. What is the point of the partition count configuration in this example? It presumably stays at the maximum value even though the instance count increases. Why should this be a configuration value if it does not change when scaling? Why would you choose any particular partition count other than the maximum value? I am also assuming that you can't have more instances running than the partition count. Is this correct?

• Anonymous
  February 26, 2015
  How long are you waiting before starting the other reader? It should take a few refresh cycles (which the default is 30 seconds) - so try waiting at least 2 minutes.

• Anonymous
  February 26, 2015
  @Dan I have waited over 30 minutes to see if the existing instances would pick up the partitions that were not being read.  Those existing instances never have.  Even after 30 minutes I can start a new instance, and it instantly picks up the partitions that were not being read.

• Anonymous
  February 26, 2015
  The comment has been removed

• Anonymous
  February 26, 2015
  Also, there is a great tool for all things service bus in Azure. blogs.msdn.com/.../service-bus-explorer-2-5-now-available.aspx This tool will allow you to hammer a hub or send a single message. Highly recommended.

• Anonymous
  March 23, 2015
  Hi, is calling CheckpointAsync aggressively like this bad practice, as it could potentially run up large azure bills? Thanks,

• Anonymous
  March 23, 2015
  @Andy - the CheckPointAsync code really just writes a number to Azure blob storage that indicates the last position processed during its lease.  You'll have to load test your application to determine if the writes to blob storage, even though asynchronous, are impeding throughput.  Buffering that write even for 30 seconds could have a performance impact on your code, but it's up to you to perform load testing to determine the impact, if any.

• Anonymous
  February 07, 2016
  There is a potential syntax error in EventHubManager method GetServiceBusConnectionStringuse this:var connectionString = CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");instead of:string connectionString = Microsoft.WindowsAzure.CloudConfigurationManager.GetSetting("Microsoft.ServiceBus.ConnectionString");HTH.