Skip to content

Latest commit

 

History

History
2197 lines (1510 loc) · 45.8 KB

AdminTools.md

File metadata and controls

2197 lines (1510 loc) · 45.8 KB
Raw

Pulsar admin tool

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 /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 /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 /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 /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 /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 /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 /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/ cl1

N/A
REST
PUT /clusters/{cluster}
Java
admin.clusters().createCluster(cluster, new ClusterData(serviceUrl, serviceUrlTls))

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
}
REST
GET /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/ cl1

N/A
REST
POST /clusters/{cluster}
Java
admin.clusters().updateCluster(cluster, new ClusterData(serviceUrl, serviceUrlTls))

delete cluster

It deletes an existing cluster from Pulsar system.

CLI
$ pulsar-admin clusters delete cl1

N/A
REST
DELETE /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 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
PUT /namespaces/{property}/{cluster}/{namespace}/{bundle}/split
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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /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 /persistent/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/skip_all
Java
admin.persistentTopics().skipAllMessages(persistentTopic, subName)

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