x/pulsarClient component: v0.33.0+ Pulsar producer component. Used to publish messages to specified Pulsar topics.

Configuration

This component supports reusing shared Pulsar connection clients through the server field to avoid creating duplicate connections. See Component Connection Reuse.

Field	Type	Required	Description	Default Value
server	string	Yes	Pulsar server address, format: pulsar://host:port	pulsar://localhost:6650
topic	string	Yes	Publish topic, supports using Component Configuration Variables for dynamic configuration	/device/msg
key	string	No	Message key template, supports using Component Configuration Variables for dynamic configuration, used for message routing and partitioning	None
headers	map[string]string	No	Custom message properties, supports both key and value using Component Configuration Variables for dynamic configuration	None
authToken	string	No	Pulsar JWT authentication token	None
certFile	string	No	TLS certificate file path	None
certKeyFile	string	No	TLS private key file path	None

Working Principle

1. The component initializes by connecting to the Pulsar cluster based on configuration
2. Creates a producer instance for sending messages
3. Upon receiving a message, publishes the message content to the specified topic
4. Supports setting message key and custom headers properties, both key and value support variable replacement
5. Uses template engine to process variable replacement in topic, key, and headers
6. Routes through Success chain upon successful publishing, routes through Failure chain on failure

Relation Type

• Success: Message is sent to Success chain in the following cases:
  • Message successfully published to Pulsar cluster
  • Received confirmation from Pulsar server
• Failure: Message is sent to Failure chain in the following cases:
  • Failed to connect to Pulsar cluster
  • Failed to create producer
  • Failed to publish message
  • Configuration parameter error
  • TLS certificate loading failure
  • Template parsing failure
  • Topic is empty

Execution Result

After component execution:

• msg.data remains unchanged
• metadata remains unchanged
• msgType remains unchanged

Configuration Examples

Basic Configuration

{
  "id": "s5",
  "type": "x/pulsarClient",
  "name": "Publish to Pulsar",
  "debugMode": true,
  "configuration": {
    "server": "pulsar://localhost:6650",
    "topic": "persistent://public/default/device-msg"
  }
}

Advanced Configuration

{
  "id": "s6",
  "type": "x/pulsarClient",
  "name": "Publish to Pulsar with Properties",
  "debugMode": true,
  "configuration": {
    "server": "pulsar://localhost:6650",
    "topic": "persistent://public/default/device-${deviceType}",
    "key": "${deviceId}",
    "headers": {
      "source": "${source}",
      "timestamp": "${ts}",
      "deviceType": "${deviceType}"
    },
    "authToken": "your-jwt-token"
  }
}

Application Example

Application example reference: pulsar_client_test (opens new window)