### Start MongoDB Example Container

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

Pull and run a Debezium example MongoDB container for testing. Ensure the container is accessible.

```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
```

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

### Start MongoDB Example Container

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

Pull and run a Debezium example MongoDB container for testing. Ensure ports are mapped correctly.

```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 PostgreSQL Example Docker Image

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

Pulls and runs a Debezium example PostgreSQL Docker image. This is the first step to setting up a Debezium environment.

```bash
docker pull debezium/example-postgres:0.8
```

```bash
docker run -d -it --rm --name pulsar-postgresql -p 5432:5432  debezium/example-postgres:0.8
```

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

### Get Partitioned Topic Metadata Example Output

Source: https://pulsar.apache.org/docs/4.0.x/admin-api-topics

Example JSON output showing the metadata for a partitioned topic, including the partition count.

```json
{
  "partitions" : 4,
  "deleted" : false
}
```

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

### Function Get Command Output Example

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

The output of the `get` command provides a comprehensive view of the function's configuration, including input/output topics, processing guarantees, runtime, and parallelism.

```json
{
  "tenant": "public",
  "namespace": "default",
  "name": "ExclamationFunctio6",
  "className": "org.example.test.ExclamationFunction",
  "inputSpecs": {
    "persistent://public/default/my-topic-1": {
      "isRegexPattern": false
    }
  },
  "output": "persistent://public/default/test-1",
  "processingGuarantees": "ATLEAST_ONCE",
  "retainOrdering": false,
  "userConfig": {},
  "runtime": "JAVA",
  "autoAck": true,
  "parallelism": 1
}
```

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

### Start All Source Connectors

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

Starts all source connectors in the specified tenant and namespace.

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

### Description
Starts all source connectors.

### Method
POST

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

### Parameters
(No specific parameters documented for this operation)

### Response
#### Success Response (200)
- (No specific response fields documented)
```

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

### Start Pulsar Configuration Store Daemon

Source: https://pulsar.apache.org/docs/4.0.x/administration-zk-bk

Start the Pulsar configuration store service using the pulsar-daemon command after configuring the ZooKeeper setup.

```bash
bin/pulsar-daemon start configuration-store  


```

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

### Get Tenant Configuration

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

Use the `get` subcommand to fetch the configuration for an existing tenant.

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

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

### MySQL Configuration File Example

Source: https://pulsar.apache.org/docs/4.0.x/io-canal-source

An example of a MySQL configuration file (`mysqld.cnf`). This snippet shows basic settings and commented-out options.

```ini
[mysqld]
pid-file    = /var/run/mysqld/mysqld.pid
socket      = /var/run/mysqld/mysqld.sock
datadir     = /var/lib/mysql
#log-error  = /var/log/mysql/error.log
# By default we only accept connections from localhost
#bind-address   = 127.0.0.1
```

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

### Start Specified Source Connector

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

Starts a specified source connector instance.

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

### Description
Starts a specified source connector instance.

### Method
POST

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

### Parameters
#### Path Parameters
- **instanceId** (string) - Required - The ID of the source connector instance to start

### Response
#### Success Response (200)
- (No specific response fields documented)
```

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

### Start Pulsar Source Connector CLI

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

Use the `start` subcommand with `pulsar-admin sources` to start a source connector. Refer to Pulsar admin docs for complete information.

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

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

### Install Python Client and Protobuf Tools

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

Install the pulsar-client and protobuf tools. Ensure you use a compatible version of protobuf (e.g., 3.20.*).

```bash
pip install pulsar-client==2.10.0

```

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

```

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

### Install htpasswd on Ubuntu/Debian

Source: https://pulsar.apache.org/docs/4.0.x/security-basic-auth

Installs the necessary utility to create password files for basic authentication.

```bash
apt install apache2-utils
```

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

### Example Package URLs

Source: https://pulsar.apache.org/docs/4.0.x/admin-api-packages

These are examples of package URLs used to reference functions, sinks, and sources within the package manager.

```text
sink://public/default/mysql-sink@1.0
```

```text
function://my-tenant/my-ns/my-function@0.1
```

```text
source://my-tenant/my-ns/mysql-cdc-source@2.3
```

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

### Admin CLI Get Schema Output Example

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

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 Simulation Client

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

Use the `pulsar-perf` script with the `simulation-client` command to start a simulation client. The client will then be ready to receive controller commands.

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

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

### List Source Connectors Response Example

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

An example of the response when listing source connectors, showing a JSON array of connector names.

```json
["f1", "f2", "f3"]  
```

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

### Start Pulsar Sink Connector CLI

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

Use the `start` subcommand with `pulsar-admin sinks` to start a sink connector. Refer to Pulsar admin docs for complete information.

```bash
pulsar-admin sinks start options
```

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

### Start Simulation Controller

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

Use the `pulsar-perf` script with the `simulation-controller` command to start the simulation controller. Ensure clients are started before the controller. This provides a prompt to issue commands to simulation clients.

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

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

### Start Cassandra Docker Container

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

Starts a Cassandra Docker container. Ensure the container is running before proceeding.

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

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

### Download and Start Pulsar Standalone

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

Download the Pulsar package and start a standalone Pulsar instance. Ensure Pulsar is running before proceeding.

```bash
curl -LO "https://www.apache.org/dyn/closer.lua/pulsar/pulsar-4.0.10/apache-pulsar-4.0.10-bin.tar.gz?action=download"
tar xvfz apache-pulsar-4.0.10-bin.tar.gz
cd apache-pulsar-4.0.10
bin/pulsar standalone
```

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

### Start All Sink Connectors

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

Starts all sink connectors in the specified tenant and namespace.

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

### Description
Starts all sink connectors.

### Method
POST

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

### Parameters
(No specific parameters documented for this operation)

### Response
#### Success Response (200)
- (No specific response fields documented)
```

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

### Start Specified Sink Connector

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

Starts a specified sink connector instance.

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

### Description
Starts a specified sink connector instance.

### Method
POST

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

### Parameters
#### Path Parameters
- **instanceId** (string) - Required - The ID of the sink connector instance to start

### Response
#### Success Response (200)
- (No specific response fields documented)
```

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

### Consumer Output Example

Source: https://pulsar.apache.org/docs/4.0.x/functions-deploy-trigger

Example of the output received by a consumer listening on the function's output topic after the function has been triggered.

```text
----- got message -----
This function has been triggered with a value of hello world
```

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

### Install htpasswd on CentOS/RHEL

Source: https://pulsar.apache.org/docs/4.0.x/security-basic-auth

Installs the necessary utility to create password files for basic authentication.

```bash
yum install httpd-tools
```

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

### Start ScyllaDB Docker Container

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

Starts a ScyllaDB Docker container with specific resource configurations. Note the required flags for containerized ScyllaDB.

```bash
docker run -d --rm --name=scylladb -p 9042:9042 \
    scylladb/scylla:latest \
    --smp 1 --memory 750M --overprovisioned 1
```

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

### Install cert-manager using script

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

Installs the cert-manager release to the 'cert-manager' namespace using a provided bash script. Ensure you clone the repository first.

```bash
git clone https://github.com/apache/pulsar-helm-chart  

cd pulsar-helm-chart  

./scripts/cert-manager/install-cert-manager.sh  
```

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

### Start Pulsar Cluster with Docker Compose

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

Execute this command in your terminal to start the Pulsar cluster. Ensure Docker is installed and running.

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

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

### REST API Get Schema Response Example

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

An example of the JSON response when retrieving schema information 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": {}
}
```

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

### Start Canal Server

Source: https://pulsar.apache.org/docs/4.0.x/io-canal-source

Pull the Canal server image and run it as a Docker container, linking it to the MySQL server and configuring its connection details.

```bash
docker pull canal/canal-server:v1.1.2

docker run -d -it --link pulsar-mysql -e canal.auto.scan=false -e canal.destinations=test -e canal.instance.master.address=pulsar-mysql:3306 -e canal.instance.dbUsername=root -e canal.instance.dbPassword=canal -e canal.instance.connectionCharset=UTF-8 -e canal.instance.tsdb.enable=true -e canal.instance.gtidon=false --name=pulsar-canal-server -p 8000:8000 -p 2222:2222 -p 11111:11111 -p 11112:11112 -m 4096m canal/canal-server:v1.1.2
```

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

### Transaction Buffer Stats Example

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

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

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

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

### Create a shared subscription consumer in Go

Source: https://pulsar.apache.org/docs/4.0.x/cookbooks-message-queue

This Go example demonstrates setting up a Pulsar consumer for message queueing with a shared subscription. It includes client options, topic, subscription name, and receiver queue size configuration.

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

client, err := pulsar.NewClient(pulsar.ClientOptions{
    URL: "pulsar://localhost:6650",
})
if err != nil {
    log.Fatal(err)
}
consumer, err := client.Subscribe(pulsar.ConsumerOptions{
    Topic:             "persistent://public/default/mq-topic-1",
    SubscriptionName:  "sub-1",
    Type:              pulsar.Shared,
    ReceiverQueueSize: 10, // If you'd like to restrict the receiver queue size
})
if err != nil {
    log.Fatal(err)
}

```

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

### Initialize Terraform

Source: https://pulsar.apache.org/docs/4.0.x/deploy-aws

Install all necessary Terraform dependencies. This command creates a `.terraform` folder.

```bash
terraform init
```

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

### Start Pulsar Standalone Instance

Source: https://pulsar.apache.org/docs/4.0.x/io-netty-source

Starts a Pulsar standalone instance using Docker. Ensure you replace `{version}` with the desired Pulsar version.

```bash
docker pull apachepulsar/pulsar:{version}

docker run -d -it -p 6650:6650 -p 8080:8080 -v $PWD/data:/pulsar/data --name pulsar-netty-standalone apachepulsar/pulsar:{version} bin/pulsar standalone

```

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

### Go: Producer, Consumer, and Reader with FileKeyReader

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

Illustrates setting up a producer, consumer, and reader with FileKeyReader for encryption and decryption in Go.

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

defer client.Close()

topic := "persistent://my-tenant/my-ns/my-topic"
keyReader := crypto.NewFileKeyReader("test_ecdsa_pubkey.pem", "test_ecdsa_privkey.pem")
producer, err := client.CreateProducer(pulsar.ProducerOptions{
    Topic: topic,
    Encryption: &pulsar.ProducerEncryptionInfo{
    	KeyReader: keyReader,
    	Keys:      []string{"myappkey"},
    },
})
if err != nil {
	log.Fatal(err)
}
defer producer.Close()

consumer, err := client.Subscribe(pulsar.ConsumerOptions{
    Topic:            topic,
    SubscriptionName: "my-subscriber-name",
    Decryption: &pulsar.MessageDecryptionInfo{
 	   KeyReader: keyReader,
    },
})
if err != nil {
    log.Fatal(err)
}
defer consumer.Close()

reader, err := client.CreateReader(pulsar.ReaderOptions{
    Topic: topic,
    Decryption: &pulsar.MessageDecryptionInfo{
 	   KeyReader: keyReader,
    },
})
if err != nil {
    log.Fatal(err)
}
defer reader.Close()
```

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

### Create Tenant and Namespace

Source: https://pulsar.apache.org/docs/4.0.x/functions-quickstart

Create a new tenant named 'test' and a namespace 'test-namespace' within it for testing purposes.

```bash
bin/pulsar-admin tenants create test
bin/pulsar-admin namespaces create test/test-namespace
```

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

### Get info of leader broker

Source: https://pulsar.apache.org/docs/4.0.x/admin-api-brokers

Fetch the information of the leader broker, for example, the service URL.

```APIDOC
## Get info of leader broker

### Description
Fetch the information of the leader broker, for example, the service URL.

### Method
GET

### Endpoint
/admin/v2/brokers/leaderBroker

### Request Example
```
GET /admin/v2/brokers/leaderBroker
```

### Response
#### Success Response (200)
- **serviceUrl** (string) - The service URL of the leader broker.

### Response Example
```json
{
  "serviceUrl": "http://localhost:8080"
}
```
```

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

### Kinesis Sink JSON Configuration Example

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

Example configuration for the Kinesis sink connector using JSON format. Ensure all required properties are set according to your AWS environment and Kinesis stream setup.

```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"
   }
}

```

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

### Create a New Tenant

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

Use the `create` subcommand to create a new tenant. Optionally, specify admin roles and allowed clusters.

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

```bash
pulsar-admin tenants create my-tenant \
    --admin-roles role1,role2,role3 \
    --allowed-clusters cluster1
```

```bash
pulsar-admin tenants create my-tenant \
    -r role1 \
    -c cluster1
```

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

### Java Producer for Non-persistent Topic

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

Example of creating a Java producer for a non-persistent topic. The topic name must start with 'non-persistent://'.

```java
Producer<byte[]> producer = client.newProducer()  
                .topic(npTopic)  
                .create();  
```

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

### Example Source Connector Configuration (Generic)

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

A generic example of a SourceConfig JSON object, illustrating the structure and common fields like tenant, namespace, className, and resources.

```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
 }
}

```

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

### Java Consumer for Non-persistent Topic

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

Example of creating a Java consumer for a non-persistent 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();  
```

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

### Get Stateful Function Information

Source: https://pulsar.apache.org/docs/4.0.x/functions-quickstart

Retrieve detailed configuration and status information for a deployed function. This is useful for verifying the function's setup and parameters.

```bash
bin/pulsar-admin functions get \
   --tenant test \
   --namespace test-namespace \
   --name word_count
```

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

### Python Language-Native Function Example

Source: https://pulsar.apache.org/docs/4.0.x/functions-develop-api

For language-native Python functions, define a method named `process` that accepts the input string and returns the processed string. Ensure Python 3 is installed and set as the default interpreter.

```python
def process(input):
    return "{}!".format(input)
```

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

### ZooKeeper Server Configuration

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

Example ZooKeeper configuration for a multi-cluster setup, defining quorum members and observers across different regions. Ensure ZK quorum members are spread across at least 3 regions, and other regions run as observers.

```properties
clientPort=2184

server.1=zk1.us-west.example.com:2185:2186
server.2=zk2.us-west.example.com:2185:2186
server.3=zk3.us-west.example.com:2185:2186
server.4=zk1.us-central.example.com:2185:2186
server.5=zk2.us-central.example.com:2185:2186
server.6=zk3.us-central.example.com:2185:2186:observer
server.7=zk1.us-east.example.com:2185:2186
server.8=zk2.us-east.example.com:2185:2186
server.9=zk3.us-east.example.com:2185:2186:observer
server.10=zk1.eu-central.example.com:2185:2186:observer
server.11=zk2.eu-central.example.com:2185:2186:observer
server.12=zk3.eu-central.example.com:2185:2186:observer
server.13=zk1.ap-south.example.com:2185:2186:observer
server.14=zk2.ap-south.example.com:2185:2186:observer
server.15=zk3.ap-south.example.com:2185:2186:observer
```

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

### Install HdrHistogram Plotter

Source: https://pulsar.apache.org/docs/4.0.x/performance-pulsar-perf

Install the HdrHistogram Plotter using Maven. Tests are skipped during this installation process.

```bash
mvn clean install -DskipTests
```

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

### Start Pulsar Broker in Background

Source: https://pulsar.apache.org/docs/4.0.x/deploy-bare-metal

Start a Pulsar broker as a background daemon using the `pulsar-daemon start broker` command.

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

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

### Extract and Install Pulsar Offloaders

Source: https://pulsar.apache.org/docs/4.0.x/tiered-storage-overview

After downloading, untar the offloaders package and move the 'offloaders' directory into your Pulsar installation directory. Verify the installation by listing the contents.

```bash
tar xvfz apache-pulsar-offloaders-4.0.10-bin.tar.gz

mv apache-pulsar-offloaders-4.0.10/offloaders offloaders

ls offloaders
# tiered-storage-file-system-4.0.10.nar
# tiered-storage-jcloud-4.0.10.nar
```

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

### List Source Connectors Options

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

Use this command to see the available options for listing source connectors via the Admin CLI.

```bash
pulsar-admin sources list options  
```

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

### Start PostgreSQL Server

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

Launches a PostgreSQL server in a Docker container with logical replication enabled, necessary for Debezium.

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

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

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

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

Use the `start` subcommand with `pulsar-admin` without specifying an instance ID to start all stopped instances of a function.

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

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

### Install Pulsar IO Connector

Source: https://pulsar.apache.org/docs/4.0.x/deploy-bare-metal

Create a 'connectors' directory and move the downloaded .nar file into it. This makes the connector available for use with Pulsar.

```bash
mkdir connectors
```

```bash
mv pulsar-io-aerospike-4.0.10.nar connectors
```

```bash
ls connectors
```

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

### Install Ansible

Source: https://pulsar.apache.org/docs/4.0.x/deploy-aws

Install Ansible on Linux or macOS using pip.

```bash
pip install ansible
```

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

### Install Pulsar Cluster using Helm

Source: https://pulsar.apache.org/docs/4.0.x/deploy-ibm

Install the Apache Pulsar cluster on Kubernetes using the Helm chart. Ensure to set `--set initialize=true` for the initial installation.

```bash
helm install --values examples/values-minikube.yaml --set initialize=true asia apache/pulsar  
```

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

### Java: Pulsar Client and Transactional Operations

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

Example demonstrating how to create a Pulsar client with transactions enabled, set up producers and consumers, and perform transactional message production and acknowledgment. Ensure the client role has necessary permissions for the 'pulsar/system' namespace.

```java
PulsarClient client = PulsarClient.builder()  
                .enableTransaction(true)  
                .serviceUrl(jct.serviceUrl)  
                .build();  

ProducerBuilder<String> producerBuilder = client.newProducer(Schema.STRING);
Producer<String> inputProducer = producerBuilder.topic(inputTopic)  
        .sendTimeout(0, TimeUnit.SECONDS).create();  
Producer<String> outputProducerOne = producerBuilder.topic(outputTopicOne)  
        .sendTimeout(0, TimeUnit.SECONDS).create();  
Producer<String> outputProducerTwo = producerBuilder.topic(outputTopicTwo)  
        .sendTimeout(0, TimeUnit.SECONDS).create();  

Consumer<String> inputConsumer = client.newConsumer(Schema.STRING)  
        .subscriptionName("your-subscription-name").topic(inputTopic).subscribe();  
Consumer<String> outputConsumerOne = client.newConsumer(Schema.STRING)  
        .subscriptionName("your-subscription-name").topic(outputTopicOne).subscribe();  
Consumer<String> outputConsumerTwo = client.newConsumer(Schema.STRING)  
        .subscriptionName("your-subscription-name").topic(outputTopicTwo).subscribe();  

int count = 2;  

for (int i = 0; i < count; i++) {  
    inputProducer.send("Hello Pulsar! count : " + i);  
}

for (int i = 0; i < count; i++) {  

    Message<String> message = inputConsumer.receive();  

    Transaction txn = null;  
    try {  
        txn = client.newTransaction()  
                .withTransactionTimeout(10, TimeUnit.SECONDS).build().get();  

        outputProducerOne.newMessage(txn).value("Hello Pulsar! outputTopicOne count : " + i).send();  
        outputProducerTwo.newMessage(txn).value("Hello Pulsar! outputTopicTwo count : " + i).send();  

        inputConsumer.acknowledgeAsync(message.getMessageId(), txn).get();  
        txn.commit().get();  
    } catch (ExecutionException e) {  
        if (!(e.getCause() instanceof PulsarClientException.TransactionConflictException)) {  
            inputConsumer.negativeAcknowledge(message);  
        }

        if (txn != null) {  
            txn.abort();  
        }  
    }  
}

for (int i = 0; i < count; i++) {  
    Message<String> message =  outputConsumerOne.receive();  
```

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

### Start a specific instance of a Pulsar Function using Admin CLI

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

Use the `start` subcommand with `pulsar-admin` and specify the `--instance-id` to start a particular stopped function instance.

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

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

### List Namespaces for Public Tenant

Source: https://pulsar.apache.org/docs/4.0.x/functions-quickstart

Verify the existence of the 'public/default' namespace.

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

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

### Configure Athenz Authentication in Go Pulsar Client

Source: https://pulsar.apache.org/docs/4.0.x/security-athenz

This Go example shows how to configure Athenz authentication for a Pulsar client using a map of string key-value pairs.

```go
provider := pulsar.NewAuthenticationAthenz(map[string]string{  
	"ztsUrl":         "http://localhost:9998",  
	// "ztsProxyUrl":    "http://localhost:9999", // Proxy for accessing ZTS (optional, since v0.16.0)  
	"providerDomain": "pulsar",  
	"tenantDomain":   "shopping",  
	"tenantService":  "some_app",  
	"privateKey":     "file:///path/to/private.pem",  
	"keyId":          "v1",  
})

  
client, err := pulsar.NewClient(pulsar.ClientOptions{  
	URL:            "pulsar://my-broker.com:6650",  
	Authentication: provider,  
})


```

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

### Configure Basic Authentication in Go Pulsar Client

Source: https://pulsar.apache.org/docs/4.0.x/security-basic-auth

Provides an example of setting up a Go Pulsar client to use basic authentication, including error handling for authentication provider creation.

```go
provider, err := pulsar.NewAuthenticationBasic("admin", "123456")  
if err != nil {  
    log.Fatal(err)  
}
client, err := pulsar.NewClient(pulsar.ClientOptions{  
    URL: "pulsar://broker.example.com:6650",  
    Authentication: provider,  
})
```

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

### Start Netty Source Connector

Source: https://pulsar.apache.org/docs/4.0.x/io-netty-source

Starts the Netty source connector using the Pulsar admin CLI. Ensure the archive path and configuration file path are correct.

```bash
./bin/pulsar-admin sources localrun \
--archive $PWD/pulsar-io-4.0.10.nar \
--tenant public \
--namespace default \
--name netty \
--destination-topic-name netty-topic \
--source-config-file $PWD/netty-source-config.yaml \
--parallelism 1

```

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

### Example Function Configuration

Source: https://pulsar.apache.org/docs/4.0.x/functions-quickstart

This YAML file defines the configuration for the example Pulsar function, including tenant, namespace, input/output topics, and class name.

```yaml
tenant: "test"
namespace: "test-namespace"
name: "example" # function name
className: "org.apache.pulsar.functions.api.examples.ExclamationFunction"
inputs: ["test_src"] # this function will read messages from these topics
output: "test_result" # the return value of this function will be sent to this topic
autoAck: true # function will acknowledge input messages if set true
parallelism: 1
```

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

### Start Broker in Foreground

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

Start a Pulsar broker in the foreground for debugging or interactive use.

```bash
bin/pulsar broker  
```

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

### Start Pulsar Broker with Load Balancer Debug Mode

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

This configuration starts a Pulsar broker with load balancer debug mode enabled. Ensure ZooKeeper and BookKeeper are healthy and started before running the broker.

```yaml
      - PULSAR_PREFIX_loadBalancerDebugModeEnabled=true  

    depends_on:  
      zookeeper:  
        condition: service_healthy  
      bookie:  
        condition: service_started  
    command: bash -c "bin/apply-config-from-env.py conf/broker.conf && exec bin/pulsar broker"  
```

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

### Sticky Partitioning Example

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

Demonstrates how ranges are defined for consumers in Sticky partitioning. A message key's hash is used with a modulo operation to determine which consumer's range it falls into.

```plaintext
C1 = [0, 16384), [32768, 49152)  
C2 = [16384, 32768), [49152, 65536)  

 0               16,384            32,768           49,152             65,536  
 |------- C1 ------|------- C2 ------|------- C1 ------|------- C2 ------|
```

```plaintext
murmur32("Order-3459134") = 3112179635
3112179635 mod 65536 = 6067
```

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

### Build a Go Function

Source: https://pulsar.apache.org/docs/4.0.x/functions-package-go

Use this command to build your Go function file. Ensure you replace `<your Go function filename>.go` with the actual name of your file.

```go
go build <your Go function filename>.go
```

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

### Start Pulsar standalone

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

Starts a Pulsar standalone instance. This is a prerequisite for performing offload operations.

```bash
bin/pulsar standalone -a 127.0.0.1
```

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

### Broker Configuration Example

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

Configure Pulsar brokers by setting essential parameters like ZooKeeper quorum, cluster name, and service ports in the `conf/broker.conf` file.

```properties
# Local ZooKeeper servers  
metadataStoreUrl=zk1.us-west.example.com:2181,zk2.us-west.example.com:2181,zk3.us-west.example.com:2181  
  
# Configuration store quorum connection string.  
configurationMetadataStoreUrl=zk1.us-west.example.com:2184,zk2.us-west.example.com:2184,zk3.us-west.example.com:2184  
  
clusterName=us-west  
  
# Broker data port  
brokerServicePort=6650  
  
# Broker data port for TLS  
brokerServicePortTls=6651  
  
# Port to use to server HTTP request  
webServicePort=8080  
  
# Port to use to server HTTPS request  
webServicePortTls=8443  

```

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

### Start Broker Daemon

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

Start a Pulsar broker in the background using the `pulsar-daemon` CLI tool.

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

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

### Start Broker Monitor

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

Use the `monitor-brokers` command to start a broker monitor. This will continuously print load data to the console until interrupted. Ensure you provide the correct ZooKeeper connection string.

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

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

### Configure Basic Authentication in C++ Pulsar Client

Source: https://pulsar.apache.org/docs/4.0.x/security-basic-auth

Shows how to set up a C++ Pulsar client for basic authentication, providing username and password during client configuration.

```cpp
#include <pulsar/Client.h>  

int main() {  
    pulsar::ClientConfiguration config;  
    AuthenticationPtr auth = pulsar::AuthBasic::create("admin", "123456")  
    config.setAuth(auth);  
    pulsar::Client client("pulsar://broker.example.com:6650/", config);  

    return 0;  
}
```

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

### Start Functions Worker using pulsar-admin

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

Start a function worker in the foreground using the `pulsar-admin` CLI.

```bash
bin/pulsar functions-worker
```

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

### List Command Output Example

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

The `list` command returns the names of functions, typically displayed one per line, for the specified tenant and namespace.

```text
ExclamationFunctio1
ExclamationFunctio2
ExclamationFunctio3
```

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

### Example Sink Configuration

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

An example JSON object representing the configuration of a Pulsar IO sink connector.

```json
{
"tenant": "tenantName",
"namespace": "namespaceName",
"name": "sinkName",
"className": "className",
"inputSpecs": {
"topicName": {
    "isRegexPattern": false
}
},
"configs": {},
"parallelism": 1,
"processingGuarantees": "ATLEAST_ONCE",
"retainOrdering": false,
"autoAck": true
}

```

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

### Start Functions Worker using pulsar-daemon

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

Start a function worker in the background using the `pulsar-daemon` CLI tool.

```bash
bin/pulsar-daemon start functions-worker
```

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

### Python Pulsar Client Example

Source: https://pulsar.apache.org/docs/4.0.x/deploy-aws

Demonstrates basic Pulsar client operations in a Python shell, including connecting to the cluster, creating a producer, sending a message, and closing the client.

```python
>>> import pulsar
>>> client = pulsar.Client('pulsar://pulsar-elb-1800761694.us-west-2.elb.amazonaws.com:6650')
# Make sure to use your connection URL
>>> producer = client.create_producer('persistent://public/default/test-topic')
>>> producer.send('Hello world')
>>> client.close()
```

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

### Example Source Connector Configuration (Debezium MySQL)

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

A detailed example of a SourceConfig for a Debezium MySQL source connector, including specific database connection details, converters, and Pulsar service URLs.

```json
{
 "tenant": "public",
 "namespace": "default",
 "name": "debezium-mysql-source",
 "className": "org.apache.pulsar.io.debezium.mysql.DebeziumMysqlSource",
 "topicName": "debezium-mysql-topic",
 "configs": {
   "database.user": "debezium",
   "database.server.id": "184054",
   "database.server.name": "dbserver1",
   "database.port": "3306",
   "database.hostname": "localhost",
   "database.password": "dbz",
   "database.history.pulsar.service.url": "pulsar://127.0.0.1:6650",
   "value.converter": "org.apache.kafka.connect.json.JsonConverter",
   "database.whitelist": "inventory",
   "key.converter": "org.apache.kafka.connect.json.JsonConverter",
   "database.history": "org.apache.pulsar.io.debezium.PulsarDatabaseHistory",
   "pulsar.service.url": "pulsar://127.0.0.1:6650",
   "database.history.pulsar.topic": "history-topic2"
 },
 "parallelism": 1,
 "processingGuarantees": "ATLEAST_ONCE",
 "resources": {
   "cpu": 1.0,
   "ram": 1073741824,
   "disk": 10737418240
 }
}

```

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

### Configure and Start LocalRunner

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

Programmatically configure a FunctionConfig and initialize a LocalRunner to start your function in localrun mode for debugging.

```java
FunctionConfig functionConfig = new FunctionConfig();
functionConfig.setName(functionName);
functionConfig.setInputs(Collections.singleton(sourceTopic));
functionConfig.setClassName(ExclamationFunction.class.getName());
functionConfig.setRuntime(FunctionConfig.Runtime.JAVA);
functionConfig.setOutput(sinkTopic);

LocalRunner localRunner = LocalRunner.builder().functionConfig(functionConfig).build();
localRunner.start(true);
```

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

### Start ZooKeeper Daemon

Source: https://pulsar.apache.org/docs/4.0.x/administration-zk-bk

Start the ZooKeeper service in the background on all configured hosts using the `pulsar-daemon` CLI tool.

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

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

### Create Kerberos Principals for Brokers and Clients

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

Use kadmin.local to create principals and keytabs for Pulsar brokers and clients. Ensure hostnames are resolvable by FQDN.

```bash
sudo /usr/sbin/kadmin.local -q 'addprinc -randkey broker/{hostname}@{REALM}'
```

```bash
sudo /usr/sbin/kadmin.local -q "ktadd -k /etc/security/keytabs/{broker-keytabname}.keytab broker/{hostname}@{REALM}"
```

```bash
sudo /usr/sbin/kadmin.local -q 'addprinc -randkey client/{hostname}@{REALM}'
```

```bash
sudo /usr/sbin/kadmin.local -q "ktadd -k /etc/security/keytabs/{client-keytabname}.keytab client/{hostname}@{REALM}"
```

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

### Get Tenant Configuration (REST API)

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

Use the GET /admin/v2/tenants/{tenant} endpoint to retrieve tenant information.

```java
admin.tenants().getTenantInfo(tenantName);
```

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

### Initialize Transaction Coordinator Metadata

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

Run this command to initialize the necessary metadata for the transaction coordinator. This command requires the ZooKeeper connection string and cluster name.

```bash
bin/pulsar initialize-transaction-coordinator-metadata -cs 127.0.0.1:2181 -c standalone
```