ConsumeAzureEventHub
Description
Receives messages from Microsoft Azure Event Hubs with checkpointing to ensure consistent event processing. Checkpoint tracking avoids consuming a message multiple times and enables reliable resumption of processing in the event of intermittent network failures. Checkpoint tracking requires external storage and provides the preferred approach to consuming messages from Azure Event Hubs. In clustered environment, ConsumeAzureEventHub processor instances form a consumer group and the messages are distributed among the cluster nodes (each message is processed on one cluster node only).
Tags
azure, cloud, eventhub, events, microsoft, streaming, streams
Properties
In the list below required Properties are shown with an asterisk (*). Other properties are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
| Display Name | API Name | Default Value | Allowable Values | Description |
|---|---|---|---|---|
| Event Hub Namespace * | event-hub-namespace | The namespace that the Azure Event Hubs is assigned to. This is generally equal to <Event Hub Names>-ns. Supports Expression Language, using Environment variables. | ||
| Event Hub Name * | event-hub-name | The name of the event hub to pull messages from. Supports Expression Language, using Environment variables. | ||
| Service Bus Endpoint * | Service Bus Endpoint | Azure |
| To support namespaces not in the default windows.net domain. |
| Transport Type * | Transport Type | AMQP |
| Advanced Message Queuing Protocol Transport Type for communication with Azure Event Hubs |
| Shared Access Policy Name | event-hub-shared-access-policy-name | The name of the shared access policy. This policy must have Listen claims. Supports Expression Language, using Environment variables. | ||
| Shared Access Policy Key | event-hub-shared-access-policy-primary-key | The key of the shared access policy. Either the primary or the secondary key can be used. | ||
| Use Azure Managed Identity * | use-managed-identity | false |
| Choose whether or not to use the managed identity of Azure VM/VMSS |
| Consumer Group * | event-hub-consumer-group | $Default | The name of the consumer group to use. Supports Expression Language, using Environment variables. | |
| Record Reader | record-reader | Controller Service: RecordReaderFactory Implementations: AvroReader CEFReader CSVReader ExcelReader GrokReader JsonPathReader JsonTreeReader ReaderLookup ScriptedReader Syslog5424Reader SyslogReader WindowsEventLogReader XMLReader YamlTreeReader | The Record Reader to use for reading received messages. The event hub name can be referred by Expression Language '${eventhub.name}' to access a schema. | |
| Record Writer | record-writer | Controller Service: RecordSetWriterFactory Implementations: AvroRecordSetWriter CSVRecordSetWriter FreeFormTextRecordSetWriter JsonRecordSetWriter RecordSetWriterLookup ScriptedRecordSetWriter XMLRecordSetWriter | The Record Writer to use for serializing Records to an output FlowFile. The event hub name can be referred by Expression Language '${eventhub.name}' to access a schema. If not specified, each message will create a FlowFile. | |
| Initial Offset * | event-hub-initial-offset | End of stream |
| Specify where to start receiving messages if offset is not yet stored in the checkpoint store. |
| Prefetch Count * | event-hub-prefetch-count | 300 | Supports Expression Language, using Environment variables. | |
| Batch Size * | event-hub-batch-size | 10 | The number of messages to process within a NiFi session. This parameter affects throughput and consistency. NiFi commits its session and Event Hubs checkpoints after processing this number of messages. If NiFi session is committed, but fails to create an Event Hubs checkpoint, then it is possible that the same messages will be received again. The higher number, the higher throughput, but possibly less consistent. Supports Expression Language, using Environment variables. | |
| Message Receive Timeout * | event-hub-message-receive-timeout | 1 min | The amount of time this consumer should wait to receive the Batch Size before returning. Supports Expression Language, using Environment variables. | |
| Checkpoint Strategy * | checkpoint-strategy | Azure Blob Storage |
| Specifies which strategy to use for storing and retrieving partition ownership and checkpoint information for each partition. |
| Storage Account Name * | storage-account-name | Name of the Azure Storage account to store event hub consumer group state. Supports Expression Language, using Environment variables. This property is only considered if:
| ||
| Storage Account Key | storage-account-key | The Azure Storage account key to store event hub consumer group state. Supports Expression Language, using Environment variables. This property is only considered if:
| ||
| Storage SAS Token | storage-sas-token | The Azure Storage SAS token to store Event Hub consumer group state. Always starts with a ? character. Supports Expression Language, using Environment variables. This property is only considered if:
| ||
| Storage Container Name | storage-container-name | Name of the Azure Storage container to store the event hub consumer group state. If not specified, event hub name is used. Supports Expression Language, using Environment variables. This property is only considered if:
| ||
| Proxy Configuration Service | proxy-configuration-service | Controller Service: ProxyConfigurationService Implementations: StandardProxyConfigurationService | Specifies the Proxy Configuration Controller Service to proxy network requests. Supported proxies: HTTP + AuthN This property is only considered if:
|
Dynamic Properties
This component does not support dynamic properties.
Relationships
| Name | Description |
|---|---|
| success | FlowFiles received from Event Hub. |
Reads Attributes
This processor does not read attributes.
Writes Attributes
| Name | Description |
|---|---|
| eventhub.enqueued.timestamp | The time (in milliseconds since epoch, UTC) at which the message was enqueued in the event hub |
| eventhub.name | The name of the event hub from which the message was pulled |
| eventhub.offset | The offset into the partition at which the message was stored |
| eventhub.partition | The name of the partition from which the message was pulled |
| eventhub.property.* | The application properties of this message. IE: 'application' would be 'eventhub.property.application' |
| eventhub.sequence | The sequence number associated with the message |
State Management
| Scope | Description |
|---|---|
| CLUSTER, LOCAL | Local state is used to store the client id. Cluster state is used to store partition ownership and checkpoint information when component state is configured as the checkpointing strategy. |
Restricted
This component is not restricted.
Input Requirement
This component does not allow an incoming relationship.
System Resource Considerations
This component does not specify system resource considerations.