Apache Pulsar

https://pulsar.apache.org/docs/4.0.x
Admin
Apache Pulsar is a multi-tenant, high-performance message broker with native multi-cluster support,...
Tokens:411,540
Snippets:3,408
Trust Score:10
Update:1 week ago
Show doc for...
Tokens:
Raw
### Get subscriptions example output

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

Example output of the subscriptions command showing the subscription name.

```text
my-subscription  
```

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

### Start MySQL server with Debezium example image

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

Starts a MySQL server with a database from which Debezium can capture changes. Uses the debezium/example-mysql:0.8 image with root password 'debezium' and user 'mysqluser'.

```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 MongoDB with Debezium example image

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

Pulls the Debezium MongoDB example image and starts a container named pulsar-mongodb with MongoDB credentials and port 27017 exposed. Use this before initializing inventory data.

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

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

### Example output of sinks get

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

Example output from the 'sinks get' command, showing the sink configuration and runtime details.

```json
{  

"tenant": "public",  

"namespace": "default",  

"name": "cassandra-test-sink",  

"className": "org.apache.pulsar.io.cassandra.CassandraStringSink",  

"inputSpecs": {  

  "test_cassandra": {  

    "isRegexPattern": false  

  }  

},  

"configs": {  

  "roots": "localhost:9042",  

  "keyspace": "pulsar_test_keyspace",  

  "columnFamily": "pulsar_test_table",  

  "keyname": "key",  

  "columnName": "col"  

},  

"parallelism": 1,  

"processingGuarantees": "ATLEAST_ONCE",  

"retainOrdering": false,  

"autoAck": true,  

"archive": "builtin://cassandra"  

}  


```

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

### Create a function named examples with pulsar-admin

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

Creates a Pulsar function named `examples` using a function config file and a JAR. The example function appends a `!` to every message. Requires Pulsar to be started and a test namespace to exist beforehand.

```bash
bin/pulsar-admin functions create \  

   --function-config-file $PWD/examples/example-function-config.yaml \  

   --jar $PWD/examples/api-examples.jar  
```

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

### Get info of leader broker - example output

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

Example output showing the leader broker's service URL.

```json
{

  "serviceUrl" : "http://localhost:8080"

}
```

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

### Start MongoDB server with Debezium example image

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

Pulls and runs the debezium/example-mongodb:0.10 Docker container with environment variables for MongoDB user and password, exposing port 27017.

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

```

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

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

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
- **tenant** (string) - Required - The tenant of the source connector.
- **namespace** (string) - Required - The namespace of the source connector.
- **sourceName** (string) - Required - The name of the source connector.
- **instanceId** (string) - Required - The ID of the source connector instance.

### Response
#### Success Response (200)
No response body is specified.

#### Response Example
{}

```

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

### Complete example of a Java function in localrun mode

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

Complete Java function example that appends an exclamation markcreates and starts a LocalRunner with start(false) in localrun mode.

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

    }  

}  

```

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

### Install htpasswd on CentOS/RHEL

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

Install the httpd-tools package to get htpasswd for creating password files.

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

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

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

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

Starts all source connectors for the specified tenant, namespace, and source name.

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

### Description
Starts all source connectors for the specified tenant, namespace, and source name.

### Method
POST

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

### Parameters
#### Path Parameters
- **tenant** (string) - Required - The tenant of the source connector.
- **namespace** (string) - Required - The namespace of the source connector.
- **sourceName** (string) - Required - The name of the source connector.

### Response
#### Success Response (200)
No response body is specified.

#### Response Example
{}

```

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

### Install htpasswd on Ubuntu/Debian

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

Install the apache2-utils package to get htpasswd for creating password files.

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

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

### Successful startup log

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

Log output indicating the Go function started successfully.

```log
...  

 07:55:03.724 [main] INFO  org.apache.pulsar.functions.runtime.ProcessRuntime - Started process successfully  

 ...
```

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

### List updatable broker conf - example output

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

Example output showing a list of updatable configuration parameters.

```text
forceDeleteNamespaceAllowed  

loadBalancerMemoryResourceWeight  

allowAutoTopicCreation  

brokerDeleteInactivePartitionedTopicMetadataEnabled  

managedLedgerInactiveLedgerRolloverTimeSeconds  

loadBalancerNamespaceBundleMaxMsgRate  

resourceUsageTransportPublishIntervalInSecs  

# omit...
```

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

### Response example for listing sources

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

Example response showing a list of source connector names.

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

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

### Log indicating Python function started successfully

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

Example log output showing that the Python function process started successfully.

```log
 ...  

 07:55:03.724 [main] INFO  org.apache.pulsar.functions.runtime.ProcessRuntime - Started process successfully  

 ...  

```

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

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

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
- **tenant** (string) - Required - The tenant of the sink connector.
- **namespace** (string) - Required - The namespace of the sink connector.
- **sinkName** (string) - Required - The name of the sink connector.
- **instanceId** (string) - Required - The ID of the sink connector instance.

### Response
#### Success Response (200)
No response body is specified.

#### Response Example
{}

```

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

### List updated broker conf - example output

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

Example output showing a dynamically updated parameter and its value.

```text
brokerShutdownTimeoutMs    100
```

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

### Start Pulsar standalone

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

Starts a Pulsar standalone instance bound to 127.0.0.1. Run this in the directory where the Pulsar tarball was downloaded (for example, ~/path/to/apache-pulsar-2.5.1).

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

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

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

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

Starts all sink connectors for the specified tenant, namespace, and sink name.

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

### Description
Starts all sink connectors for the specified tenant, namespace, and sink name.

### Method
POST

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

### Parameters
#### Path Parameters
- **tenant** (string) - Required - The tenant of the sink connector.
- **namespace** (string) - Required - The namespace of the sink connector.
- **sinkName** (string) - Required - The name of the sink connector.

### Response
#### Success Response (200)
No response body is specified.

#### Response Example
{}

```

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

### Start Pulsar standalone for Netty TCP example

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

Starts Pulsar standalone in a Docker container named pulsar-netty-standalone, mapping ports 6650 and 8080.

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

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

### Get message TTL (pulsar-admin)

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

Gets the configured message TTL for a namespace. The example continues from the set command, so the returned value is 100 seconds.

```bash
pulsar-admin namespaces get-message-ttl test-tenant/namespace1
```

```bash
100
```

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

### example-window-function-config.yaml

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

Configuration file for the window function. Defines tenant, namespace, name, className, inputs, output, autoAck, parallelism, and windowConfig with windowLengthCount and slidingIntervalCount.

```yaml
tenant: "test"  

namespace: "test-namespace"  

name: "window-example"  

className: "org.apache.pulsar.functions.api.examples.AddWindowFunction"  

inputs: ["test_window_src"]  

output: "test_window_result"  

autoAck: true  

parallelism: 1  

  

# every 5 messages, calculate sum of the latest 10 messages  

windowConfig:  

  windowLengthCount: 10  

  slidingIntervalCount: 5  

```

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

### Get offload policies with Java admin client

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

Use the Java admin client to get the offload policies. The example calls admin.topics().getOffloadPolicies(topic).

```java
admin.topics().getOffloadPolicies(topic)
```

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

### Example config.ini for Python runtime

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

Example configuration file in .ini format. Each parameter is set as key = value. Command-line values take precedence over configuration file values.

```ini
[DEFAULT]  

logging_level = info  

max_pending_async_requests = 1000  

max_concurrent_requests = 50  
```

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

### List tenants to verify creation

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

Lists all tenants to confirm that the 'apache' tenant was created successfully.

```bash
bin/pulsar-admin tenants list  
```

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

### Get inactive topic policies with Java admin client

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

Use the Java admin client to get the inactive topic policies. The example calls admin.topics().getInactiveTopicPolicies(topic).

```java
admin.topics().getInactiveTopicPolicies(topic)
```

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

### Login to IBM Cloud and list plugins

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

Run these commands to log in to IBM Cloud and verify installed plugins. The first command logs in, and the second lists installed plugins to confirm the required IBM Cloud CLI plugins are present.

```bash
$ ibmcloud login  

   $ ibmcloud plugin list  
```

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

### Get deduplication snapshot interval with Java admin client

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

Use the Java admin client to get the deduplication snapshot interval. The example calls admin.topics().getDeduplicationSnapshotInterval(topic).

```java
admin.topics().getDeduplicationSnapshotInterval(topic)
```

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

### Example service URLs for Pulsar client

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

These are example URLs to connect to the Pulsar client. Replace the values in the client configuration with these.

```text
webServiceUrl=http://127.0.0.1:61853/  
brokerServiceUrl=pulsar://127.0.0.1:61854/  
```

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

### Example response for position stats in pending ack

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

Example JSON response returned when getting position stats in pending ack. It shows the State field with a value of MarkDelete, indicating the position is already acknowledged.

```json
{  

  "State" : "MarkDelete"  

}  
```

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

### Get managedLedgerName for a persistent topic

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

Shows the managedLedgerName format for a persistent topic. The example uses persistent://public/default/topics-name.

```text
managedLedgerName of persistent://public/default/topics-name is public/default/persistent/topics-name.
```

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

### Package URL examples

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

Shows the URL format for referencing packages of different types (sink, function, source).

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

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

### Get backlog size with Java admin client

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

Use the Java admin client to get the backlog size by message ID. The example creates a topic string and a MessageId, then calls admin.topics().getBacklogSizeByMessageId(topic, messageId).

```java
String topic = "persistent://my-tenant/my-namespace/my-topic";
MessageId messageId = MessageId.earliest;
admin.topics().getBacklogSizeByMessageId(topic, messageId);
```

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

### Get retention (pulsar-admin)

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

Shows the retention information of a given namespace. The example output shows the configured retention time and size.

```bash
pulsar-admin namespaces get-retention test-tenant/namespace1
```

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

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

### Transaction coordinator metadata setup success output

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

Expected output after running the initialize-transaction-coordinator-metadata command.

```text
Transaction coordinator metadata setup success  
```

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

### Configure Athenz authentication in Java client

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

Java client example for Athenz authentication. Requires the four parameters tenantDomain, tenantService, providerDomain, and privateKey; keyId is optional (default "0"). The privateKey supports file:///path, file:/path, or data:application/x-pem-file;base64,<value> formats.

```java
Map<String, String> authParams = new HashMap();  

authParams.put("ztsUrl", "http://localhost:9998");  

// authParams.put("ztsProxyUrl", "http://localhost:9999"); // Proxy for accessing ZTS (optional, since v3.0.10/v4.0.3/v4.1.0)  

authParams.put("tenantDomain", "shopping"); // Tenant domain name  

authParams.put("tenantService", "some_app"); // Tenant service name  

authParams.put("providerDomain", "pulsar"); // Provider domain name  

authParams.put("privateKey", "file:///path/to/private.pem"); // Tenant private key path  

authParams.put("keyId", "v1"); // Key id for the tenant private key (optional, default: "0")  

  

Authentication athenzAuth = AuthenticationFactory  

        .create(AuthenticationAthenz.class.getName(), authParams);  

  

PulsarClient client = PulsarClient.builder()  

        .serviceUrl("pulsar://my-broker.com:6650")  

        .authentication(athenzAuth)  

        .build();  

```

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

### Get Replication Clusters

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

Gets the list of replication clusters for a given namespace. The pulsar-admin command uses the `get-clusters` subcommand and specifies the namespace. The REST API uses a GET request to the `/admin/v2/namespaces/{tenant}/{namespace}/replication` endpoint. The Java API uses the `getNamespaceReplicationClusters` method. The example output shows a single cluster name.

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

```text
cluster2
```

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

```java
admin.namespaces().getNamespaceReplicationClusters(namespace)
```

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

### SourceConfig example (generic)

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

Example of a SourceConfig object with placeholder values for tenant, namespace, name, className, topicName, configs, parallelism, processingGuarantees, 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  

 }  

}  
```

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

### Check pod status with kubectl get pods

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

Checks the status of all pods in the pulsar namespace. If all pods start up successfully, the STATUS is changed to Running or Completed.

```bash
kubectl get pods -n pulsar  
```

```bash
NAME                                         READY   STATUS      RESTARTS   AGE  

pulsar-mini-bookie-0                         1/1     Running     0          9m27s  

pulsar-mini-bookie-init-5gphs                0/1     Completed   0          9m27s  

pulsar-mini-broker-0                         1/1     Running     0          9m27s  

pulsar-mini-grafana-6b7bcc64c7-4tkxd         1/1     Running     0          9m27s  

pulsar-mini-prometheus-5fcf5dd84c-w8mgz      1/1     Running     0          9m27s  

pulsar-mini-proxy-0                          1/1     Running     0          9m27s  

pulsar-mini-pulsar-init-t7cqt                0/1     Completed   0          9m27s  

pulsar-mini-toolset-0                        1/1     Running     0          9m27s  

pulsar-mini-zookeeper-0                      1/1     Running     0          9m27s  
```

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

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

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

Retrieves transaction stats in the transaction buffer, including the start position and whether the transaction has been aborted.

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

### Description
Retrieves transaction stats in the transaction buffer, including the start position and whether the transaction has been aborted.

### 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** (integer) - Required - The most significant bits of the transaction ID.
- **leastSigBits** (integer) - Required - The least significant bits of the transaction ID.

### Request Example
No request body required.

### 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
{
  "startPosition" : "137:49759",
  "aborted" : false
}
```

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

### List namespaces owned by broker - example output

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

Example output showing owned namespaces with broker assignment details.

```text
public/default/0x00000000_0x40000000    [broker_assignment=shared is_controlled=false is_active=true]  

public/default/0xc0000000_0xffffffff    [broker_assignment=shared is_controlled=false is_active=true]  

public/functions/0x40000000_0x80000000    [broker_assignment=shared is_controlled=false is_active=true]  

public/functions/0x00000000_0x40000000    [broker_assignment=shared is_controlled=false is_active=true]  

pulsar/standalone/localhost:8080/0x00000000_0xffffffff    [broker_assignment=shared is_controlled=false is_active=true]  

pulsar/localhost:8080/0x00000000_0xffffffff    [broker_assignment=shared is_controlled=false is_active=true]  

public/functions/0x80000000_0xc0000000    [broker_assignment=shared is_controlled=false is_active=true]  

public/default/0x80000000_0xc0000000    [broker_assignment=shared is_controlled=false is_active=true]
```

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

### Start configuration store with pulsar-daemon

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

Command to start the configuration store service using `pulsar-daemon` after the configuration is in place.

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

```

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

### Get function version of a Java window function

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

Use the getFunctionVersion method to retrieve the version of the window function from the WindowContext. The example prints the version to standard output.

```java
public class GetVersionOfWindowFunction implements WindowFunction<String, Void> {  

    @Override  

    public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {  

        String functionVersion = context.getFunctionVersion();  

        System.out.println(functionVersion);  

  

        return null;  

    }  

}  

```
Tokens:
Raw