AdminTools.md

Pulsar admin tool

• Pulsar admin tool
  • What is Pulsar admin tool?
    • How does it work?
  • Prerequisite
    • Pulsar admin CLI tool
    • REST API
    • Java API
    • Pulsar entities
    • Brokers
      • list of active brokers
      • list of namespaces owned by a given broker
    • Properties
      • list existing properties
      • create property
      • get property
      • update property
      • delete property
    • Clusters
      • provision cluster
      • get cluster
      • update cluster
      • delete cluster
      • list all clusters
    • Namespace
      • create namespace
      • get namespace
      • list all namespaces under property
      • list all namespaces under cluster
      • delete namespace
      • grant permission
      • get permission
      • revoke permission
      • set replication cluster
      • get replication cluster
      • set backlog quota policies
      • get backlog quota policies
      • remove backlog quota policies
      • set persistence policies
      • get persistence policies
      • unload namespace bundle
      • set message-ttl
      • get message-ttl
      • split bundle
      • clear backlog
      • clear bundle backlog
      • set retention
      • get retention
    • Persistent
      • create partitioned topic
      • get partitioned topic
      • delete partitioned topic
      • Delete topic
      • List of topics
      • Grant permission
      • Get permission
      • Revoke permission
      • get partitioned topic stats
      • Get stats
      • Get detailed stats
      • Peek messages
      • Skip messages
      • Skip all messages
      • Expire messages
      • Expire all messages
      • Reset cursor
      • Lookup of topic
      • Get subscriptions
      • unsubscribe
    • Namespace isolation policy
      • create/update ns-isolation policy
      • get ns-isolation policy
      • delete ns-isolation policy
      • list all ns-isolation policy
    • Resource-Quotas
      • set namespace resource quota
      • get namespace resource quota
      • reset namespace resource quota
    • Pulsar client tool
      • produce message command
      • consume message command

What is Pulsar admin tool?

Pulsar admin tool is a command line utility which provides an interface between Pulsar-broker and administration to configure and monitor various entities of broker such as properties, clusters, topics, namespaces, etc. It is an admin tool which will be useful to onboard application on Pulsar by allowing creation of property, cluster and namespace. It also helps to administrate status and usage of topics. Other than CLI (command line interface), there are other two alternatives such as REST api and java-api also available to manage and monitor above described entities. Hence, in this document we will walk through how to manage Pulsar entities using CLI, REST api and Java-api.

How does it work?

As we discussed above, Pulsar-broker also exposes admin REST api which allows similar functionality to configure and administrate various entities of Pulsar. Pulsar admin CLI tool calls these REST apis using its java-client api to execute admin commands. You can learn more about list of available REST apis in swagger-documentation.

Prerequisite

Pulsar admin CLI tool

As we discussed Pulsar-Security in other section which allows Pulsar-broker to authenticate and authorize incoming requests. Admin-tool calls Pulsar-broker’s REST api to execute list of commands. However, if Pulsar-broker has security enabled then admin-tool has to pass additional information while calling broker’s REST apis in order to get requests authenticated. Therefore, make sure that you have this information correctly configured in conf/client.conf file.

Now, you are all set to use Pulsar-admin CLI tool. Go to below directory to start with admin tool.

$ bin/pulsar-admin --help

In earlier section, we also talked about other alternative approaches to manage Pulsar-entities - REST api and java-api. In this document we will also talk about REST api end-points and java-api’s snippet to manage the Pulsar entities.

REST API

You can learn about Pulsar-admin REST api definition at swagger-documentation. However, in this document we will see use of each API and how admin CLI command maps to its appropriate REST api.

Java API

Java-api can be accessed by : com.yahoo.pulsar.client.admin.PulsarAdmin

Below code snippet shows how to initialize PulsarAdmin and later in the document we will see how to use it in order to manage Pulsar entities.

URL url = new URL("http://localhost:8080");
String authPluginClassName = "com.org.MyAuthPluginClass"; //Pass auth-plugin class fully-qualified name if Pulsar-security enabled
String authParams = "param1=value1";//Pass auth-param if auth-plugin class requires it
boolean useTls = false;
boolean tlsAllowInsecureConnection = false;
String tlsTrustCertsFilePath = null;

ClientConfiguration config = new ClientConfiguration();
config.setAuthentication(authPluginClassName, authParams);
config.setUseTls(useTls);
config.setTlsAllowInsecureConnection(tlsAllowInsecureConnection);
config.setTlsTrustCertsFilePath(tlsTrustCertsFilePath);

PulsarAdmin admin = new PulsarAdmin(url, config);

Pulsar entities

We can mainly access following broker’s entities using admin command line tool. If you are not already aware about any of following entity then you can read in details at Getting-started-guide. [TODO: add link]

Brokers

It allows to get list of active brokers and list of namespaces owned by a given broker.

list of active brokers

It fetches available active brokers those are serving traffic.

CLI

$ pulsar-admin brokers list use

broker1.use.org.com:8080

REST

GET /admin/brokers/{cluster}

Java

admin.brokers().getActiveBrokers(clusterName)

list of namespaces owned by a given broker

It finds all namespaces which are owned and served by a given broker.

CLI

$ pulsar-admin brokers namespaces --url broker1.use.org.com:8080 use

{
    "my-property/use/my-ns/0x00000000_0xffffffff": {
        "broker_assignment": "shared",
        "is_controlled": false,
        "is_active": true
    }
}

REST

GET /admin/brokers/{cluster}/{broker}/ownedNamespaces

Java

admin.brokers().getOwnedNamespaces(cluster,brokerUrl)

Properties

A property identifies an application domain. For e.g. finance, mail, sports etc are examples of a property. Tool allows to do CRUD operation to manage property in Pulsar.

list existing properties

It lists down all existing properties in Pulsar system.

CLI

$ pulsar-admin properties list

my-property

REST

GET /admin/properties

Java

admin.properties().getProperties()

create property

It creates a new property in Pulsar system. For a property, you can configure admin roles (comma separated) who can manage property and clusters (comma separated) where property will be available.

CLI

pulsar-admin properties create my-property --admin-roles admin1,admin2 --allowed-clusters cl1,cl2

N/A

REST

PUT /admin/properties/{property}

Java

admin.properties().createProperty(property, propertyAdmin)

get property

It gets property configuration for a given existing property.

CLI

$pulsar-admin properties get my-property

{
    "adminRoles": [
        "admin1",
        "admin2"
    ],
    "allowedClusters": [
        "cl1",
        "cl2"
    ]
}

REST

GET /admin/properties/{property}

Java

admin.properties().getPropertyAdmin(property)

update property

It also updates configuration of already created property. You can update admin roles and cluster information for a given existing property.

CLI

$ pulsar-admin properties update my-property --admin-roles admin-update-1,admin-update-2 --allowed-clusters cl1,cl2

N/A

REST

POST /admin/properties/{property}

Java

admin.properties().updateProperty(property, propertyAdmin)

delete property

It deletes an existing property from Pulsar system.

CLI

$ pulsar-admin properties delete my-property

N/A

REST

DELETE /admin/properties/{property}

Java

admin.properties().deleteProperty(property)

Clusters

A cluster allows a property and its namespaces to be available in one or more geographic locations. A cluster typically maps to a region’s colocation name such as use, usw, etc. Tool allows to do CRUD operation to manage cluster in Pulsar.

provision cluster

It provisions a new cluster in Pulsar. It also requires Pulsar super-user privileges to make sure only super-admin can perform such system level operation.

CLI

$ pulsar-admin clusters create --url http://my-cluster.org.com:8080/ --broker-url pulsar://my-cluster.org.com:6650/ cl1

N/A

REST

PUT /admin/clusters/{cluster}

Java

admin.clusters().createCluster(cluster, new ClusterData(serviceUrl, serviceUrlTls, brokerServiceUrl, brokerServiceUrlTls))

get cluster

It gets a cluster configuration for a given existing cluster.

CLI

$ pulsar-admin clusters get cl1

{
    "serviceUrl": "http://my-cluster.org.com:8080/",
    "serviceUrlTls": null,
    "brokerServiceUrl": "pulsar://my-cluster.org.com:6650/",
    "brokerServiceUrlTls": null
}

REST

GET /admin/clusters/{cluster}

Java

admin.clusters().getCluster(cluster)

update cluster

It updates cluster configuration data for a given existing cluster.

CLI

$ pulsar-admin clusters update --url http://my-cluster.org.com:4081/ --broker-url pulsar://my-cluster.org.com:3350/ cl1

N/A

REST

POST /admin/clusters/{cluster}

Java

admin.clusters().updateCluster(cluster, new ClusterData(serviceUrl, serviceUrlTls, brokerServiceUrl, brokerServiceUrlTls))

delete cluster

It deletes an existing cluster from Pulsar system.

CLI

$ pulsar-admin clusters delete cl1

N/A

REST

DELETE /admin/clusters/{cluster}

Java

admin.clusters().deleteCluster(cluster)

list all clusters

It gives a list of call created clusters in Pulsar system.

CLI

$ pulsar-admin clusters list

cl1

REST

GET /admin/clusters

Java

admin.clusters().getClusters()

Namespace

A namespace is a logical nomenclature within a property. A property may have multiple namespaces to manage different applications under that property.

create namespace

It creates a namespace for a property under a given existing cluster.

CLI

$ pulsar-admin namespaces create test-property/cl1/ns1

N/A

REST

PUT /admin/namespaces/{property}/{cluster}/{namespace}

Java

admin.namespaces().createNamespace(namespace)

get namespace

It gets created namespace policies information.

CLI

$pulsar-admin namespaces policies test-property/cl1/ns1

{
    "auth_policies": {
        "namespace_auth": {},
        "destination_auth": {}
    },
    "replication_clusters": [],
    "bundles_activated": true,
    "bundles": {
        "boundaries": [
            "0x00000000",
            "0xffffffff"
        ],
        "numBundles": 1
    },
    "backlog_quota_map": {},
    "persistence": null,
    "latency_stats_sample_rate": {},
    "message_ttl_in_seconds": 0,
    "retention_policies": null,
    "deleted": false
}

REST

GET /admin/namespaces/{property}/{cluster}/{namespace}

Java

admin.namespaces().getPolicies(namespace)

list all namespaces under property

It gets all created namespace names under a given property.

CLI

$ pulsar-admin namespaces list test-property

test-property/cl1/ns1

REST

GET /admin/namespaces/{property}

Java

admin.namespaces().getNamespaces(property)

list all namespaces under cluster

It gets all namespace names under a property for a given cluster

CLI

$ pulsar-admin namespaces list-cluster test-property/cl1

test-property/cl1/ns1

REST

GET /admin/namespaces/{property}/{cluster}

Java

admin.namespaces().getNamespaces(property, cluster)

delete namespace

It deletes an existing namespace

CLI

$ pulsar-admin namespaces delete test-property/cl1/ns1

N/A

REST

DELETE /admin/namespaces/{property}/{cluster}/{namespace}

Java

admin.namespaces().deleteNamespace(namespace)

grant permission

It grants a permission to a specific client role for list of required operations such as produce and consume.

CLI

$ pulsar-admin namespaces grant-permission --actions produce,consume --role admin10 test-property/cl1/ns1

N/A

REST

POST /admin/namespaces/{property}/{cluster}/{namespace}/permissions/{role}

Java

admin.namespaces().grantPermissionOnNamespace(namespace, role, getAuthActions(actions))

get permission

It shows created permission rules for a given namespace

CLI

$ pulsar-admin namespaces permissions test-property/cl1/ns1

{
    "admin10": [
        "produce",
        "consume"
    ]
}   

REST

GET /admin/namespaces/{property}/{cluster}/{namespace}/permissions

Java

admin.namespaces().getPermissions(namespace)

revoke permission

It revokes the permission from a specific client role so, that client-role will not be able to access a given namespace.

CLI

$ pulsar-admin namespaces revoke-permission --role admin10 test-property/cl1/ns1

N/A

REST

DELETE /admin/namespaces/{property}/{cluster}/{namespace}/permissions/{role}

Java

admin.namespaces().revokePermissionsOnNamespace(namespace, role)

set replication cluster

It sets replication clusters for a namespace, so Pulsar can internally replicate publish message from one colo to another colo. However, in order to set replication clusters, your namespace has to be global such as: test-property/global/ns1. It means cluster-name has to be “global”

CLI

$ pulsar-admin namespaces set-clusters --clusters cl2 test-property/cl1/ns1

N/A

REST

POST /admin/namespaces/{property}/{cluster}/{namespace}/replication

Java

admin.namespaces().setNamespaceReplicationClusters(namespace, clusters)

get replication cluster

It gives a list of replication clusters for a given namespace.

CLI

$ pulsar-admin namespaces get-clusters test-property/cl1/ns1

cl2

REST

GET /admin/namespaces/{property}/{cluster}/{namespace}/replication

Java

admin.namespaces().getNamespaceReplicationClusters(namespace)

set backlog quota policies

Backlog quota helps broker to restrict bandwidth/storage of a namespace once it reach certain threshold limit . Admin can set this limit and one of the following action after the limit is reached.

1. producer_request_hold: broker will hold and not persist produce request payload

2. producer_exception: broker will disconnects with client by giving exception

3. consumer_backlog_eviction: broker will start discarding backlog messages

Backlog quota restriction can be taken care by defining restriction of backlog-quota-type: destination_storage

CLI

$ pulsar-admin namespaces set-backlog-quota --limit 10 --policy producer_request_hold test-property/cl1/ns1

N/A

REST

POST /admin/namespaces/{property}/{cluster}/{namespace}/backlogQuota

Java

admin.namespaces().setBacklogQuota(namespace, new BacklogQuota(limit, policy))

get backlog quota policies

It shows a configured backlog quota for a given namespace.

CLI

$ pulsar-admin namespaces get-backlog-quotas test-property/cl1/ns1

{
    "destination_storage": {
        "limit": 10,
        "policy": "producer_request_hold"
    }
}          

REST

GET /admin/namespaces/{property}/{cluster}/{namespace}/backlogQuotaMap

Java

admin.namespaces().getBacklogQuotaMap(namespace)

remove backlog quota policies

It removes backlog quota policies for a given namespace

CLI

$ pulsar-admin namespaces remove-backlog-quota test-property/cl1/ns1

N/A

REST

DELETE /admin/namespaces/{property}/{cluster}/{namespace}/backlogQuota

Java

admin.namespaces().removeBacklogQuota(namespace, backlogQuotaType)

set persistence policies

Persistence policies allow to configure persistency-level for all topic messages under a given namespace.

• Bookkeeper-ack-quorum: Number of acks (guaranteed copies) to wait for each entry, default: 0

• Bookkeeper-ensemble: Number of bookies to use for a topic, default: 0

• Bookkeeper-write-quorum: How many writes to make of each entry, default: 0

• Ml-mark-delete-max-rate: Throttling rate of mark-delete operation (0 means no throttle), default: 0.0

CLI

$ pulsar-admin namespaces set-persistence --bookkeeper-ack-quorum 2 --bookkeeper-ensemble 3 --bookkeeper-write-quorum 2 --ml-mark-delete-max-rate 0 test-property/cl1/ns1

N/A

REST

POST /admin/persistent/{property}/{cluster}/{namespace}/persistence

Java

admin.namespaces().setPersistence(namespace,new PersistencePolicies(bookkeeperEnsemble, bookkeeperWriteQuorum,bookkeeperAckQuorum,managedLedgerMaxMarkDeleteRate))

get persistence policies

It shows configured persistence policies of a given namespace.

CLI

$ pulsar-admin namespaces set-persistence test-property/cl1/ns1

{
    "bookkeeperEnsemble": 3,
    "bookkeeperWriteQuorum": 2,
    "bookkeeperAckQuorum": 2,
    "managedLedgerMaxMarkDeleteRate": 0
}

REST

GET /admin/namespaces/{property}/{cluster}/{namespace}/persistence

Java

admin.namespaces().getPersistence(namespace)

unload namespace bundle

  Namespace bundle is a virtual group of topics which belong to same namespace. If broker gets overloaded with number of bundles then this command can help to unload heavy bundle from that broker, so it can be served by some other less loaded broker. Namespace bundle is defined with it’s start and end range such as 0x00000000 and 0xffffffff.  

CLI

$ pulsar-admin namespaces unload --bundle 0x00000000_0xffffffff test-property/pstg-gq1/ns1

N/A

REST

PUT /admin/namespaces/{property}/{cluster}/{namespace}/unload

Java

admin.namespaces().unloadNamespaceBundle(namespace, bundle)

set message-ttl

It configures message’s time to live (in seconds) duration.

CLI

$ pulsar-admin namespaces set-message-ttl --messageTTL 100 test-property/cl1/ns1

N/A

REST

POST /admin/namespaces/{property}/{cluster}/{namespace}/messageTTL

Java

admin.namespaces().setNamespaceMessageTTL(namespace, messageTTL)

get message-ttl

It gives a message ttl of configured namespace.

CLI

$ pulsar-admin namespaces get-message-ttl test-property/cl1/ns1

100

REST

GET /admin/namespaces/{property}/{cluster}/{namespace}/messageTTL

Java

admin.namespaces().getNamespaceReplicationClusters(namespace)

split bundle

Each namespace bundle can contain multiple topics and each bundle can be served by only one broker. If bundle gets heavy with multiple live topics in it then it creates load on that broker and in order to resolve this issue, admin can split bundle using this command.

CLI

$ pulsar-admin namespaces split-bundle --bundle 0x00000000_0xffffffff test-property/cl1/ns1

N/A

REST

PUT /admin/namespaces/{property}/{cluster}/{namespace}/{bundle}/split

Java

admin.namespaces().splitNamespaceBundle(namespace, bundle)

clear backlog

It clears all message backlog for all the topics those belong to specific namespace. You can also clear backlog for a specific subscription as well.

CLI

$ pulsar-admin namespaces clear-backlog --sub my-subscription test-property/pstg-gq1/ns1

N/A

REST

POST /admin/namespaces/{property}/{cluster}/{namespace}/clearBacklog

Java

admin.namespaces().clearNamespaceBacklogForSubscription(namespace, subscription)

clear bundle backlog

It clears all message backlog for all the topics those belong to specific NamespaceBundle. You can also clear backlog for a specific subscription as well.

CLI

$ pulsar-admin namespaces clear-backlog  --bundle 0x00000000_0xffffffff  --sub my-subscription test-property/pstg-gq1/ns1

N/A

REST

POST /admin/namespaces/{property}/{cluster}/{namespace}/{bundle}/clearBacklog

Java

admin.namespaces().clearNamespaceBundleBacklogForSubscription(namespace, bundle, subscription)

set retention

Each namespace contains multiple topics and each topic’s retention size (storage size) should not exceed to a specific threshold or it should be stored till certain time duration. This command helps to configure retention size and time of topics in a given namespace.

CLI

$ pulsar-admin set-retention --size 10 --time 100 test-property/cl1/ns1

N/A

REST

POST /admin/namespaces/{property}/{cluster}/{namespace}/retention

Java

admin.namespaces().setRetention(namespace, new RetentionPolicies(retentionTimeInMin, retentionSizeInMB))

get retention

It shows retention information of a given namespace.

CLI

$ pulsar-admin namespaces get-retention test-property/cl1/ns1

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

REST

GET /admin/namespaces/{property}/{cluster}/{namespace}/retention

Java

admin.namespaces().getRetention(namespace)

Persistent

Persistent helps to access topic which is a logical endpoint for publishing and consuming messages. Producers publish messages to the topic and consumers subscribe to the topic, to consume messages published to the topic.

In below instructions and commands - persistent topic format is:

persistent://<property_name> <cluster_name> <namespace_name> <topic-name>

create partitioned topic

It creates a partitioned topic under a given namespace. In order to create a partitioned topic, no of partitioned must be more than 1.

CLI

$ pulsar-admin persistent create-partitioned-topic --partitions 2 persistent://test-property/cl1/ns1/pt1

N/A

REST

PUT /admin/persistent/{property}/{cluster}/{namespace}/{destination}/partitions

Java

admin.persistentTopics().createPartitionedTopic(persistentTopic, numPartitions)

get partitioned topic

It gives metadata of created partitioned topic.

CLI

$ pulsar-admin persistent get-partitioned-topic-metadata persistent://test-property/cl1/ns1/pt1

{
    "partitions": 2
}

REST

GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/partitions

Java

admin.persistentTopics().getPartitionedTopicMetadata(persistentTopic)

delete partitioned topic

It deletes a created partitioned topic.

CLI

$ pulsar-admin persistent delete-partitioned-topic persistent://test-property/cl1/ns1/pt1

N/A

REST

DELETE /admin/persistent/{property}/{cluster}/{namespace}/{destination}/partitions

Java

admin.persistentTopics().deletePartitionedTopic(persistentTopic)

Delete topic

It deletes a topic. The topic cannot be deleted if there's any active subscription or producers connected to it.

CLI

pulsar-admin persistent delete persistent://test-property/cl1/ns1/my-topic

N/A

REST

DELETE /admin/persistent/{property}/{cluster}/{namespace}/{destination}

Java

admin.persistentTopics().delete(persistentTopic)

List of topics

It provides a list of persistent topics exist under a given namespace.

CLI

$ pulsar-admin persistent list test-property/cl1/ns1

my-topic

REST

GET /admin/persistent/{property}/{cluster}/{namespace}

Java

admin.persistentTopics().getList(namespace)

Grant permission

It grants permissions on a client role to perform specific actions on a given topic.

CLI

$ pulsar-admin persistent grant-permission --actions produce,consume --role application1 persistent://test-property/cl1/ns1/tp1

N/A

REST

POST /admin/persistent/{property}/{cluster}/{namespace}/{destination}/permissions/{role}

Java

admin.persistentTopics().grantPermission(destination, role, getAuthActions(actions))

Get permission

It shows a list of client role permissions on a given topic.

CLI

$ pulsar-admin permissions persistent://test-property/cl1/ns1/tp1

{
    "application1": [
        "consume",
        "produce"
    ]
}

REST

GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/permissions

Java

admin.persistentTopics().getPermissions(destination)

Revoke permission

It revokes a permission which was granted on a client role.

CLI

$ pulsar-admin persistent revoke-permission --role application1 persistent://test-property/cl1/ns1/tp1

N/A

REST

DELETE /admin/persistent/{property}/{cluster}/{namespace}/{destination}/permissions/{role}

Java

admin.persistentTopics().revokePermissions(destination, role)

get partitioned topic stats

It shows current statistics of a given partitioned topic.

CLI

$ pulsar-admin persistent partitioned-stats --per-partition persistent://test-property/cl1/ns1/tp1

{
    "msgRateIn": 4641.528542257553,
    "msgThroughputIn": 44663039.74947473,
    "msgRateOut": 0,
    "msgThroughputOut": 0,
    "averageMsgSize": 1232439.816728665,
    "storageSize": 135532389160,
    "publishers": [
        {
            "msgRateIn": 57.855383881403576,
            "msgThroughputIn": 558994.7078932219,
            "averageMsgSize": 613135,
            "producerId": 0,
            "producerName": null,
            "address": null,
            "connectedSince": null
        }
    ],
    "subscriptions": {
        "my-topic_subscription": {
            "msgRateOut": 0,
            "msgThroughputOut": 0,
            "msgBacklog": 116632,
            "type": null,
            "msgRateExpired": 36.98245516804671,
            "consumers": []
        }
    },
    "replication": {}
}          

REST

GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/partitioned-stats

Java

admin.persistentTopics().getPartitionedStats(persistentTopic, perPartition)

Get stats

It shows current statistics of a given non-partitioned topic.

• msgRateIn: The sum of all local and replication publishers' publish rates in messages per second

• msgThroughputIn: Same as above, but in bytes per second instead of messages per second

• msgRateOut: The sum of all local and replication consumers' dispatch rates in messages per second

• msgThroughputOut: Same as above, but in bytes per second instead of messages per second

• averageMsgSize: The average size in bytes of messages published within the last interval

• storageSize: The sum of the ledgers' storage size for this topic. See

• publishers: The list of all local publishers into the topic. There can be zero or thousands

• averageMsgSize: Average message size in bytes from this publisher within the last interval

• producerId: Internal identifier for this producer on this topic

• producerName: Internal identifier for this producer, generated by the client library

• address: IP address and source port for the connection of this producer

• connectedSince: Timestamp this producer was created or last reconnected

• subscriptions: The list of all local subscriptions to the topic

• my-subscription: The name of this subscription (client defined)

• msgBacklog: The count of messages in backlog for this subscription

• type: This subscription type

• msgRateExpired: The rate at which messages were discarded instead of dispatched from this subscription due to TTL

• consumers: The list of connected consumers for this subscription

• consumerName: Internal identifier for this consumer, generated by the client library

• availablePermits: The number of messages this consumer has space for in the client library's listen queue. A value of 0 means the client library's queue is full and receive() isn't being called. A nonzero value means this consumer is ready to be dispatched messages.

• replication: This section gives the stats for cross-colo replication of this topic

• replicationBacklog: The outbound replication backlog in messages

• connected: Whether the outbound replicator is connected

• replicationDelayInSeconds: How long the oldest message has been waiting to be sent through the connection, if connected is true

• inboundConnection: The IP and port of the broker in the remote cluster's publisher connection to this broker

• inboundConnectedSince: The TCP connection being used to publish messages to the remote cluster. If there are no local publishers connected, this connection is automatically closed after a minute.

CLI

$ pulsar-admin persistent stats persistent://test-property/cl1/ns1/tp1

{
    "msgRateIn": 0,
    "msgThroughputIn": 0,
    "msgRateOut": 0,
    "msgThroughputOut": 0,
    "averageMsgSize": 0,
    "storageSize": 11814,
    "publishers": [
        {
            "msgRateIn": 0,
            "msgThroughputIn": 0,
            "averageMsgSize": 0,
            "producerId": 0,
            "producerName": "gq1-54-4001",
            "address": "/10.215.138.238:44458",
            "connectedSince": "2016-06-16 22:56:56.509"
        }
    ],
    "subscriptions": {
        "my-subscription": {
            "msgRateOut": 0,
            "msgThroughputOut": 0,
            "msgBacklog": 17,
            "type": "Shared",
            "msgRateExpired": 2.1771406267194497,
            "consumers": [
                {
                    "msgRateOut": 0,
                    "msgThroughputOut": 0,
                    "consumerName": "a67f7",
                    "availablePermits": 1186,
                    "address": "/10.215.166.121:35095",
                    "connectedSince": "2016-06-25 00:05:58.312"
                }
            ]
        }
    },
    "replication": {}
}

REST

GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/stats

Java

admin.persistentTopics().getStats(persistentTopic)

Get detailed stats

It shows detailed statistics of a topic.

• entriesAddedCounter: Messages published since this broker loaded this topic

• numberOfEntries: Total number of messages being tracked

• totalSize: Total storage size in bytes of all messages

• currentLedgerEntries: Count of messages written to the ledger currently open for writing

• currentLedgerSize: Size in bytes of messages written to ledger currently open for writing

• lastLedgerCreatedTimestamp: time when last ledger was created

• lastLedgerCreationFailureTimestamp: time when last ledger was failed

• waitingCursorsCount: How many cursors are "caught up" and waiting for a new message to be published

• pendingAddEntriesCount: How many messages have (asynchronous) write requests we are waiting on completion

• lastConfirmedEntry: The ledgerid:entryid of the last message successfully written. If the entryid is -1, then the ledger has been opened or is currently being opened but has no entries written yet.

• state: The state of this ledger for writing. LedgerOpened means we have a ledger open for saving published messages.

• ledgers: The ordered list of all ledgers for this topic holding its messages

• cursors: The list of all cursors on this topic. There will be one for every subscription you saw in the topic stats.

• markDeletePosition: The ack position: the last message the subscriber acknowledged receiving

• readPosition: The latest position of subscriber for reading message

• waitingReadOp: This is true when the subscription has read the latest message published to the topic and is waiting on new messages to be published.

• pendingReadOps: The counter for how many outstanding read requests to the BookKeepers we have in progress

• messagesConsumedCounter: Number of messages this cursor has acked since this broker loaded this topic

• cursorLedger: The ledger being used to persistently store the current markDeletePosition

• cursorLedgerLastEntry: The last entryid used to persistently store the current markDeletePosition

• individuallyDeletedMessages: If Acks are being done out of order, shows the ranges of messages Acked between the markDeletePosition and the read-position

• lastLedgerSwitchTimestamp: The last time the cursor ledger was rolled over

• state: The state of the cursor ledger: Open means we have a cursor ledger for saving updates of the markDeletePosition.

CLI

$ pulsar-admin persistent stats-internal persistent://test-property/cl1/ns1/tp1

{
    "entriesAddedCounter": 20449518,
    "numberOfEntries": 3233,
    "totalSize": 331482,
    "currentLedgerEntries": 3233,
    "currentLedgerSize": 331482,
    "lastLedgerCreatedTimestamp": "2016-06-29 03:00:23.825",
    "lastLedgerCreationFailureTimestamp": null,
    "waitingCursorsCount": 1,
    "pendingAddEntriesCount": 0,
    "lastConfirmedEntry": "324711539:3232",
    "state": "LedgerOpened",
    "ledgers": [
        {
            "ledgerId": 324711539,
            "entries": 0,
            "size": 0
        }
    ],
    "cursors": {
        "my-subscription": {
            "markDeletePosition": "324711539:3133",
            "readPosition": "324711539:3233",
            "waitingReadOp": true,
            "pendingReadOps": 0,
            "messagesConsumedCounter": 20449501,
            "cursorLedger": 324702104,
            "cursorLedgerLastEntry": 21,
            "individuallyDeletedMessages": "[(324711539:3134‥324711539:3136], (324711539:3137‥324711539:3140], ]",
            "lastLedgerSwitchTimestamp": "2016-06-29 01:30:19.313",
            "state": "Open"
        }
    }
}

REST

GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/internalStats

Java

admin.persistentTopics().getInternalStats(persistentTopic)

Peek messages

It peeks N messages for a specific subscription of a given topic.

CLI

$ pulsar-admin persistent peek-messages --count 10 --subscription my-subscription persistent://test-property/cl1/ns1/my-topic

Message ID: 315674752:0  
Properties:  {  "X-Pulsar-publish-time" : "2015-07-13 17:40:28.451"  }
msg-payload

REST

GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/position/{messagePosition}

Java

admin.persistentTopics().peekMessages(persistentTopic, subName, numMessages)

Skip messages

It skips N messages for a specific subscription of a given topic.

CLI

$ pulsar-admin persistent skip --count 10 --subscription my-subscription persistent://test-property/cl1/ns1/my-topic

N/A

REST

POST /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/skip/{numMessages}

Java

admin.persistentTopics().skipMessages(persistentTopic, subName, numMessages)

Skip all messages

It skips all old messages for a specific subscription of a given topic.

CLI

$ pulsar-admin persistent skip-all --subscription my-subscription persistent://test-property/cl1/ns1/my-topic

N/A

REST

POST /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/skip_all

Java

admin.persistentTopics().skipAllMessages(persistentTopic, subName)

Expire messages

It expires messages which are older than given expiry time (in seconds) for a specific subscription of a given topic.

CLI

$ pulsar-admin persistent expire-messages --subscription my-subscription --expireTime 120 persistent://test-property/cl1/ns1/my-topic

N/A

REST

POST /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/expireMessages/{expireTimeInSeconds}

Java

admin.persistentTopics().expireMessages(persistentTopic, subName, expireTimeInSeconds)

Expire all messages

It expires messages which are older than given expiry time (in seconds) for all subscriptions of a given topic.

CLI

$ pulsar-admin persistent expire-messages-all-subscriptions --expireTime 120 persistent://test-property/cl1/ns1/my-topic

N/A

REST

POST /admin/persistent/{property}/{cluster}/{namespace}/{destination}/all_subscription/expireMessages/{expireTimeInSeconds}

Java

admin.persistentTopics().expireMessagesForAllSubscriptions(persistentTopic, expireTimeInSeconds)

Reset cursor

It resets a subscription’s cursor position back to the position which was recorded X minutes before. It essentially calculates time and position of cursor at X minutes before and resets it at that position.

CLI

$ pulsar-admin persistent reset-cursor --subscription my-subscription --time 10 persistent://test-property/pstg-gq1/ns1/my-topic

N/A

REST

POST /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/resetcursor/{timestamp}

Java

admin.persistentTopics().resetCursor(persistentTopic, subName, timestamp)

Lookup of topic

It locates broker url which is serving the given topic.

CLI

$ pulsar-admin persistent lookup persistent://test-property/pstg-gq1/ns1/my-topic

"pulsar://broker1.org.com:4480"

REST

GET http://<broker-url>:<port>/lookup/v2/destination/persistent/{property}/{cluster}/{namespace}/{dest}
(\* this api serves by “lookup” resource and not “persistent”)

Java

admin.lookups().lookupDestination(destination)

Get subscriptions

It shows all subscription names for a given topic.

CLI

$ pulsar-admin persistent subscriptions persistent://test-property/pstg-gq1/ns1/my-topic

my-subscription

REST

GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscriptions

Java

admin.persistentTopics().getSubscriptions(persistentTopic)

unsubscribe

It can also help to unsubscribe a subscription which is no more processing further messages.

CLI

$pulsar-admin persistent unsubscribe --subscription my-subscription persistent://test-property/pstg-gq1/ns1/my-topic

N/A

REST

DELETE /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}

Java

admin.persistentTopics().deleteSubscription(persistentTopic, subName)

Namespace isolation policy

create/update ns-isolation policy

It creates a namespace isolation policy.

• auto-failover-policy-params: comma separated name=value auto failover policy parameters

• auto-failover-policy-type: auto failover policy type name ['min_available']

• namespaces: comma separated namespaces-regex list

• primary: comma separated primary-broker-regex list

• secondary: comma separated secondary-broker-regex list

CLI

$ pulsar-admin ns-isolation-policy --auto-failover-policy-params min_limit=0 --auto-failover-policy-type min_available --namespaces test-property/cl1/ns.*|test-property/cl1/test-ns*.* --secondary broker2.* --primary broker1.* cl1 ns-is-policy1

N/A

REST

POST /admin/clusters/{cluster}/namespaceIsolationPolicies/{policyName}

Java

admin.clusters().createNamespaceIsolationPolicy(clusterName, policyName, namespaceIsolationData);

get ns-isolation policy

It shows a created namespace isolation policy.

CLI

$ pulsar-admin ns-isolation-policy get cl1 ns-is-policy1

{
    "namespaces": [
        "test-property/cl1/ns.*|test-property/cl1/test-ns*.*"
    ],
    "primary": [
        "broker1.*"
    ],
    "secondary": [
        "broker2.*"
    ],
    "auto_failover_policy": {
        "policy_type": "min_available",
        "parameters": {
            "min_limit": "0"
        }
    }
}

REST

GET /admin/clusters/{cluster}/namespaceIsolationPolicies/{policyName}

Java

admin.clusters().getNamespaceIsolationPolicy(clusterName, policyName)

delete ns-isolation policy

It deletes a namespace isolation policy.

CLI

$ pulsar-admin ns-isolation-policy delete ns-is-policy1

N/A

REST

DELETE /admin/clusters/{cluster}/namespaceIsolationPolicies/{policyName}

Java

admin.clusters().deleteNamespaceIsolationPolicy(clusterName, policyName)

list all ns-isolation policy

It shows all namespace isolation policies served by a given cluster.

CLI

$ pulsar-admin ns-isolation-policy list cl1

{
    "ns-is-policy1": {
        "namespaces": [
            "test-property/cl1/ns.*|test-property/cl1/test-ns*.*"
        ],
        "primary": [
            "broker1.*"
        ],
        "secondary": [
            "broker2.*"
        ],
        "auto_failover_policy": {
            "policy_type": "min_available",
            "parameters": {
                "min_limit": "0"
            }
        }
    }
}

REST

GET /admin/clusters/{cluster}/namespaceIsolationPolicies

Java

admin.clusters().getNamespaceIsolationPolicies(clusterName)

Resource-Quotas

set namespace resource quota

It sets customize quota information for a given namespace bundle.

CLI

$ pulsar-admin resource-quotas set --bandwidthIn 10 --bandwidthOut 10 --bundle 0x00000000_0xffffffff --memory 10 --msgRateIn 10 --msgRateOut 10 --namespace test-property/cl1/ns1

N/A

REST

POST /admin/resource-quotas/{property}/{cluster}/{namespace}/{bundle}

Java

admin.resourceQuotas().setNamespaceBundleResourceQuota(namespace, bundle, quota)

get namespace resource quota

It shows configured resource quota information.

CLI

$ pulsar-admin resource-quotas get  --bundle 0x00000000_0xffffffff --namespace test-property/cl1/my-topic

{
    "msgRateIn": 80.40352101165782,
    "msgRateOut": 132.58187392933146,
    "bandwidthIn": 144273.8819600397,
    "bandwidthOut": 234497.9190227951,
    "memory": 199.91739142481595,
    "dynamic": true
}          

REST

GET /admin/resource-quotas/{property}/{cluster}/{namespace}/{bundle}

Java

admin.resourceQuotas().getNamespaceBundleResourceQuota(namespace, bundle)

reset namespace resource quota

It again reverts back the customized resource quota and sets back default resource_quota.

CLI

$ pulsar-admin resource-quotas reset-namespace-bundle-quota --bundle 0x00000000_0xffffffff --namespace test-property/cl1/my-topic

N/A

REST

DELETE /admin/resource-quotas/{property}/{cluster}/{namespace}/{bundle}

Java

admin.resourceQuotas().resetNamespaceBundleResourceQuota(namespace, bundle)

Pulsar additional tools

Pulsar client tool

As we know that Pulsar provides Java api to produce and consume messages on a given topic. However, Pulsar also provides CLI utility which also helps to produce and consume messages on a topic.

In your terminal, go to below directory to play with client tool.

$ $PULSAR_HOME/bin

$ ./pulsar-client --help

produce message command

```pulsar-client produce```
options	description
```-f, --files```	```Comma separated file paths to send. Cannot be used with -m. Either -f or -m must be provided```
```-m, --messages```	```Comma separted string messages to send. Cannot be used with -f. Either -m or -f must be provided```
```-np, --non-persistent```	```Produce non-persistent message, Default: false```
```-n, --num-produce```	```Number of times to send message(s), Default: 1```
```-r, --rate```	```Rate (in msg/sec) at which to produce. Value of 0 will produce messages as fast as possible, Default: 0.0```
```--ttl```	```Time to live (in ms) for the messages. Value of 0 will result in unlimited TTL, Default: 3600000```

consume message command

```pulsar-client consume```
options	description
```-c, --client-id```	```Client ID for subscriber connection. Required if subscriber ID also specified```
```--hex```	```Display binary messages in hex, Default: false```
```-n, --num-messages```	```Number of messages to consume, Default: 1```
```-r, --rate```	```Rate (in msg/sec) at which to consume. Value of 0 will consume messages as fast as possible, Default: 0.0```
```-s, --subscriber-id```	```Subscriber ID for a durable subscriber```