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.
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.
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.
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 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);
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]
It allows to get list of active brokers and list of namespaces owned by a given broker.
It fetches available active brokers those are serving traffic.
$ pulsar-admin brokers list use
broker1.use.org.com:8080
GET /admin/brokers/{cluster}
admin.brokers().getActiveBrokers(clusterName)
It finds all namespaces which are owned and served by a given broker.
$ 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 } }
GET /admin/brokers/{cluster}/{broker}/ownedNamespaces
admin.brokers().getOwnedNamespaces(cluster,brokerUrl)
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.
It lists down all existing properties in Pulsar system.
$ pulsar-admin properties list
my-property
GET /admin/properties
admin.properties().getProperties()
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.
pulsar-admin properties create my-property --admin-roles admin1,admin2 --allowed-clusters cl1,cl2
N/A
PUT /admin/properties/{property}
admin.properties().createProperty(property, propertyAdmin)
It gets property configuration for a given existing property.
$pulsar-admin properties get my-property
{ "adminRoles": [ "admin1", "admin2" ], "allowedClusters": [ "cl1", "cl2" ] }
GET /admin/properties/{property}
admin.properties().getPropertyAdmin(property)
It also updates configuration of already created property. You can update admin roles and cluster information for a given existing property.
$ pulsar-admin properties update my-property --admin-roles admin-update-1,admin-update-2 --allowed-clusters cl1,cl2
N/A
POST /admin/properties/{property}
admin.properties().updateProperty(property, propertyAdmin)
It deletes an existing property from Pulsar system.
$ pulsar-admin properties delete my-property
N/A
DELETE /admin/properties/{property}
admin.properties().deleteProperty(property)
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.
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.
$ pulsar-admin clusters create --url http://my-cluster.org.com:8080/ --broker-url pulsar://my-cluster.org.com:6650/ cl1
N/A
PUT /admin/clusters/{cluster}
admin.clusters().createCluster(cluster, new ClusterData(serviceUrl, serviceUrlTls, brokerServiceUrl, brokerServiceUrlTls))
It gets a cluster configuration for a given existing cluster.
$ pulsar-admin clusters get cl1
{ "serviceUrl": "http://my-cluster.org.com:8080/", "serviceUrlTls": null, "brokerServiceUrl": "pulsar://my-cluster.org.com:6650/", "brokerServiceUrlTls": null }
GET /admin/clusters/{cluster}
admin.clusters().getCluster(cluster)
It updates cluster configuration data for a given existing cluster.
$ pulsar-admin clusters update --url http://my-cluster.org.com:4081/ --broker-url pulsar://my-cluster.org.com:3350/ cl1
N/A
POST /admin/clusters/{cluster}
admin.clusters().updateCluster(cluster, new ClusterData(serviceUrl, serviceUrlTls, brokerServiceUrl, brokerServiceUrlTls))
It deletes an existing cluster from Pulsar system.
$ pulsar-admin clusters delete cl1
N/A
DELETE /admin/clusters/{cluster}
admin.clusters().deleteCluster(cluster)
It gives a list of call created clusters in Pulsar system.
$ pulsar-admin clusters list
cl1
GET /admin/clusters
admin.clusters().getClusters()
A namespace is a logical nomenclature within a property. A property may have multiple namespaces to manage different applications under that property.
It creates a namespace for a property under a given existing cluster.
$ pulsar-admin namespaces create test-property/cl1/ns1
N/A
PUT /admin/namespaces/{property}/{cluster}/{namespace}
admin.namespaces().createNamespace(namespace)
It gets created namespace policies information.
$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 }
GET /admin/namespaces/{property}/{cluster}/{namespace}
admin.namespaces().getPolicies(namespace)
It gets all created namespace names under a given property.
$ pulsar-admin namespaces list test-property
test-property/cl1/ns1
GET /admin/namespaces/{property}
admin.namespaces().getNamespaces(property)
It gets all namespace names under a property for a given cluster
$ pulsar-admin namespaces list-cluster test-property/cl1
test-property/cl1/ns1
GET /admin/namespaces/{property}/{cluster}
admin.namespaces().getNamespaces(property, cluster)
It deletes an existing namespace
$ pulsar-admin namespaces delete test-property/cl1/ns1
N/A
DELETE /admin/namespaces/{property}/{cluster}/{namespace}
admin.namespaces().deleteNamespace(namespace)
It grants a permission to a specific client role for list of required operations such as produce and consume.
$ pulsar-admin namespaces grant-permission --actions produce,consume --role admin10 test-property/cl1/ns1
N/A
POST /admin/namespaces/{property}/{cluster}/{namespace}/permissions/{role}
admin.namespaces().grantPermissionOnNamespace(namespace, role, getAuthActions(actions))
It shows created permission rules for a given namespace
$ pulsar-admin namespaces permissions test-property/cl1/ns1
{ "admin10": [ "produce", "consume" ] }
GET /admin/namespaces/{property}/{cluster}/{namespace}/permissions
admin.namespaces().getPermissions(namespace)
It revokes the permission from a specific client role so, that client-role will not be able to access a given namespace.
$ pulsar-admin namespaces revoke-permission --role admin10 test-property/cl1/ns1
N/A
DELETE /admin/namespaces/{property}/{cluster}/{namespace}/permissions/{role}
admin.namespaces().revokePermissionsOnNamespace(namespace, role)
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”
$ pulsar-admin namespaces set-clusters --clusters cl2 test-property/cl1/ns1
N/A
POST /admin/namespaces/{property}/{cluster}/{namespace}/replication
admin.namespaces().setNamespaceReplicationClusters(namespace, clusters)
It gives a list of replication clusters for a given namespace.
$ pulsar-admin namespaces get-clusters test-property/cl1/ns1
cl2
GET /admin/namespaces/{property}/{cluster}/{namespace}/replication
admin.namespaces().getNamespaceReplicationClusters(namespace)
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.
producer_request_hold: broker will hold and not persist produce request payload
producer_exception: broker will disconnects with client by giving exception
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
$ pulsar-admin namespaces set-backlog-quota --limit 10 --policy producer_request_hold test-property/cl1/ns1
N/A
POST /admin/namespaces/{property}/{cluster}/{namespace}/backlogQuota
admin.namespaces().setBacklogQuota(namespace, new BacklogQuota(limit, policy))
It shows a configured backlog quota for a given namespace.
$ pulsar-admin namespaces get-backlog-quotas test-property/cl1/ns1
{ "destination_storage": { "limit": 10, "policy": "producer_request_hold" } }
GET /admin/namespaces/{property}/{cluster}/{namespace}/backlogQuotaMap
admin.namespaces().getBacklogQuotaMap(namespace)
It removes backlog quota policies for a given namespace
$ pulsar-admin namespaces remove-backlog-quota test-property/cl1/ns1
N/A
DELETE /admin/namespaces/{property}/{cluster}/{namespace}/backlogQuota
admin.namespaces().removeBacklogQuota(namespace, backlogQuotaType)
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
$ 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
POST /admin/persistent/{property}/{cluster}/{namespace}/persistence
admin.namespaces().setPersistence(namespace,new PersistencePolicies(bookkeeperEnsemble, bookkeeperWriteQuorum,bookkeeperAckQuorum,managedLedgerMaxMarkDeleteRate))
It shows configured persistence policies of a given namespace.
$ pulsar-admin namespaces get-persistence test-property/cl1/ns1
{ "bookkeeperEnsemble": 3, "bookkeeperWriteQuorum": 2, "bookkeeperAckQuorum": 2, "managedLedgerMaxMarkDeleteRate": 0 }
GET /admin/namespaces/{property}/{cluster}/{namespace}/persistence
admin.namespaces().getPersistence(namespace)
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.
$ pulsar-admin namespaces unload --bundle 0x00000000_0xffffffff test-property/pstg-gq1/ns1
N/A
PUT /admin/namespaces/{property}/{cluster}/{namespace}/unload
admin.namespaces().unloadNamespaceBundle(namespace, bundle)
It configures message’s time to live (in seconds) duration.
$ pulsar-admin namespaces set-message-ttl --messageTTL 100 test-property/cl1/ns1
N/A
POST /admin/namespaces/{property}/{cluster}/{namespace}/messageTTL
admin.namespaces().setNamespaceMessageTTL(namespace, messageTTL)
It gives a message ttl of configured namespace.
$ pulsar-admin namespaces get-message-ttl test-property/cl1/ns1
100
GET /admin/namespaces/{property}/{cluster}/{namespace}/messageTTL
admin.namespaces().getNamespaceReplicationClusters(namespace)
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.
$ pulsar-admin namespaces split-bundle --bundle 0x00000000_0xffffffff test-property/cl1/ns1
N/A
PUT /admin/namespaces/{property}/{cluster}/{namespace}/{bundle}/split
admin.namespaces().splitNamespaceBundle(namespace, bundle)
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.
$ pulsar-admin namespaces clear-backlog --sub my-subscription test-property/pstg-gq1/ns1
N/A
POST /admin/namespaces/{property}/{cluster}/{namespace}/clearBacklog
admin.namespaces().clearNamespaceBacklogForSubscription(namespace, subscription)
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.
$ pulsar-admin namespaces clear-backlog --bundle 0x00000000_0xffffffff --sub my-subscription test-property/pstg-gq1/ns1
N/A
POST /admin/namespaces/{property}/{cluster}/{namespace}/{bundle}/clearBacklog
admin.namespaces().clearNamespaceBundleBacklogForSubscription(namespace, bundle, subscription)
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.
$ pulsar-admin set-retention --size 10 --time 100 test-property/cl1/ns1
N/A
POST /admin/namespaces/{property}/{cluster}/{namespace}/retention
admin.namespaces().setRetention(namespace, new RetentionPolicies(retentionTimeInMin, retentionSizeInMB))
It shows retention information of a given namespace.
$ pulsar-admin namespaces get-retention test-property/cl1/ns1
{ "retentionTimeInMinutes": 10, "retentionSizeInMB": 100 }
GET /admin/namespaces/{property}/{cluster}/{namespace}/retention
admin.namespaces().getRetention(namespace)
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>
It creates a partitioned topic under a given namespace. In order to create a partitioned topic, no of partitioned must be more than 1.
$ pulsar-admin persistent create-partitioned-topic --partitions 2 persistent://test-property/cl1/ns1/pt1
N/A
PUT /admin/persistent/{property}/{cluster}/{namespace}/{destination}/partitions
admin.persistentTopics().createPartitionedTopic(persistentTopic, numPartitions)
It gives metadata of created partitioned topic.
$ pulsar-admin persistent get-partitioned-topic-metadata persistent://test-property/cl1/ns1/pt1
{ "partitions": 2 }
GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/partitions
admin.persistentTopics().getPartitionedTopicMetadata(persistentTopic)
It deletes a created partitioned topic.
$ pulsar-admin persistent delete-partitioned-topic persistent://test-property/cl1/ns1/pt1
N/A
DELETE /admin/persistent/{property}/{cluster}/{namespace}/{destination}/partitions
admin.persistentTopics().deletePartitionedTopic(persistentTopic)
It deletes a topic. The topic cannot be deleted if there's any active subscription or producers connected to it.
pulsar-admin persistent delete persistent://test-property/cl1/ns1/my-topic
N/A
DELETE /admin/persistent/{property}/{cluster}/{namespace}/{destination}
admin.persistentTopics().delete(persistentTopic)
It provides a list of persistent topics exist under a given namespace.
$ pulsar-admin persistent list test-property/cl1/ns1
my-topic
GET /admin/persistent/{property}/{cluster}/{namespace}
admin.persistentTopics().getList(namespace)
It grants permissions on a client role to perform specific actions on a given topic.
$ pulsar-admin persistent grant-permission --actions produce,consume --role application1 persistent://test-property/cl1/ns1/tp1
N/A
POST /admin/persistent/{property}/{cluster}/{namespace}/{destination}/permissions/{role}
admin.persistentTopics().grantPermission(destination, role, getAuthActions(actions))
It shows a list of client role permissions on a given topic.
$ pulsar-admin permissions persistent://test-property/cl1/ns1/tp1
{ "application1": [ "consume", "produce" ] }
GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/permissions
admin.persistentTopics().getPermissions(destination)
It revokes a permission which was granted on a client role.
$ pulsar-admin persistent revoke-permission --role application1 persistent://test-property/cl1/ns1/tp1
N/A
DELETE /admin/persistent/{property}/{cluster}/{namespace}/{destination}/permissions/{role}
admin.persistentTopics().revokePermissions(destination, role)
It shows current statistics of a given partitioned topic.
$ 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": {} }
GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/partitioned-stats
admin.persistentTopics().getPartitionedStats(persistentTopic, perPartition)
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.
$ 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": {} }
GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/stats
admin.persistentTopics().getStats(persistentTopic)
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.
$ 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" } } }
GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/internalStats
admin.persistentTopics().getInternalStats(persistentTopic)
It peeks N messages for a specific subscription of a given topic.
$ 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
GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/position/{messagePosition}
admin.persistentTopics().peekMessages(persistentTopic, subName, numMessages)
It skips N messages for a specific subscription of a given topic.
$ pulsar-admin persistent skip --count 10 --subscription my-subscription persistent://test-property/cl1/ns1/my-topic
N/A
POST /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/skip/{numMessages}
admin.persistentTopics().skipMessages(persistentTopic, subName, numMessages)
It skips all old messages for a specific subscription of a given topic.
$ pulsar-admin persistent skip-all --subscription my-subscription persistent://test-property/cl1/ns1/my-topic
N/A
POST /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/skip_all
admin.persistentTopics().skipAllMessages(persistentTopic, subName)
It expires messages which are older than given expiry time (in seconds) for a specific subscription of a given topic.
$ pulsar-admin persistent expire-messages --subscription my-subscription --expireTime 120 persistent://test-property/cl1/ns1/my-topic
N/A
POST /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/expireMessages/{expireTimeInSeconds}
admin.persistentTopics().expireMessages(persistentTopic, subName, expireTimeInSeconds)
It expires messages which are older than given expiry time (in seconds) for all subscriptions of a given topic.
$ pulsar-admin persistent expire-messages-all-subscriptions --expireTime 120 persistent://test-property/cl1/ns1/my-topic
N/A
POST /admin/persistent/{property}/{cluster}/{namespace}/{destination}/all_subscription/expireMessages/{expireTimeInSeconds}
admin.persistentTopics().expireMessagesForAllSubscriptions(persistentTopic, expireTimeInSeconds)
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.
$ pulsar-admin persistent reset-cursor --subscription my-subscription --time 10 persistent://test-property/pstg-gq1/ns1/my-topic
N/A
POST /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/resetcursor/{timestamp}
admin.persistentTopics().resetCursor(persistentTopic, subName, timestamp)
It locates broker url which is serving the given topic.
$ pulsar-admin persistent lookup persistent://test-property/pstg-gq1/ns1/my-topic
"pulsar://broker1.org.com:4480"
GET http://<broker-url>:<port>/lookup/v2/destination/persistent/{property}/{cluster}/{namespace}/{dest}
(\* this api serves by “lookup” resource and not “persistent”)
admin.lookups().lookupDestination(destination)
It shows all subscription names for a given topic.
$ pulsar-admin persistent subscriptions persistent://test-property/pstg-gq1/ns1/my-topic
my-subscription
GET /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscriptions
admin.persistentTopics().getSubscriptions(persistentTopic)
It can also help to unsubscribe a subscription which is no more processing further messages.
$pulsar-admin persistent unsubscribe --subscription my-subscription persistent://test-property/pstg-gq1/ns1/my-topic
N/A
DELETE /admin/persistent/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}
admin.persistentTopics().deleteSubscription(persistentTopic, subName)
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
$ 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
POST /admin/clusters/{cluster}/namespaceIsolationPolicies/{policyName}
admin.clusters().createNamespaceIsolationPolicy(clusterName, policyName, namespaceIsolationData);
It shows a created namespace isolation policy.
$ 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" } } }
GET /admin/clusters/{cluster}/namespaceIsolationPolicies/{policyName}
admin.clusters().getNamespaceIsolationPolicy(clusterName, policyName)
It deletes a namespace isolation policy.
$ pulsar-admin ns-isolation-policy delete ns-is-policy1
N/A
DELETE /admin/clusters/{cluster}/namespaceIsolationPolicies/{policyName}
admin.clusters().deleteNamespaceIsolationPolicy(clusterName, policyName)
It shows all namespace isolation policies served by a given cluster.
$ 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" } } } }
GET /admin/clusters/{cluster}/namespaceIsolationPolicies
admin.clusters().getNamespaceIsolationPolicies(clusterName)
It sets customize quota information for a given namespace bundle.
$ 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
POST /admin/resource-quotas/{property}/{cluster}/{namespace}/{bundle}
admin.resourceQuotas().setNamespaceBundleResourceQuota(namespace, bundle, quota)
It shows configured resource quota information.
$ 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 }
GET /admin/resource-quotas/{property}/{cluster}/{namespace}/{bundle}
admin.resourceQuotas().getNamespaceBundleResourceQuota(namespace, bundle)
It again reverts back the customized resource quota and sets back default resource_quota.
$ pulsar-admin resource-quotas reset-namespace-bundle-quota --bundle 0x00000000_0xffffffff --namespace test-property/cl1/my-topic
N/A
DELETE /admin/resource-quotas/{property}/{cluster}/{namespace}/{bundle}
admin.resourceQuotas().resetNamespaceBundleResourceQuota(namespace, bundle)
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