Sends the contents of a FlowFile as a message to Apache Pulsar using the Pulsar Producer API.The messages to send may be individual FlowFiles or may be delimited, using a user-specified delimiter, such as a new-line. The complementary NiFi processor for fetching messages is ConsumePulsar.
| Display Name | API Name | Default Value | Allowable Values | Description |
|---|
| Pulsar Client Service | PULSAR_CLIENT_SERVICE | | Controller Service API:
PulsarClientService
Implementation: StandardPulsarClientService | Specified the Pulsar Client Service that can be used to create Pulsar connections |
| Topic Name | TOPIC | | | The name of the Pulsar Topic.
Supports Expression Language: true (will be evaluated using flow file attributes and variable registry) |
| Async Enabled | ASYNC_ENABLED | false | | Control whether the messages will be sent asynchronously or not. Messages sent synchronously will be acknowledged immediately before processing the next message, while asynchronous messages will be acknowledged after the Pulsar broker responds. Running the processor with async enabled will result in increased the throughput at the risk of potential duplicate data being sent to the Pulsar broker. |
| Maximum Async Requests | MAX_ASYNC_REQUESTS | 50 | | The maximum number of outstanding asynchronous publish requests for this processor. Each asynchronous call requires memory, so avoid setting this value to high. |
| Batching Enabled | BATCHING_ENABLED | false | | Control whether automatic batching of messages is enabled for the producer. default: false [No batching] When batching is enabled, multiple calls to Producer.sendAsync can result in a single batch to be sent to the broker, leading to better throughput, especially when publishing small messages. If compression is enabled, messages will be compressed at the batch level, leading to a much better compression ratio for similar headers or contents. When enabled default batch delay is set to 10 ms and default batch size is 1000 messages |
| Batching Max Messages | BATCHING_MAX_MESSAGES | 1000 | | Set the maximum number of messages permitted in a batch within the Pulsar client. default: 1000. If set to a value greater than 1, messages will be queued until this threshold is reached or the batch interval has elapsed, whichever happens first.
Supports Expression Language: true (will be evaluated using variable registry only) |
| Batch Interval | BATCH_INTERVAL | 10 ms | | Set the time period within which the messages sent will be batched if batch messages are enabled. If set to a non zero value, messages will be queued until this time interval has been reached OR until the Batching Max Messages threshould has been reached, whichever occurs first.
Supports Expression Language: true (will be evaluated using variable registry only) |
| Block if Message Queue Full | BLOCK_IF_QUEUE_FULL | false | | Set whether the processor should block when the outgoing message queue is full. Default is false. If set to false, send operations will immediately fail with ProducerQueueIsFullError when there is no space left in pending queue. |
| Compression Type | COMPRESSION_TYPE | None | | Set the compression type for the producer. |
| Message Routing Mode | MESSAGE_ROUTING_MODE | Round Robin Partition | - Custom Partition
 - Round Robin Partition
- Single Partition
| Set the message routing mode for the producer. This applies only if the destination topic is partitioned |
| Message Demarcator | MESSAGE_DEMARCATOR | | | Specifies the string (interpreted as UTF-8) to use for demarcating multiple messages within a single FlowFile. If not specified, the entire content of the FlowFile will be used as a single message. If specified, the contents of the FlowFile will be split on this delimiter and each section sent as a separate Pulsar message. To enter special character such as 'new line' use CTRL+Enter or Shift+Enter, depending on your OS.
Supports Expression Language: true (will be evaluated using flow file attributes and variable registry) |
| Max Pending Messages | PENDING_MAX_MESSAGES | 1000 | | Set the max size of the queue holding the messages pending to receive an acknowledgment from the broker.
Supports Expression Language: true (will be evaluated using variable registry only) |
| Mapped Message Properties | MAPPED_MESSAGE_PROPERTIES | | | A comma-delimited list of message properties to set based on FlowFile attributes. Syntax for an individual property entry is <property name>[=<source attribute name>]. If the optional source attribute name is omitted, it is assumed to be the same as the property. |
| Message Key | MESSAGE_KEY | | | The key of the outgoing message.
Supports Expression Language: true (will be evaluated using flow file attributes and variable registry) |