Apache Pulsar

https://pulsar.apache.org/docs/4.1.x
Admin
Apache Pulsar is an open-source, distributed messaging and streaming platform built for the cloud.
Tokens:336,293
Snippets:3,918
Trust Score:10
Update:5 months ago
Show doc for...
Tokens:
Raw
### Start MongoDB Example Container

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

Pull and run the Debezium MongoDB example container. Ensure the necessary ports are exposed for local access.

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

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

### Example Schema Get Output

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

An example of the JSON output when retrieving a schema using the Admin CLI.

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

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

### Get Partitioned Topic Metadata Example Output

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

Example JSON output when retrieving metadata for a partitioned topic, showing the number of partitions and whether the topic is deleted.

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

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

### Package URL examples

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

Examples of package URLs for sinks, functions, and sources.

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

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

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

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

### Implement and Run ExclamationFunction

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

This example demonstrates a complete Java function implementing the `Function` interface and includes a `main` method to configure and run the function in localrun mode. The `start(false)` argument means the runner will not block.

```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 All Source Connectors

Source: https://pulsar.apache.org/docs/4.1.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
```

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

### MySQL Configuration File Example

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

This is an example of a MySQL configuration file (`mysqld.cnf`). It sets up essential parameters for the MySQL server.

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

### Method
POST

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

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

### Start Pulsar Manager

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

Execute the startup script to run Pulsar Manager. This command is used after completing the bare-metal installation steps.

```bash
./bin/pulsar-manager
```

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

### Configure and Start LocalRunner

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

This code configures a FunctionConfig and uses it to build and start a LocalRunner instance for debugging. The `start(true)` argument indicates that the runner should block until the function completes.

```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 Pulsar Simulation Controller

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

Use this command to start the simulation controller. Ensure clients are already started before starting the controller. This provides a shell interface 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>
```

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

### List Source Connectors Response Example

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

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

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

```

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

### Function Output Example

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

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

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

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

### Start Pulsar Simulation Client

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

Use this command to start a simulation client. The client will then be ready to receive commands from the simulation controller.

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

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

### Start All Sink Connectors

Source: https://pulsar.apache.org/docs/4.1.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
```

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

### Start Specified Sink Connector

Source: https://pulsar.apache.org/docs/4.1.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.

### Method
POST

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

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

### Quick Install Pulsar Manager with Docker

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

Use this command to quickly start Pulsar Manager in a Docker container for basic usage. It maps the necessary ports and sets the default configuration file.

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

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

### Install Pulsar Python Client

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

Install the pulsar-client library 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 Pulsar Cluster using Helm

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

Install the Apache Pulsar cluster on Kubernetes using the Helm chart. Specify the values file and set `initialize=true` for the initial installation.

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

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

### Install Pulsar Python Client

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

Install the Pulsar Python module using pip before using the client.

```bash
pip install pulsar-client
```

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

### Download and Start Pulsar Standalone

Source: https://pulsar.apache.org/docs/4.1.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.1.3/apache-pulsar-4.1.3-bin.tar.gz?action=download"

tar xvfz apache-pulsar-4.1.3-bin.tar.gz

cd apache-pulsar-4.1.3

bin/pulsar standalone
```

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

### Install Ansible

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

Install the Ansible automation tool on Linux or macOS using pip.

```bash
pip install ansible
```

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

### Start Pulsar Standalone Mode

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

Execute this command to start the Pulsar standalone instance.

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

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

### Start Pulsar standalone with custom configurations on Windows

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

Run this command on Windows to start Pulsar in Docker 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:4.1.3 sh ^
-c "bin/apply-config-from-env.py ^
conf/standalone.conf && ^
bin/pulsar standalone"  
```

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

### Consumer Output Example

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

Example of the message received by the consumer terminal after the File source connector processes the file.

```text
----- got message -----
hello world!
```

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

### Install Pulsar Helm Chart

Source: https://pulsar.apache.org/docs/4.1.x/getting-started-helm

Install the Apache Pulsar cluster using the Helm chart, specifying values file and namespace.

```bash
helm install \
    --values examples/values-minikube.yaml \
    --namespace pulsar \
    pulsar-mini apache/pulsar
```

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

### Start Pulsar Proxy

Source: https://pulsar.apache.org/docs/4.1.x/concepts-architecture-overview

Use this command to start a Pulsar proxy instance. Provide connection strings for the metadata store clusters.

```bash
cd /path/to/pulsar/directory  

bin/pulsar proxy \ 
    --metadata-store zk:my-zk-1:2181,my-zk-2:2181,my-zk-3:2181 \ 
    --configuration-metadata-store zk:my-zk-1:2181,my-zk-2:2181,my-zk-3:2181  
```

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

### Start PostgreSQL Docker Container

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

Starts a PostgreSQL 12 instance in a Docker container, mapping port 5432 and setting user credentials.

```bash
docker run -d -it --rm \
    --name pulsar-postgres \
    -p 5432:5432 \
    -e POSTGRES_PASSWORD=password \
    -e POSTGRES_USER=postgres \
    postgres:12
```

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

### REST API GET Endpoint Response Example

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

An example of the JSON response format when retrieving a schema 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.1.x/io-canal-source

Pull the Canal server image and run it, 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
```

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

### Start MySQL Server

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

Launches a MySQL server instance using Docker 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  
```

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

### Start Cassandra Docker Container

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

Starts a single-node Cassandra cluster using a Docker image. Ensure the Cassandra cluster is running before proceeding.

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

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

### Java Consumer for Non-persistent Topic

Source: https://pulsar.apache.org/docs/4.1.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();  

```

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

### Create Tenant and Namespace

Source: https://pulsar.apache.org/docs/4.1.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 Replication Clusters for Namespace

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

Retrieve the list of replication clusters configured for a given namespace. This helps in understanding the replication setup.

```bash
pulsar-admin namespaces get-clusters test-tenant/cluster1/namespace1
```

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

### Example krb5.conf File

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

Define the default Kerberos realm and Key Distribution Center (KDC) server information in the krb5.conf file.

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

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

```

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

### List Topics using pulsar-admin

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

Use this command to get a list of topics under a given namespace. Ensure you have the `pulsar-admin` tool installed and configured.

```bash
pulsar-admin topics list my-tenant/my-namespace
```

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

### List IBM Cloud CLI Plugins

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

Verify that the necessary IBM Cloud CLI plugins are installed. This command lists all installed plugins.

```bash
$ ibmcloud plugin list
```

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

### Initialize Terraform

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

Install all necessary Terraform dependencies. This command creates a `.terraform` folder in your current directory.

```bash
terraform init
```

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

### Get internal topic statistics

Source: https://pulsar.apache.org/docs/4.1.x/administration-geo

Validate that subscription replication is active by inspecting the internal statistics of a topic. This command helps in verifying the geo-replication setup.

```bash
bin/pulsar-admin topics stats-internal public/default/t1
```

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

### Start Broker Monitor

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

Use this command to start the broker monitor, which prints tabular load data to the console. It requires the ZooKeeper connection string.

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

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

### Get Namespace Retention Policy (JSON Output)

Source: https://pulsar.apache.org/docs/4.1.x/cookbooks-retention-expiry

Example JSON output for a namespace retention policy, showing retention time in minutes and size in megabytes.

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

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

### Python Language-Native Function Example

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

Define a `process` method for a language-native Python Pulsar Function. Ensure Python 3 is installed and set as the default interpreter for the functions worker.

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

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

### Get JDBC Sink Information

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

Retrieves detailed information about a specific JDBC sink connector, including its configuration, tenant, namespace, and topic. Use this to verify sink setup.

```bash
bin/pulsar-admin sinks get \
  --tenant public \
  --namespace default \
  --name pulsar-postgres-jdbc-sink
```

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

### Verify Tenant and Namespace Creation

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

List namespaces for the 'test' tenant to confirm successful creation of 'test/test-namespace'.

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

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

### Consume messages with Pulsar Python client

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

Example Python code to create a Pulsar client, subscribe to a topic, receive messages, acknowledge them, and close the client. Ensure the Pulsar client is installed and the cluster is running.

```python
import pulsar  

client = pulsar.Client('pulsar://localhost:6650')  
consumer = client.subscribe('my-topic', subscription_name='my-sub')  

while True:  
    msg = consumer.receive()  
    print("Received message: '%s'" % msg.data())
    consumer.acknowledge(msg)  

client.close()  
```

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

### Install HdrHistogram Plotter

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

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

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

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

### Start PostgreSQL Docker Image

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

Pulls and runs a PostgreSQL Docker image for Debezium to capture changes from. Ensure this is running before proceeding.

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

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

### Start all function instances

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

Start all stopped instances of a function.

```APIDOC
## Start all instances of a function

### Description
Start all stopped function instances.

### Method
POST

### Endpoint
/admin/v3/functions/{tenant}/{namespace}/{functionName}/start

### Parameters
#### Path Parameters
- **tenant** (string) - Required - The tenant of the function.
- **namespace** (string) - Required - The namespace of the function.
- **functionName** (string) - Required - The name of the function.

### Request Example (Java Admin API)
```java
admin.functions().startFunction(tenant, namespace, functionName);
```

### Request Example (Admin CLI)
```bash
pulsar-admin functions start \
    --tenant public \
    --namespace default \
    --name (the name of Pulsar Functions)
```
```

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

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

Source: https://pulsar.apache.org/docs/4.1.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.1.x/deploy-bare-metal

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

```bash
mkdir connectors
```

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

```bash
ls connectors
```

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

### Start PostgreSQL Server

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

Launches a PostgreSQL 13.3 server in a Docker container with logical replication enabled, required 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 a specific instance of a Pulsar Function using Admin CLI

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

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

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

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

### Start Minikube Cluster

Source: https://pulsar.apache.org/docs/4.1.x/getting-started-helm

Creates a Kubernetes cluster on Minikube with specified memory and CPU resources. Ensure your system meets the resource requirements.

```bash
minikube start --memory=8192 --cpus=4  

```

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

### Extract and Install Tiered Storage Offloaders

Source: https://pulsar.apache.org/docs/4.1.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 of the 'offloaders' directory.

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

mv apache-pulsar-offloaders-4.1.3/offloaders offloaders

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

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

### Initialize Transaction Coordinator Metadata

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

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

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

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

### Create Kerberos Principals for Brokers and Clients

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

Use `kadmin.local` to add principals for brokers and clients. This involves generating random keys and creating keytab files for secure communication.

```bash
### add Principals for broker  

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

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

### add Principals for client  

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

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

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

### Transaction Metadata Example

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

Example of the JSON response when retrieving transaction metadata.

```json
{
  "txnId" : "(1,18)",
  "status" : "ABORTING",
  "openTimestamp" : 1656592983374,
  "timeoutAt" : 5000,
  "producedPartitions" : {
    "my-topic" : {
      "startPosition" : "127:4959",
      "aborted" : true
    }
  },
  "ackedPartitions" : {
    "my-topic" : {
      "mysubName" : {
        "cumulativeAckPosition" : null
      }
    }
  }
}
```

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

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

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

Illustrates setting up a Pulsar client, producer, consumer, and reader with FileKeyReader for encryption and decryption in Go. Ensure key files are correctly specified.

```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 Pulsar Client with Transactions Enabled

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

Initialize a Pulsar client and enable transaction support. Ensure the service URL is correctly configured.

```go
client, err := pulsar.NewClient(pulsar.ClientOptions{
  URL:               "<serviceUrl>",
  EnableTransaction: true,
})
if err != nil {
  log.Fatalf("create client fail, err = %v", err)
}
deferr client.Close()
```

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

### Start Pulsar Standalone

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

Pull the Pulsar standalone image and run it, linking it to the Canal server and mounting a data volume.

```bash
docker pull apachepulsar/pulsar:4.1.3

docker run --user 0 -d -it --link pulsar-canal-server -p 6650:6650 -p 8080:8080 -v $PWD/data:/pulsar/data --name pulsar-standalone apachepulsar/pulsar:4.1.3 bin/pulsar standalone
```

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

### Start MySQL Server for Canal Connector

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

This command pulls the MySQL 5.7 Docker image and starts a container. It maps port 3306 and sets root and user credentials for the MySQL instance.

```bash
docker pull mysql:5.7
docker run -d -it --rm --name pulsar-mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=canal -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw mysql:5.7
```

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

### Start a function instance

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

Start a stopped function instance with a specific instance ID.

```APIDOC
## Start an instance of a function

### Description
Start a stopped function instance with a specific instance ID.

### Method
POST

### Endpoint
/admin/v3/functions/{tenant}/{namespace}/{functionName}/{instanceId}/start

### Parameters
#### Path Parameters
- **tenant** (string) - Required - The tenant of the function.
- **namespace** (string) - Required - The namespace of the function.
- **functionName** (string) - Required - The name of the function.
- **instanceId** (string) - Required - The ID of the instance to start.

### Request Example (Java Admin API)
```java
admin.functions().startFunction(tenant, namespace, functionName, Integer.parseInt(instanceId));
```

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

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

### Transaction Buffer Stats Example

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

Example of the JSON response for transaction buffer statistics.

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

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

### Create Producers and Consumers for Transactional Operations

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

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

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

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

### Build a Go Function

Source: https://pulsar.apache.org/docs/4.1.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 Go file.

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

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

### Start Pulsar Broker with Load Balancer Debug Mode

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

This configuration starts a Pulsar broker with debug mode enabled for the load balancer. 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"
```

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

### Create Tenant - Pulsar Admin

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

Use the `create` subcommand to provision a new tenant. Specify admin roles and allowed clusters using flags.

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

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

### Start Functions Worker in Foreground

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

Use the `pulsar-admin` CLI to start a function worker in the foreground.

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

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

### Start ZooKeeper Daemon

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

Use the `pulsar-daemon` CLI tool to start the ZooKeeper service in the background.

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

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

### Go Function Logging

Source: https://pulsar.apache.org/docs/4.1.x/functions-develop-log

Import the logutil package for logging in Go functions. This example demonstrates different log levels based on input length.

```go
import (
    "context"

    "github.com/apache/pulsar/pulsar-function-go/pf"

    log "github.com/apache/pulsar/pulsar-function-go/logutil"
)

func loggerFunc(ctx context.Context, input []byte) {
	if len(input) <= 100 {
		log.Infof("This input has a length of: %d", len(input))
	} else {
		log.Warnf("This input is getting too long! It has {%d} characters", len(input))
	}
}

func main() {
	pf.Start(loggerFunc)
}
```

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

### Transaction Coordinator Stats Example

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

Example of the JSON response when retrieving transaction coordinator statistics.

```json
{
  "state" : "Ready",
  "leastSigBits" : 1,
  "lowWaterMark" : 0,
  "ongoingTxnSize" : 0,
  "recoverStartTime" : 1657021892377,
  "recoverEndTime" : 1657021892378
}
```

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

### Start Pulsar standalone with custom configurations on macOS & Linux

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

Run this command to start Pulsar in Docker with custom configurations applied via environment variables. It uses `apply-config-from-env.py` to load configurations before starting the standalone cluster.

```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:4.1.3 sh \
-c "bin/apply-config-from-env.py \
conf/standalone.conf && \
bin/pulsar standalone"  
```

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

### Initialize MongoDB Data

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

Execute the provided script to set up the initial data within the MongoDB instance for Debezium to capture.

```bash
./usr/local/bin/init-inventory.sh
```

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

### Install htpasswd on CentOS/RHEL

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

Install the httpd-tools package, which includes the htpasswd utility, on CentOS or RHEL-based systems.

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

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

### Install htpasswd on Ubuntu/Debian

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

Install the apache2-utils package, which includes the htpasswd utility, on Ubuntu or Debian-based systems.

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

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

### Configure Go Client with Athenz Authentication

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

Set up the Go Pulsar client for Athenz authentication. Use `pulsar.NewAuthenticationAthenz` with a map of authentication parameters.

```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",
	"x509CertChain":  "file:///path/to/x509cert.pem",
	"privateKey":     "file:///path/to/private.pem",
	"caCert":         "file:///path/to/cacert.pem",
})

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


```

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

### Start Pulsar Standalone

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

Starts a Pulsar standalone instance. This command initiates all Pulsar service components.

```bash
bin/pulsar standalone  
```

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

### Create MySQL Test Database

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

Access the MySQL server and create a test database named 'test'.

```bash
docker exec -it pulsar-mysql /bin/bash

mysql -h 127.0.0.1 -uroot -pcanal -e 'create database test;'
```

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

### Start Functions Worker as a Daemon

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

Use the `pulsar-daemon` CLI tool to start a function worker in the background.

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

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

### Start MySQL Docker Container

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

Launches a MySQL instance in a Docker container, essential for Debezium to capture changes from. Ensure the correct image and ports are 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

```

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

### Java: Create Producers and Consumers

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

Set up producers and consumers for input and output topics. Configure send timeouts and subscription names as needed for your application.

```java
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();
```

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

### Start BookKeeper bookie in background

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

Use the pulsar-daemon CLI tool to start a BookKeeper bookie in the background.

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

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

### Transaction Pending Ack Stats Example

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

Example of the JSON response for transaction pending acknowledgment statistics.

```json
{
  "cumulativeAckPosition" : "137:49959"
}
```
Tokens:
Raw