Pulsar Ingress Configuration

Pulsar data ingestion is available using ingress of type pulsar.

Examples Repo

Example Pulsar configuration files showcasing all features we will discuss can be found in our Examples Repo .

Settings

The Pulsar settings in the configuration file are used to configure data ingestion from Pulsar topics. Below is a detailed description of each field within the Pulsar settings:

Field	Required	Configurable	Description
`serviceUrl`	`true`	`true`	The URL of the Pulsar service to connect to.
`topic`	`true`	`true`	The name of the Pulsar topic to subscribe to.
`subscriptionName`	`true`	`true`	The name of the subscription for this consumer.
`valueSchema`	`true`	`true`	The schema type for the message value, e.g., `string`, `integer`, `avro`. For generic Avro schema use `avro`, the following two fields can then be omitted.
`valueContentType`	`false`	`true`	The content type of the value, accepted types: `avro`, `csv`, `json`, `recon`, `xml` (default `plaintext`).
`valueAvroSchema`	`false`	`true`	The Avro schema for the value, required iff `valueContentType`=`avro` AND `valueSchema` is “raw” (e.g. `BYTES`)
`clientLoadConf`	`false`	`false`	Additional Pulsar client configuration properties as key-value pairs.
`consumerLoadConf`	`false`	`false`	Additional Pulsar consumer configuration properties as key-value pairs.

Message Structure

The message structure of received Pulsar messages is:

{ "key": ..., "value": {...} }

Example

The configuration below defines a Pulsar ingress where messages are consumed from the specified Pulsar topic and relayed to the appropriate agents for processing.

{
  "type": "pulsar",
  "settings": {
    "serviceUrl": "my-pulsar-server:9092",
    "topic": "my-vehicle-topic",
    "subscriptionName": "subName",
    "valueSchema": "string",
    "valueContentType": "json",
  },
  "relay": {
    "agent": "vehicle",
    "idExtractor": "$key",
    "valueExtractor": "$value"
  }
}

Kafka Ingress Configuration

Redis Streams Ingress Configuration

Nstream is licensed under the Redis Source Available License 2.0 (RSALv2).