In the last blog we talked about the basics of the EventProcessorHost. In this blog we’ll talk about some deeper details around lease management and EventProcessorHost options.
Checkpointing is not the only use of the storage connection string performed by EventProcessorHost. Partition ownership (that is reader ownership) is also performed for you. This way only a single reader can read from any given partition at a time within a consumer group. This is accomplished using Azure Storage Blob Leases and implemented using Epoch. This greatly simplifies the auto-scale nature of EventProcessorHost. As an instance of EventProcessorHost starts it will acquire as many leases as possible and begin reading events. As the leases draw near expiration EventProcessorHost will attempt to renew them by placing a reservation. If the lease is available for renewal the processor continues reading, but if it is not the reader is closed and CloseAsync is called – this is a good time to perform any final cleanup for that partition.
EventProcessorHost has a member PartitionManagerOptions. This member allows for control over lease management. Set these options before registering your IEventProcessor implementation.
Controlling the runtime
Additionally the call to RegisterEventProcessorAsync allows for a parameter EventProcessorOptions. This is where you can control the behavior of the EventProcessorHost itself. There are four properties and one event that you should be aware of.
MaxBatchSize – this is the maximum size of the collection the user wants to receive in an invocation of ProcessEventsAsync. Note that this is not the minimum, only the maximum. If there are not this many messages to be received the ProcessEventsAsync will execute with as many as were available.
PrefetchCount – this is a value used by the underlying AMQP channel to determine the upper limit of how many messages the client should receive. This value should be greater than or equal to MaxBatchSize.
InvokeProcessorAfterReceiveTimeout – setting this parameter to true will result in ProcessEventsAsync being called when the underlying call the receive events on a partition times out. This is useful for taking time based actions during periods of inactivity on the partition.
InitialOffsetProvider – this allows a function pointer or lambda expression to be set that will be called to provide the initial offset when a reader begins reading a partition. Without setting this the reader will start at the oldest event unless a JSON file with an offset has already been saved in the storage account supplied to the EventProcessorHost constructor. This is useful when you want to change the behavior of reader start up. When this method is invoked the object parameter will contain the partition id that the reader is being started for.
ExceptionReceived – this event allows you to receive notification of any underlying exceptions that occur in the EventProcessorHost. If things aren’t working as you expect, this is a great place to start looking.
If there is one best practice you take from this blog make sure it is to use the ExceptionReceived event when things don’t work the way you expect.