receiveFromEventHubs

Source processor that consumes messages from Azure Event Hubs.

The consumer will listen for messages published to the specific Event Hubs instance (referenced by authenticationConfigKey). Consumer groups enable multiple consumers to collaborate on processing messages from multiple partitions concurrently. It is the main tool for scaling on the consumer side. Furthermore, consumer groups enable multiple consuming applications to each have a separate view of the event stream and to read the stream independently at their own pace and with their own offsets.

The message will not be acknowledged to Azure Event Hubs until it has been successfully persisted in the source processor.

The Lease Manager (ref. the 'leaseManagerCredentialsKey' property) is responsible for coordinating partition ownership and checkpointing among multiple consumers in an Event Hubs consumer group. It uses Azure Storage to:

Track which consumer instance owns which partition (partition leasing)
Ensure only one consumer processes a given partition at a time
Store checkpoints (the last successfully processed event) for each partition, enabling reliable recovery and load balancing.

This mechanism enables distributed, scalable, and fault-tolerant event processing.

Properties

Name Summary

Name	Summary
`authenticationConfigKey`	Key from the server configuration used to look up the credentials needed to connect to Azure Event Hubs.
`leaseManagerCredentialsKey`	Key from the server configuration used to look up the credentials needed for the lease manager.
`consumerGroupId`	The consumer group ID. Consumer groups enable multiple consumer instances to process messages from different partitions in parallel and support individual message offset tracking for multiple consuming applications.
`maxBatchSize`	The maximum batch size to consume.
`maxBatchWaitMillis`	The maximum wait time (in milliseconds) to wait for a batch to be consumed.
`prefetchCount`	The internal prefetch count in the Azure SDK. If not set, uses the SDK default of `500`. Legal values are between `1` and `8000` inclusive.
`initialPosition`	Position to start consuming from when no checkpoint is available. Can be one of the following string values: `end-of-stream`: Start consuming the newest messages. Default. `start-of-stream`: Start consuming from the oldest messages in retention. `<UTC date time>`: ISO 8601 compatible UTC date time with a hardcoded UTC `Z` time zone offset. (e.g., `YYYY-MM-DDThh:mm:ss.sZ`, or `2020-01-27T13:00:23.123Z`).
`enabled`	Whether the consumer should actually be polling messages from Event Hubs. Defaults to `true`.
`name`	Optional, descriptive name for the processor.
`id`	Required identifier of the processor, unique across all processors within the flow. Must be between 3 and 30 characters long; contain only lower and uppercase alphabetical characters (a-z and A-Z), numbers, dashes ("-"), and underscores ("_"); and start with an alphabetical character. In other words, it adheres to the regex pattern `[a-zA-Z][a-zA-Z0-9_-]{2,29}`.
`exchangeProperties`	Optional set of custom properties in a simple jdk-format, that are added to the message exchange properties before processing the incoming payload. Any existing properties with the same name will be replaced by properties defined here.
`retainPayloadOnFailure`	Whether the incoming payload is available for error processing on failure. Defaults to `false`.

authenticationConfigKey

Key from the server configuration used to look up the credentials needed to connect to Azure Event Hubs.

leaseManagerCredentialsKey

Key from the server configuration used to look up the credentials needed for the lease manager.

consumerGroupId

The consumer group ID. Consumer groups enable multiple consumer instances to process messages from different partitions in parallel and support individual message offset tracking for multiple consuming applications.

maxBatchSize

The maximum batch size to consume.

maxBatchWaitMillis

The maximum wait time (in milliseconds) to wait for a batch to be consumed.

prefetchCount

The internal prefetch count in the Azure SDK. If not set, uses the SDK default of 500. Legal values are between 1 and 8000 inclusive.

initialPosition

Position to start consuming from when no checkpoint is available. Can be one of the following string values:

end-of-stream: Start consuming the newest messages. Default.
start-of-stream: Start consuming from the oldest messages in retention.
<UTC date time>: ISO 8601 compatible UTC date time with a hardcoded UTC Z time zone offset. (e.g., YYYY-MM-DDThh:mm:ss.sZ, or 2020-01-27T13:00:23.123Z).

enabled

Whether the consumer should actually be polling messages from Event Hubs. Defaults to true.

name

Optional, descriptive name for the processor.

id

Required identifier of the processor, unique across all processors within the flow. Must be between 3 and 30 characters long; contain only lower and uppercase alphabetical characters (a-z and A-Z), numbers, dashes ("-"), and underscores ("_"); and start with an alphabetical character. In other words, it adheres to the regex pattern [a-zA-Z][a-zA-Z0-9_-]{2,29}.

exchangeProperties

Optional set of custom properties in a simple jdk-format, that are added to the message exchange properties before processing the incoming payload. Any existing properties with the same name will be replaced by properties defined here.

retainPayloadOnFailure

Whether the incoming payload is available for error processing on failure. Defaults to false.

Sub-builders

Name	Summary
externalSystemDetails	Strategy for describing the external system integration. Optional.
messageLoggingStrategy	Strategy for describing how a processor’s message is logged on the server.
payloadArchivingStrategy	Strategy for archiving payloads.
inboundTransformationStrategy	Strategy that customizes the conversion of an incoming payload by a processor (e.g., string to object). Should be used when the processor’s default conversion logic cannot be used.

Name

Summary

externalSystemDetails

Strategy for describing the external system integration. Optional.

messageLoggingStrategy

Strategy for describing how a processor’s message is logged on the server.

payloadArchivingStrategy

Strategy for archiving payloads.

inboundTransformationStrategy

Strategy that customizes the conversion of an incoming payload by a processor (e.g., string to object). Should be used when the processor’s default conversion logic cannot be used.

Details

Authentication

The authenticationConfigKey property supports secrets of type AzureConnectionString. The leaseManagerCredentialsKey property supports secrets of type AzureLeaseManagerCredentials. See the Secret Types documentation for formatting details.

Was this page helpful?