Apache Pulsar

https://pulsar.apache.org/docs
Admin
This portal holds a variety of topics, tutorials, guides, and reference material to help you work...
Tokens:497,994
Snippets:5,816
Trust Score:10
Update:3 months ago
Show doc for...
Tokens:
Raw
### Start MongoDB Example Container

Source: https://pulsar.apache.org/docs/io-debezium-source

Pulls and runs a Debezium example MongoDB container. Ensure this is accessible by your Pulsar instance.

```bash
docker pull debezium/example-mongodb:0.10

docker run -d -it --rm --name pulsar-mongodb -e MONGODB_USER=mongodb -e MONGODB_PASSWORD=mongodb -p 27017:27017  debezium/example-mongodb:0.10
```

--------------------------------

### Start Example MySQL Server

Source: https://pulsar.apache.org/docs/5.0.x/io-cdc-debezium

Launches a MySQL server instance in a Docker container for Debezium to capture changes from. Ensure this is running before proceeding.

```bash
docker run -it --rm \
--name mysql \
-p 3306:3306 \
-e MYSQL_ROOT_PASSWORD=debezium \
-e MYSQL_USER=mysqluser \
-e MYSQL_PASSWORD=mysqlpw debezium/example-mysql:0.8  
```

--------------------------------

### Create and use Go readers for specific message ranges

Source: https://pulsar.apache.org/docs/client-libraries/readers

This example shows how to create Go readers to consume messages starting from a specific message ID, with options to include or exclude the starting message itself. It also includes setup for producing messages to test the reader functionality.

```go
client, err := pulsar.NewClient(pulsar.ClientOptions{
    URL: "pulsar://localhost:6650",
})

if err != nil {
    log.Fatal(err)
}
def συντελεστές client.Close()

topic := "topic-1"
ctx := context.Background()

// create producer
producer, err := client.CreateProducer(pulsar.ProducerOptions{
    Topic:           topic,
    DisableBatching: true,
})
if err != nil {
    log.Fatal(err)
}
def συντελεστές producer.Close()

// send 10 messages
msgIDs := [10]pulsar.MessageID{}
for i := 0; i < 10; i++ {
    msgID, _ := producer.Send(ctx, &pulsar.ProducerMessage{
        Payload: []byte(fmt.Sprintf("hello-%d", i)),
    })
    msgIDs[i] = msgID
}

// create reader on 5th message (not included)
reader, err := client.CreateReader(pulsar.ReaderOptions{
    Topic:                   topic,
    StartMessageID:          msgIDs[4],
    StartMessageIDInclusive: false,
})

if err != nil {
    log.Fatal(err)
}
def συντελεστές reader.Close()

// receive the remaining 5 messages
for i := 5; i < 10; i++ {
    msg, err := reader.Next(context.Background())
    if err != nil {
        log.Fatal(err)
    }

    fmt.Printf("Read %d-th msg: %s\n", i, string(msg.Payload()))
}
// create reader on 5th message (included)
readerInclusive, err := client.CreateReader(pulsar.ReaderOptions{
    Topic:                   topic,
    StartMessageID:          msgIDs[4],
    StartMessageIDInclusive: true,
})

if err != nil {
    log.Fatal(err)
}
def συντελεστές readerInclusive.Close()
```

--------------------------------

### Start MySQL Server

Source: https://pulsar.apache.org/docs/5.0.x/io-debezium-source

Starts a MySQL server in a Docker container for Debezium to capture changes from. Ensure the Debezium MySQL example image is used.

```bash
docker run -it --rm \
  --name mysql \
  -p 3306:3306 \
  -e MYSQL_ROOT_PASSWORD=debezium \
  -e MYSQL_USER=mysqluser \
  -e MYSQL_PASSWORD=mysqlpw debezium/example-mysql:0.8  
```

--------------------------------

### Install Pulsar Go Client with go get

Source: https://pulsar.apache.org/docs/client-libraries/go-setup

Use 'go get' to download the Pulsar Go client library to your local environment. This command fetches the latest version of the library.

```go
go get -u "github.com/apache/pulsar-client-go/pulsar"
```

--------------------------------

### List Brokers Output Example

Source: https://pulsar.apache.org/docs/5.0.x/develop-load-manager

Example output showing a list of brokers available in the cluster.

```text
[broker-1:8080, broker-2:8080]
```

--------------------------------

### Get Message TTL Example Output

Source: https://pulsar.apache.org/docs/5.0.x/admin-api-namespaces

Example output when retrieving the message TTL for a namespace, showing the configured TTL in seconds.

```text
100
```

--------------------------------

### Start Python Function with Configuration File

Source: https://pulsar.apache.org/docs/5.0.x/functions-runtime-process

This example demonstrates how to run a Python Pulsar Function locally using a configuration file. The command-line arguments take precedence over settings in the configuration file.

```bash
pulsar-admin functions localrun \
  --py /path/to/python_instance.py \
  --config-file /path/to/config.ini \
  --classname MyFunction \
  --logging_level debug \
  --inputs persistent://public/default/my-input-topic \
  --output persistent://public/default/my-output-topic \
  --log-topic persistent://public/default/functions-logs  

```

--------------------------------

### Install ws package

Source: https://pulsar.apache.org/docs/client-libraries/websocket

Install the 'ws' package for Node.js using npm. This is a prerequisite for the following examples.

```bash
npm install ws  

```

--------------------------------

### Get Retention Policies Example Output

Source: https://pulsar.apache.org/docs/5.0.x/admin-api-namespaces

Example JSON output when retrieving retention policies for a namespace, showing retention time and size.

```json
{
  "retentionTimeInMinutes": 10,
  "retentionSizeInMB": 100
}
```

--------------------------------

### Get Function Information by Tenant, Namespace, and Name

Source: https://pulsar.apache.org/docs/5.0.x/functions-debug-cli

Alternatively, use the `get` command specifying tenant, namespace, and name to get function details. This is useful when the FQFN is not readily available.

```bash
./bin/pulsar-admin functions get \
    --tenant public \
    --namespace default \
    --name ExclamationFunctio6
```

--------------------------------

### Import Pulsar Go Client in Project

Source: https://pulsar.apache.org/docs/client-libraries/go-setup

After installing the library, import it into your Go project to start using its functionalities.

```go
import "github.com/apache/pulsar-client-go/pulsar"
```

--------------------------------

### Start MongoDB Docker Container

Source: https://pulsar.apache.org/docs/5.0.x/io-cdc-debezium

Pulls and runs a Debezium example MongoDB Docker image. This is the first step to set up a source database for change data capture.

```bash
docker pull debezium/example-mongodb:0.10
```

```bash
docker run -d -it --rm --name pulsar-mongodb -e MONGODB_USER=mongodb -e MONGODB_PASSWORD=mongodb -p 27017:27017  debezium/example-mongodb:0.10
```

--------------------------------

### Get Schema CLI Example Output

Source: https://pulsar.apache.org/docs/5.0.x/schema-manage

Example output when retrieving a schema using the Admin CLI, showing version, type, timestamp, data, and properties.

```json
{
    "version": 0,
    "type": "String",
    "timestamp": 0,
    "data": "string",
    "properties": {
        "property1": "string",
        "property2": "string"
    }
}
```

--------------------------------

### Start Source Connector

Source: https://pulsar.apache.org/docs/5.0.x/io-use

Starts source connectors, either all or a specified instance.

```APIDOC
## POST /admin/v3/sources/{tenant}/{namespace}/{sourceName}/start

### Description
Starts all source connectors for a given tenant, namespace, and source name.

### Method
POST

### Endpoint
/admin/v3/sources/{tenant}/{namespace}/{sourceName}/start

### Response
#### Success Response (200)
- (No specific fields mentioned, indicates success)

## POST /admin/v3/sources/{tenant}/{namespace}/{sourceName}/{instanceId}/start

### Description
Starts a specified instance of a source connector.

### Method
POST

### Endpoint
/admin/v3/sources/{tenant}/{namespace}/{sourceName}/{instanceId}/start

### Response
#### Success Response (200)
- (No specific fields mentioned, indicates success)
```

--------------------------------

### Install websocket-client

Source: https://pulsar.apache.org/docs/client-libraries/websocket

Install the necessary Python package for WebSocket communication.

```bash
pip install websocket-client
```

--------------------------------

### Pulsar Consumer with Prometheus Metrics

Source: https://pulsar.apache.org/docs/client-libraries/go-use

This example sets up a Pulsar consumer and simultaneously starts an HTTP server to expose Prometheus metrics. It also includes a web endpoint to consume messages.

```Go
// Create a Pulsar client

client, err := pulsar.NewClient(pulsar.ClientOptions{

    URL: "pulsar://localhost:6650",

})

if err != nil {

    log.Fatal(err)

}

  

defer client.Close()

  

// Start a separate goroutine for Prometheus metrics

// In this case, Prometheus metrics can be accessed via http://localhost:2112/metrics

go func() {

    prometheusPort := 2112

    log.Printf("Starting Prometheus metrics at http://localhost:%v/metrics\n", prometheusPort)

    http.Handle("/metrics", promhttp.Handler())

    err = http.ListenAndServe(":"+strconv.Itoa(prometheusPort), nil)

    if err != nil {

        log.Fatal(err)

    }

}()

  

// Create a consumer

consumer, err := client.Subscribe(pulsar.ConsumerOptions{

    Topic:            "topic-1",

    SubscriptionName: "sub-1",

    Type:             pulsar.Shared,

})

if err != nil {

    log.Fatal(err)

}

  

defer consumer.Close()

  

ctx := context.Background()

  

// Write your business logic here

// In this case, you build a simple Web server. You can consume messages by requesting http://localhost:8083/consume

webPort := 8083

http.HandleFunc("/consume", func(w http.ResponseWriter, r *http.Request) {

    msg, err := consumer.Receive(ctx)

    if err != nil {

        log.Fatal(err)

    } else {

        log.Printf("Received message msgId: %v -- content: '%s'\n", msg.ID(), string(msg.Payload()))

        fmt.Fprintf(w, "Received message msgId: %v -- content: '%s'\n", msg.ID(), string(msg.Payload()))

        consumer.Ack(msg)

    }

})

  

err = http.ListenAndServe(":"+strconv.Itoa(webPort), nil)

if err != nil {

    log.Fatal(err)

}
```

--------------------------------

### Start a source instance

Source: https://pulsar.apache.org/docs/io-cli

Starts a Pulsar IO source instance. Can start all instances or a specific instance if instance-id is provided.

```APIDOC
## start

### Description
Start a source instance.

### Usage
```
pulsar-admin sources start --tenant <tenant> --namespace <namespace> --name <name> [--instance-id <instance-id>]
```

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Options
* `--instance-id` (string) - Optional - The source instanceID. If not provided, Pulsar starts all instances.
* `--name` (string) - Required - The source's name.
* `--namespace` (string) - Required - The source's namespace.
* `--tenant` (string) - Required - The source's tenant.
```

--------------------------------

### Example Source Connector Configuration

Source: https://pulsar.apache.org/docs/5.0.x/io-use

An example JSON object representing the configuration of a source connector.

```json
{
 
 "tenant": "tenantName",  
 "namespace": "namespaceName",  
 "name": "sourceName",  
 "className": "className",  
 "topicName": "topicName",  
 "configs": {},  
 "parallelism": 1,  
 "processingGuarantees": "ATLEAST_ONCE",  
 "resources": {  
   "cpu": 1.0,  
   "ram": 1073741824,  
   "disk": 10737418240  
 }
}

```

--------------------------------

### Node.js Client: Producer, Consumer, and Encryption Setup

Source: https://pulsar.apache.org/docs/5.0.x/security-encryption

Example of creating a Pulsar client, producer, and consumer with encryption enabled in Node.js. Requires specifying public and private key paths for encryption.

```javascript
const Pulsar = require('pulsar-client');

const topic = 'persistent://my-tenant/my-ns/my-topic';

(async () => {
// Create a client
const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
    operationTimeoutSeconds: 30,
});

// Create a producer
const producer = await client.createProducer({
    topic: topic,
    sendTimeoutMs: 30000,
    batchingEnabled: true,
    publicKeyPath: "test_rsa_pubkey.pem",
    encryptionKey: "encryption-key"
});

// Create a consumer
const consumer = await client.subscribe({
    topic: topic,
    subscription: 'my-subscriber-name',
    subscriptionType: 'Shared',
    ackTimeoutMs: 10000,
    privateKeyPath: "test_rsa_privkey.pem"
});
await consumer.close();
await producer.close();
await client.close();
})();

```

--------------------------------

### Install Pulsar Python Client with Optional Components

Source: https://pulsar.apache.org/docs/client-libraries/python-setup

Install the Pulsar Python client with optional components like Avro serialization or functions runtime. Use '[all]' to install all optional components.

```bash
# avro serialization

pip install 'pulsar-client[avro]==3.12.0'

# functions runtime

pip install 'pulsar-client[functions]==3.12.0'

# all optional components

pip install 'pulsar-client[all]==3.12.0'
```

--------------------------------

### Start Sink Connector

Source: https://pulsar.apache.org/docs/5.0.x/io-use

Starts sink connectors, either all or a specified instance.

```APIDOC
## POST /admin/v3/sinks/{tenant}/{namespace}/{sinkName}/start

### Description
Starts all sink connectors for a given tenant, namespace, and sink name.

### Method
POST

### Endpoint
/admin/v3/sinks/{tenant}/{namespace}/{sinkName}/start

### Response
#### Success Response (200)
- (No specific fields mentioned, indicates success)

## POST /admin/v3/sinks/{tenant}/{namespace}/{sinkName}/{instanceId}/start

### Description
Starts a specified instance of a sink connector.

### Method
POST

### Endpoint
/admin/v3/sinks/{tenant}/{namespace}/{sinkName}/{instanceId}/start

### Response
#### Success Response (200)
- (No specific fields mentioned, indicates success)
```

--------------------------------

### Start all instances of a Pulsar Function using Admin CLI

Source: https://pulsar.apache.org/docs/5.0.x/admin-api-functions

Use the 'start' subcommand without an instance ID to start all stopped function instances via the Admin CLI.

```bash
pulsar-admin functions start \
    --tenant public \
    --namespace default \
    --name (the name of Pulsar Functions) \
  
```

--------------------------------

### String SchemaInfo Example

Source: https://pulsar.apache.org/docs/5.0.x/schema-understand

This is a string example of SchemaInfo, demonstrating its structure with name, type, schema, and properties.

```json
{
    "name": "test-string-schema",
    "type": "STRING",
    "schema": "",
    "properties": {}
}
```

--------------------------------

### Initialize Go Module and Download Dependencies

Source: https://pulsar.apache.org/docs/client-libraries/go-setup

Set up a new Go module for your project and download necessary dependencies. This is part of the 'go module' installation process.

```go
mkdir test_dir && cd test_dir
go mod init test_dir
go mod tidy && go mod download
go build test_example.go
./test_example
```

--------------------------------

### Quick Install Pulsar Manager with Docker

Source: https://pulsar.apache.org/docs/administration-pulsar-manager

Use this command to pull the latest Pulsar Manager Docker image and run it. This is the simplest way to get started.

```bash
docker pull apachepulsar/pulsar-manager:latest  

docker run -it \
  -p 9527:9527 -p 7750:7750 \
  -e SPRING_CONFIGURATION_FILE=/pulsar-manager/pulsar-manager/application.properties \
  apachepulsar/pulsar-manager:latest  
```

--------------------------------

### Get a Specific Schema (Java Example)

Source: https://pulsar.apache.org/docs/5.0.x/admin-api-schemas

Example Java code to get a specific schema version using the PulsarAdmin client.

```java
PulsarAdmin admin = …;

SchemaInfo si = admin.getSchema("my-tenant/my-ns/my-topic", 1L);
```

--------------------------------

### Setup Producers and Consumers

Source: https://pulsar.apache.org/docs/5.0.x/txn-use

Creates producers for input and output topics, and consumers for input and output topics. Each producer and consumer is configured with appropriate options.

```go
inputTopic := "inputTopic"
outputTopicOne := "outputTopicOne"
outputTopicTwo := "outputTopicTwo"
subscriptionName := "your-subscription-name"

inputProducer, _ := client.CreateProducer(pulsar.ProducerOptions{
  Topic:       inputTopic,
  SendTimeout: 0,
})
deferr inputProducer.Close()

outputProducerOne, _ := client.CreateProducer(pulsar.ProducerOptions{
  Topic:       outputTopicOne,
  SendTimeout: 0,
})
deferr outputProducerOne.Close()

outputProducerTwo, _ := client.CreateProducer(pulsar.ProducerOptions{
  Topic:       outputTopicTwo,
  SendTimeout: 0,
})
deferr outputProducerTwo.Close()

inputConsumer, _ := client.Subscribe(pulsar.ConsumerOptions{
  Topic:            inputTopic,
  SubscriptionName: subscriptionName,
})
deferr inputConsumer.Close()

outputConsumerOne, _ := client.Subscribe(pulsar.ConsumerOptions{
  Topic:            outputTopicOne,
  SubscriptionName: subscriptionName,
})
deferr outputConsumerOne.Close()

outputConsumerTwo, _ := client.Subscribe(pulsar.ConsumerOptions{
  Topic:            outputTopicTwo,
  SubscriptionName: subscriptionName,
})
deferr outputConsumerTwo.Close()
```

--------------------------------

### Full Consumer Example with Client Initialization

Source: https://pulsar.apache.org/docs/client-libraries/node-use

This example demonstrates creating a Pulsar client, subscribing to a topic with a specific subscription type, receiving and acknowledging messages in a loop, and then closing both the consumer and the client.

```javascript
const Pulsar = require('pulsar-client');

(async () => {
  // Create a client
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
  });

  // Create a consumer
  const consumer = await client.subscribe({
    topic: 'my-topic',
    subscription: 'my-subscription',
    subscriptionType: 'Exclusive',
  });

  // Receive messages
  for (let i = 0; i < 10; i += 1) {
    const msg = await consumer.receive();
    console.log(msg.getData().toString());
    consumer.acknowledge(msg);
  }

  await consumer.close();
  await client.close();
})();
```

--------------------------------

### Start Bookie Daemon

Source: https://pulsar.apache.org/docs/5.0.x/deploy-bare-metal-multi-cluster

Use this command to start a BookKeeper bookie as a background daemon. Ensure you are in the Pulsar installation directory.

```bash
bin/pulsar-daemon start bookie  
```

--------------------------------

### Example Output: Monitor Brokers (ZooKeeper, Simplified)

Source: https://pulsar.apache.org/docs/5.0.x/develop-load-manager

This is a simplified example output for monitoring brokers via ZooKeeper, focusing on system and count metrics.

```bash
===================================================================================================================
||SYSTEM         |CPU %          |MEMORY %       |DIRECT %       |BW IN %        |BW OUT %       |MAX %          ||
||               |0.00           |48.33          |0.01           |0.00           |0.00           |48.33          ||
||COUNT          |TOPIC          |BUNDLE         |PRODUCER       |CONSUMER       |BUNDLE +       |BUNDLE -       ||
||               |4              |4              |0              |2              |4              |0              ||
||LATEST         |MSG/S IN       |MSG/S OUT      |TOTAL          |KB/S IN        |KB/S OUT       |TOTAL          ||
||               |0.00           |0.00           |0.00           |0.00           |0.00           |0.00           ||
||SHORT          |MSG/S IN       |MSG/S OUT      |TOTAL          |KB/S IN        |KB/S OUT       |TOTAL          ||
||               |0.00           |0.00           |0.00           |0.00           |0.00           |0.00           ||
||LONG           |MSG/S IN       |MSG/S OUT      |TOTAL          |KB/S IN        |KB/S OUT       |TOTAL          ||
||               |0.00           |0.00           |0.00           |0.00           |0.00           |0.00           ||
===================================================================================================================
```

--------------------------------

### Start Cassandra Docker Container

Source: https://pulsar.apache.org/docs/5.0.x/io-quickstart

Starts a single-node Cassandra cluster using a Docker image. Ensure Docker is installed and running.

```docker
docker run -d --rm --name=cassandra -p 9042:9042 cassandra:3.11
```

--------------------------------

### Start Simulation Controller

Source: https://pulsar.apache.org/docs/5.0.x/develop-tools

Initiates the simulation controller. Specify the target cluster and the port for client communication. Ensure clients are already running.

```bash
pulsar-perf simulation-controller --cluster <cluster to simulate on> --client-port <listen port for clients>  
--clients <comma-separated list of client host names>
```

--------------------------------

### Example: Configure Rack-Aware Placement Policy

Source: https://pulsar.apache.org/docs/5.0.x/administration-isolation-bookie

Configure bookie instances with their rack properties. This example demonstrates setting up multiple bookies within specific racks and groups.

```bash
bin/pulsar-admin bookies set-bookie-rack --bookie bookie1:3181 --hostname bookie1.pulsar.com:3181 --group group1 --rack rack1  
bin/pulsar-admin bookies set-bookie-rack --bookie bookie2:3181 --hostname bookie2.pulsar.com:3181 --group group1 --rack rack1  
bin/pulsar-admin bookies set-bookie-rack --bookie bookie3:3181 --hostname bookie3.pulsar.com:3181 --group group1 --rack rack1  
bin/pulsar-admin bookies set-bookie-rack --bookie bookie4:3181 --hostname bookie4.pulsar.com:3181 --group group1 --rack rack1  
bin/pulsar-admin bookies set-bookie-rack --bookie bookie5:3181 --hostname bookie5.pulsar.com:3181 --group group1 --rack rack2  
...
```

--------------------------------

### Create a Single-Topic Pulsar Consumer

Source: https://pulsar.apache.org/docs/client-libraries/go-use

This example shows how to initialize a Pulsar client and create a consumer that subscribes to a specific topic. Ensure the Pulsar client is properly configured.

```Go
client, err := pulsar.NewClient(pulsar.ClientOptions{URL: "pulsar://localhost:6650"})

if err != nil {

    log.Fatal(err)

}

  

defer client.Close()

  

consumer, err := client.Subscribe(pulsar.ConsumerOptions{

    // fill `Topic` field will create a single-topic consumer

    Topic:            "topic-1",

    SubscriptionName: "my-sub",

    Type:             pulsar.Shared,

})

if err != nil {

    log.Fatal(err)

}

defer consumer.Close()
```

--------------------------------

### Example Output for Get Schema (Admin CLI)

Source: https://pulsar.apache.org/docs/5.0.x/admin-api-schemas

This is an example of the JSON output when retrieving the latest schema for a topic using the Admin CLI.

```json
{
  "version": 0,
  "type": "String",
  "timestamp": 0,
  "data": "string",
  "properties": {
    "property1": "string",
    "property2": "string"
  }
}
```

--------------------------------

### Example Output: Monitor Brokers (Pulsar Connection with Extensions)

Source: https://pulsar.apache.org/docs/5.0.x/develop-load-manager

This example output shows broker metrics when using a Pulsar connection string with extensions enabled. It highlights system and message throughput.

```bash
===================================================================================================================
||SYSTEM         |CPU %          |MEMORY %       |DIRECT %       |BW IN %        |BW OUT %       |MAX %          ||
||               |17.24          |12.40          |26.95          |0.00           |0.00           |26.95          ||
||COUNT          |TOPIC          |BUNDLE         |PRODUCER       |CONSUMER       |BUNDLE +       |BUNDLE -       ||
||               |               |4              |               |               |               |               ||
||LATEST         |MSG/S IN       |MSG/S OUT      |TOTAL          |KB/S IN        |KB/S OUT       |TOTAL          ||
||               |100.02         |0.02           |100.03         |103.89         |0.01           |103.90         ||
===================================================================================================================
```

--------------------------------

### Start Pulsar Cluster with Docker Compose

Source: https://pulsar.apache.org/docs/5.0.x/concepts-broker-load-balancing-quick-start

Execute this command to start a Pulsar cluster using Docker Compose. Ensure Docker is installed and configured.

```bash
docker compose up -d
```

--------------------------------

### Transaction Buffer Stats Example Response

Source: https://pulsar.apache.org/docs/5.0.x/admin-api-transactions

Example JSON response for transaction buffer statistics, indicating the start position and whether the transaction has been aborted.

```json
{
  "startPosition" : "137:49759",
  "aborted" : false
}

```

--------------------------------

### Create a Python Consumer

Source: https://pulsar.apache.org/docs/client-libraries/python-use

This example demonstrates how to create a consumer, subscribe to a topic with a subscription name, receive messages, print their data and ID, and acknowledge them. It includes basic error handling for message processing.

```python
import pulsar  

client = pulsar.Client('pulsar://localhost:6650')  

consumer = client.subscribe('my-topic', 'my-subscription')  

while True:  
    msg = consumer.receive()  
    try:  
        print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))  
        # Acknowledge successful processing of the message  
        consumer.acknowledge(msg)  
    except Exception:  
        # Message failed to be processed  
        consumer.negative_acknowledge(msg)  

client.close()  
```

--------------------------------

### Get a Specific Schema (REST API Example Response)

Source: https://pulsar.apache.org/docs/5.0.x/admin-api-schemas

Example JSON response when retrieving a specific schema version via the REST API.

```json
{
  "version": "<the-version-number-of-the-schema>",
  "type": "<the-schema-type>",
  "timestamp": "<the-creation-timestamp-of-the-version-of-the-schema>",
  "data": "<an-utf8-encoded-string-of-schema-definition-data>",
  "properties": {}
}
```

--------------------------------

### Install Protobuf Tools

Source: https://pulsar.apache.org/docs/5.0.x/functions-package-python

Install protobuf tools to generate necessary proto files for Python function development. Pinning to a specific minor version is recommended.

```bash
pip install 'protobuf==3.20.*'
```

--------------------------------

### Create a New Tenant (Pulsar Admin CLI)

Source: https://pulsar.apache.org/docs/5.0.x/admin-api-tenants

Use the `create` subcommand to provision a new tenant. Basic creation requires only the tenant name.

```bash
pulsar-admin tenants create my-tenant
```

--------------------------------

### Example: Configure Region-Aware Placement Policy

Source: https://pulsar.apache.org/docs/5.0.x/administration-isolation-bookie

Configure bookie instances with their region/rack properties. This example shows how to set up bookies across different regions and racks.

```bash
bin/pulsar-admin bookies set-bookie-rack --bookie bookie1:3181 --hostname bookie1.pulsar.com:3181 --group group1 --rack RegionA/rack1  
bin/pulsar-admin bookies set-bookie-rack --bookie bookie2:3181 --hostname bookie2.pulsar.com:3181 --group group1 --rack RegionA/rack1  
bin/pulsar-admin bookies set-bookie-rack --bookie bookie3:3181 --hostname bookie3.pulsar.com:3181 --group group1 --rack RegionA/rack1  
bin/pulsar-admin bookies set-bookie-rack --bookie bookie4:3181 --hostname bookie4.pulsar.com:3181 --group group1 --rack RegionA/rack1  
bin/pulsar-admin bookies set-bookie-rack --bookie bookie5:3181 --hostname bookie5.pulsar.com:3181 --group group1 --rack RegionA/rack2  
bin/pulsar-admin bookies set-bookie-rack --bookie bookie6:3181 --hostname bookie6.pulsar.com:3181 --group group1 --rack RegionA/rack2  
bin/pulsar-admin bookies set-bookie-rack --bookie bookie7:3181 --hostname bookie7.pulsar.com:3181 --group group1 --rack RegionA/rack2  
bin/pulsar-admin bookies set-bookie-rack --bookie bookie8:3181 --hostname bookie8.pulsar.com:3181 --group group1 --rack RegionB/rack3  
...
```

--------------------------------

### Install Pulsar with Default Configuration

Source: https://pulsar.apache.org/docs/5.0.x/helm-install

Install Apache Pulsar using the Helm chart with default settings. This is a quick way to get a basic Pulsar cluster running.

```bash
helm install my-pulsar apache-pulsar/pulsar
```

--------------------------------

### Example krb5.conf File

Source: https://pulsar.apache.org/docs/5.0.x/security-kerberos

An example krb5.conf file showing the default realm and KDC server information. This configuration is essential for Kerberos to function correctly.

```ini
[libdefaults]
 default_realm = EXAMPLE.COM

[realms]
 EXAMPLE.COM  = {
  kdc = localhost:62037
 }
```

--------------------------------

### Full Example: ExclamationFunction with Localrun

Source: https://pulsar.apache.org/docs/5.0.x/functions-debug-localrun

This complete Java example defines an ExclamationFunction and includes a main method to run it in localrun mode. It configures the function with specific inputs, outputs, and class name, then starts the local runner.

```java
public class ExclamationFunction implements Function<String, String> {

    @Override
    public String process(String s, Context context) throws Exception {
        return s + "!";
    }

    public static void main(String[] args) throws Exception {
        FunctionConfig functionConfig = new FunctionConfig();
        functionConfig.setName("exclamation");
        functionConfig.setInputs(Collections.singleton("input"));
        functionConfig.setClassName(ExclamationFunction.class.getName());
        functionConfig.setRuntime(FunctionConfig.Runtime.JAVA);
        functionConfig.setOutput("output");

        LocalRunner localRunner = LocalRunner.builder().functionConfig(functionConfig).build();
        localRunner.start(false);
    }

}
```

--------------------------------

### Start PostgreSQL Server

Source: https://pulsar.apache.org/docs/5.0.x/io-debezium-source

Launches a PostgreSQL 13.3 instance in a Docker container with logical replication enabled.

```bash
docker run -d -it --rm \
    --name pulsar-postgres \
    -p 5432:5432 \
    -e POSTGRES_PASSWORD=changeme \
    postgres:13.3 -c wal_level=logical
```

--------------------------------

### Get Transaction Stats in Transaction Buffer

Source: https://pulsar.apache.org/docs/5.0.x/admin-api-transactions

Retrieves the start position and abortion status of a transaction within the transaction buffer.

```APIDOC
## GET /admin/v3/transactions/transactionInBufferStats/{tenant}/{namespace}/{topic}/{mostSigBits}/{leastSigBits}

### Description
Retrieves the start position and abortion status of a transaction within the transaction buffer.

### Method
GET

### Endpoint
/admin/v3/transactions/transactionInBufferStats/{tenant}/{namespace}/{topic}/{mostSigBits}/{leastSigBits}

### Parameters
#### Path Parameters
- **tenant** (string) - Required - The tenant of the topic.
- **namespace** (string) - Required - The namespace of the topic.
- **topic** (string) - Required - The topic name.
- **mostSigBits** (long) - Required - The most significant bits of the transaction ID.
- **leastSigBits** (long) - Required - The least significant bits of the transaction ID.

### Response
#### Success Response (200)
- **startPosition** (string) - The start position of this transaction in the transaction buffer.
- **aborted** (boolean) - The flag of whether this transaction has been aborted.

#### Response Example
```json
{
  "startPosition" : "137:49759",
  "aborted" : false
}
```
```

--------------------------------

### Start a source instance

Source: https://pulsar.apache.org/docs/io-cli

Initiate the execution of a Pulsar IO source connector instance. This command can start all instances or a specific one via `--instance-id`.

```bash
pulsar-admin sources start options  
```

--------------------------------

### InfluxDBv1 Sink Configuration (YAML)

Source: https://pulsar.apache.org/docs/5.0.x/io-influxdb-sink

Example YAML configuration for the InfluxDBv1 sink connector. This format can be used as an alternative to JSON for connector setup.

```yaml
configs:
    influxdbUrl: "http://localhost:8086"
    database: "test_db"
    consistencyLevel: "ONE"
    logLevel: "NONE"
    retentionPolicy: "autogen"
    gzipEnable: false
    batchTimeMs: 1000
    batchSize: 100

```

--------------------------------

### Java Consumer for Non-persistent Topic

Source: https://pulsar.apache.org/docs/5.0.x/concepts-messaging

Example of a Java consumer connecting to a non-persistent Pulsar topic. Ensure the topic name starts with 'non-persistent://'.

```java
PulsarClient client = PulsarClient.builder()  
        .serviceUrl("pulsar://localhost:6650")  
        .build();  
String npTopic = "non-persistent://public/default/my-topic";  
String subscriptionName = "my-subscription-name";  
  
Consumer<byte[]> consumer = client.newConsumer()  
        .topic(npTopic)  
        .subscriptionName(subscriptionName)  
        .subscribe();  

```

--------------------------------

### Write Message to File

Source: https://pulsar.apache.org/docs/io-file-source

Command to write a sample message 'hello world!' to a file named 'test.txt' in the /opt directory.

```bash
echo "hello world!" > /opt/test.txt

```

--------------------------------

### Create Producer and Send Message with Key to Compacted Topic (Java)

Source: https://pulsar.apache.org/docs/5.0.x/cookbooks-compaction

This example demonstrates creating a Pulsar client, a producer for a compacted topic, and sending a message with a key.

```java
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;

PulsarClient client = PulsarClient.builder()
        .serviceUrl("pulsar://localhost:6650")
        .build();

Producer<byte[]> compactedTopicProducer = client.newProducer()
        .topic("some-compacted-topic")
        .create();

compactedTopicProducer.newMessage()
        .key("some-key")
        .value(someByteArray)
        .send();
```

--------------------------------

### Get Entry Filters Policies

Source: https://pulsar.apache.org/docs/5.0.x/admin-api-namespaces

Retrieves the configured entry filter policies for a given namespace. This allows inspection of the current server-side message filtering setup.

```bash
pulsar-admin namespaces get-entry-filters test-tenant/namespace1
```

```http
GET /admin/v2/namespaces/{tenant}/{namespace}/entryFilters
```

```java
admin.namespaces().getEntryFilters(namespace);
```

--------------------------------

### ClickHouse JDBC Sink Configuration (JSON)

Source: https://pulsar.apache.org/docs/5.0.x/io-jdbc-sink

Example JSON configuration for the ClickHouse JDBC sink connector. Ensure the 'useTransactions' property is correctly set for your ClickHouse setup.

```json
{
  "configs": {
    "userName": "clickhouse",
    "password": "password",
    "jdbcUrl": "jdbc:clickhouse://localhost:8123/pulsar_clickhouse_jdbc_sink",
    "tableName": "pulsar_clickhouse_jdbc_sink"
    "useTransactions": "false"
  }
}
```

--------------------------------

### List Namespaces to Verify Creation

Source: https://pulsar.apache.org/docs/5.0.x/tutorials-namespace

After creating a namespace, use this command to list all namespaces within a tenant to confirm successful creation. This helps verify the namespace `apache/pulsar` exists.

```bash
bin/pulsar-admin namespaces list apache
```

--------------------------------

### Enable Generic Authentication Providers

Source: https://pulsar.apache.org/docs/5.0.x/functions-worker-run-separately

Enable authentication providers on function workers by specifying the desired providers in the 'authenticationProviders' list. This example shows a generic setup for multiple providers.

```yaml
authenticationEnabled: true

authenticationProviders: [provider1, provider2]
```

--------------------------------

### Start Simulation Client

Source: https://pulsar.apache.org/docs/5.0.x/develop-tools

Use the `pulsar-perf` script with the `simulation-client` command to start a simulation client. The client listens on a specified port and connects to the Pulsar service URL.

```bash
pulsar-perf simulation-client --port <listen port> --service-url <pulsar service url>
```

--------------------------------

### Start a source instance

Source: https://pulsar.apache.org/docs/5.0.x/io-cli

Start a specific Pulsar IO source instance or all instances if no instance ID is provided. Requires specifying the name, namespace, and tenant.

```bash
pulsar-admin sources start --name <source-name> --namespace <namespace> --tenant <tenant> [--instance-id <instance-id>]
```

--------------------------------

### Pulsar Go Client with Prometheus Metrics

Source: https://pulsar.apache.org/docs/client-libraries/go-use

This example shows how to set up a Pulsar Go client, producer, and expose Prometheus metrics via an HTTP server. It also includes a basic web server to trigger message production.

```go
// Create a Pulsar client
client, err := pulsar.NewClient(pulsar.ClientOptions{

    URL: "pulsar://localhost:6650",

})

if err != nil {

    log.Fatal(err)

}

defer client.Close()

// Start a separate goroutine for Prometheus metrics
// In this case, Prometheus metrics can be accessed via http://localhost:2112/metrics
go func() {

    prometheusPort := 2112

    log.Printf("Starting Prometheus metrics at http://localhost:%v/metrics\n", prometheusPort)

    http.Handle("/metrics", promhttp.Handler())

    err = http.ListenAndServe(":"+strconv.Itoa(prometheusPort), nil)

    if err != nil {

        log.Fatal(err)

    }

}()

// Create a producer
producer, err := client.CreateProducer(pulsar.ProducerOptions{

    Topic: "topic-1",

})

if err != nil {

    log.Fatal(err)

}

defer producer.Close()

ctx := context.Background()

// Write your business logic here
// In this case, you build a simple Web server. You can produce messages by requesting http://localhost:8082/produce
webPort := 8082
http.HandleFunc("/produce", func(w http.ResponseWriter, r *http.Request) {

    msgId, err := producer.Send(ctx, &pulsar.ProducerMessage{

        Payload: []byte(fmt.Sprintf("hello world")),

    })

    if err != nil {

        log.Fatal(err)

    } else {

        log.Printf("Published message: %v", msgId)

        fmt.Fprintf(w, "Published message: %v", msgId)

    }

})

err = http.ListenAndServe(":"+strconv.Itoa(webPort), nil)

if err != nil {

    log.Fatal(err)

}
```

--------------------------------

### Write Python Function with External Dependencies

Source: https://pulsar.apache.org/docs/5.0.x/functions-package-python

Example of a Python function that can utilize external libraries like 'js2xml'. The system will install dependencies if the function is a .whl file and 'installUserCodeDependencies' is enabled.

```python
from pulsar import Function
import js2xml

# The classic ExclamationFunction that appends an exclamation at the end
# of the input
class ExclamationFunction(Function):
    def __init__(self):
        pass

    def process(self, input, context):
        # add your logic
        return input + '!'
```

--------------------------------

### Set filesystem offloader threshold

Source: https://pulsar.apache.org/docs/5.0.x/tiered-storage-filesystem

This example sets the filesystem offloader threshold to 10 MB for a specific namespace using the pulsar-admin CLI tool. Ensure you have the pulsar-admin tool installed and configured.

```bash
pulsar-admin namespaces set-offload-threshold --size 10M my-tenant/my-namespace  
```

--------------------------------

### Create Producers and Consumers for Topics

Source: https://pulsar.apache.org/docs/txn-use

Sets up producers and consumers for input and output topics. This involves defining topic names and subscription names for consumers.

```go
// Step 4: create three producers to produce messages to input and output topics.  

inputTopic := "inputTopic"

outputTopicOne := "outputTopicOne"

outputTopicTwo := "outputTopicTwo"

subscriptionName := "your-subscription-name"

inputProducer, _ := client.CreateProducer(pulsar.ProducerOptions{
  Topic:       inputTopic,
  SendTimeout: 0,
})
defer inputProducer.Close()

outputProducerOne, _ := client.CreateProducer(pulsar.ProducerOptions{
  Topic:       outputTopicOne,
  SendTimeout: 0,
})
defer outputProducerOne.Close()

outputProducerTwo, _ := client.CreateProducer(pulsar.ProducerOptions{
  Topic:       outputTopicTwo,
  SendTimeout: 0,
})
defer outputProducerTwo.Close()

  

// Step 4: create three consumers to consume messages from input and output topics.  

inputConsumer, _ := client.Subscribe(pulsar.ConsumerOptions{
  Topic:            inputTopic,
  SubscriptionName: subscriptionName,
})
defer inputConsumer.Close()

outputConsumerOne, _ := client.Subscribe(pulsar.ConsumerOptions{
  Topic:            outputTopicOne,
  SubscriptionName: subscriptionName,
})
defer outputConsumerOne.Close()

outputConsumerTwo, _ := client.Subscribe(pulsar.ConsumerOptions{
  Topic:            outputTopicTwo,
  SubscriptionName: subscriptionName,
})
defer outputConsumerTwo.Close()
```

--------------------------------

### RabbitMQ Source Connector Configuration (YAML)

Source: https://pulsar.apache.org/docs/5.0.x/io-rabbitmq-source

Example configuration for the RabbitMQ source connector using YAML format. This setup specifies connection details and queue information for reading messages from RabbitMQ.

```yaml
configs:
    host: "localhost"
    port: 5672
    virtualHost: "/"
    username: "guest"
    password: "guest"
    queueName: "test-queue"
    connectionName: "test-connection"
    requestedChannelMax: 0
    requestedFrameMax: 0
    connectionTimeout: 60000
    handshakeTimeout: 10000
    requestedHeartbeat: 60
    prefetchCount: 0
    prefetchGlobal: "false"
    passive: "false"
```

--------------------------------

### Example Output: Monitor Brokers (ZooKeeper)

Source: https://pulsar.apache.org/docs/5.0.x/develop-load-manager

This is an example of the output generated when monitoring brokers via ZooKeeper. It includes system and message-related load statistics.

```bash
===================================================================================================================
||COUNT          |TOPIC          |BUNDLE         |PRODUCER       |CONSUMER       |BUNDLE +       |BUNDLE -       ||
||               |4              |4              |0              |2              |0              |0              ||
||RAW SYSTEM     |CPU %          |MEMORY %       |DIRECT %       |BW IN %        |BW OUT %       |MAX %          ||
||               |0.25           |47.94          |0.01           |0.00           |0.00           |47.94          ||
||ALLOC SYSTEM   |CPU %          |MEMORY %       |DIRECT %       |BW IN %        |BW OUT %       |MAX %          ||
||               |0.20           |1.89           |               |1.27           |3.21           |3.21           ||
||RAW MSG        |MSG/S IN       |MSG/S OUT      |TOTAL          |KB/S IN        |KB/S OUT       |TOTAL          ||
||               |0.00           |0.00           |0.00           |0.01           |0.01           |0.01           ||
||ALLOC MSG      |MSG/S IN       |MSG/S OUT      |TOTAL          |KB/S IN        |KB/S OUT       |TOTAL          ||
||               |54.84          |134.48         |189.31         |126.54         |320.96         |447.50         ||
===================================================================================================================
```

--------------------------------

### RabbitMQ Source Connector Configuration (JSON)

Source: https://pulsar.apache.org/docs/5.0.x/io-rabbitmq-source

Example configuration for the RabbitMQ source connector using JSON format. This setup specifies connection details and queue information for reading messages from RabbitMQ.

```json
{
  "configs": {
    "host": "localhost",
    "port": "5672",
    "virtualHost": "/",
    "username": "guest",
    "password": "guest",
    "queueName": "test-queue",
    "connectionName": "test-connection",
    "requestedChannelMax": "0",
    "requestedFrameMax": "0",
    "connectionTimeout": "60000",
    "handshakeTimeout": "10000",
    "requestedHeartbeat": "60",
    "prefetchCount": "0",
    "prefetchGlobal": "false",
    "passive": "false"
  }
}
```

--------------------------------

### Monitor Brokers with Default Configuration

Source: https://pulsar.apache.org/docs/develop-load-manager

Use this command to start a broker monitor and view detailed system and message statistics. The output is formatted for readability.

```bash
pulsar-perf monitor-brokers --connect-string <zookeeper host:port>
```

```bash
===================================================================================================================
||COUNT          |TOPIC          |BUNDLE         |PRODUCER       |CONSUMER       |BUNDLE +       |BUNDLE -       ||
||               |4              |4              |0              |2              |0              |0              ||
||RAW SYSTEM     |CPU %          |MEMORY %       |DIRECT %       |BW IN %        |BW OUT %       |MAX %          ||
||               |0.25           |47.94          |0.01           |0.00           |0.00           |47.94          ||
||ALLOC SYSTEM   |CPU %          |MEMORY %       |DIRECT %       |BW IN %        |BW OUT %       |MAX %          ||
||               |0.20           |1.89           |               |1.27           |3.21           |3.21           ||
||RAW MSG        |MSG/S IN       |MSG/S OUT      |TOTAL          |KB/S IN        |KB/S OUT       |TOTAL          ||
||               |0.00           |0.00           |0.00           |0.01           |0.01           |0.01           ||
||ALLOC MSG      |MSG/S IN       |MSG/S OUT      |TOTAL          |KB/S IN        |KB/S OUT       |TOTAL          ||
||               |54.84          |134.48         |189.31         |126.54         |320.96         |447.50         ||
===================================================================================================================
```

--------------------------------

### Create a Pulsar Reader

Source: https://pulsar.apache.org/docs/client-libraries/python-use

Use the Pulsar Python API to create a reader interface. This example demonstrates how to create a reader for a specific topic starting from a given message ID and continuously read messages.

```python
# MessageId taken from a previously fetched message  

msg_id = msg.message_id()  

reader = client.create_reader('my-topic', msg_id)  

while True:  
    msg = reader.read_next()  
    print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))  
    # No acknowledgment  
```

--------------------------------

### Get Namespace-Level Schema Compatibility Strategy (Java)

Source: https://pulsar.apache.org/docs/5.0.x/admin-api-schemas

Retrieve the schema compatibility strategy for a namespace using the Java client. The provided strategy in the example is for demonstration and might not reflect the actual retrieved value.

```java
admin.namespaces().getSchemaCompatibilityStrategy("test", SchemaCompatibilityStrategy.FULL);
```

--------------------------------

### Kinesis Sink Connector Configuration (YAML)

Source: https://pulsar.apache.org/docs/5.0.x/io-kinesis-sink

Example configuration for the Kinesis sink connector using YAML format. This setup specifies AWS endpoint, region, stream name, credentials, message format, and ordering retention.

```yaml
configs:
    awsEndpoint: "some.endpoint.aws"
    awsRegion: "us-east-1"
    awsKinesisStreamName: "my-stream"
    awsCredentialPluginParam: "{"accessKey":"myKey","secretKey":"my-Secret"}"
    messageFormat: "ONLY_RAW_PAYLOAD"
    retainOrdering: "true"
```

--------------------------------

### Kinesis Sink Connector Configuration (JSON)

Source: https://pulsar.apache.org/docs/5.0.x/io-kinesis-sink

Example configuration for the Kinesis sink connector using JSON format. This setup specifies AWS endpoint, region, stream name, credentials, message format, and ordering retention.

```json
{
  "configs": {
    "awsEndpoint": "some.endpoint.aws",
    "awsRegion": "us-east-1",
    "awsKinesisStreamName": "my-stream",
    "awsCredentialPluginParam": "{\"accessKey\":\"myKey\",\"secretKey\":\"my-Secret\"}",
    "messageFormat": "ONLY_RAW_PAYLOAD",
    "retainOrdering": "true"
  }
}
```

--------------------------------

### Start Pulsar standalone with custom configurations (Windows)

Source: https://pulsar.apache.org/docs/5.0.x/getting-started-docker

Run this command on Windows to start Pulsar with custom configurations applied via environment variables. It uses `apply-config-from-env.py` to load configurations before starting the standalone cluster, using '^' for line continuation.

```bash
docker run -it ^
-e PULSAR_PREFIX_xxx=yyy ^
-p 6650:6650  ^
-p 8080:8080 ^
--mount source=pulsardata,target=/pulsar/data ^
--mount source=pulsarconf,target=/pulsar/conf ^
apachepulsar/pulsar:5.0.0-M1 sh ^
-c "bin/apply-config-from-env.py ^
conf/standalone.conf && ^
bin/pulsar standalone --advertised-address localhost"  
```
Tokens:
Raw