Manage topics

Important

This page only shows some frequently used operations.

  • For the latest and complete information about Pulsar admin, including commands, flags, descriptions, and more, see Pulsar admin doc

  • For the latest and complete information about REST API, including parameters, responses, samples, and more, see REST API doc.

  • For the latest and complete information about Java admin API, including classes, methods, descriptions, and more, see Java admin API doc.

Pulsar has persistent and non-persistent topics. Persistent topic is a logical endpoint for publishing and consuming messages. The topic name structure for persistent topics is:

  2. persistent://tenant/namespace/topic

Non-persistent topics are used in applications that only consume real-time published messages and do not need persistent guarantee. In this way, it reduces message-publish latency by removing overhead of persisting messages. The topic name structure for non-persistent topics is:

  2. non-persistent://tenant/namespace/topic

Manage topic resources

Whether it is persistent or non-persistent topic, you can obtain the topic resources through pulsar-admin tool, REST API and Java.

note

In REST API, :schema stands for persistent or non-persistent. :tenant, :namespace, :x are variables, replace them with the real tenant, namespace, and x names when using them.
Take GET /admin/v2/:schema/:tenant/:namespace as an example, to get the list of persistent topics in REST API, use https://pulsar.apache.org/admin/v2/persistent/my-tenant/my-namespace. To get the list of non-persistent topics in REST API, use https://pulsar.apache.org/admin/v2/non-persistent/my-tenant/my-namespace.

List of topics

You can get the list of topics under a given namespace in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics list \
  3.   my-tenant/my-namespace

GET /admin/v2/:schema/:tenant/:namespace

  2. String namespace = "my-tenant/my-namespace";
  3. admin.topics().getList(namespace);

Grant permission

You can grant permissions on a client role to perform specific actions on a given topic in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics grant-permission \
  3.   --actions produce,consume --role application1 \
  4.   persistent://test-tenant/ns1/tp1 \

POST /admin/v2/:schema/:tenant/:namespace/:topic/permissions/:role

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. String role = "test-role";
  4. Set<AuthAction> actions  = Sets.newHashSet(AuthAction.produce, AuthAction.consume);
  5. admin.topics().grantPermission(topic, role, actions);

Get permission

You can fetch permission in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics permissions \
  3.   persistent://test-tenant/ns1/tp1 \
  5. {
  6.     "application1": [
  7.         "consume",
  8.         "produce"
  9.     ]
  10. }

GET /admin/v2/:schema/:tenant/:namespace/:topic/permissions

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.topics().getPermissions(topic);

Revoke permission

You can revoke a permission granted on a client role in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics revoke-permission \
  3.   --role application1 \
  4.   persistent://test-tenant/ns1/tp1 \
  6. {
  7.   "application1": [
  8.     "consume",
  9.     "produce"
  10.   ]
  11. }

DELETE /admin/v2/:schema/:tenant/:namespace/:topic/permissions/:role

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. String role = "test-role";
  4. admin.topics().revokePermissions(topic, role);

Delete topic

You can delete a topic in the following ways. You cannot delete a topic if any active subscription or producers is connected to the topic.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics delete \
  3.   persistent://test-tenant/ns1/tp1 \

DELETE /admin/v2/:schema/:tenant/:namespace/:topic

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.topics().delete(topic);

Unload topic

You can unload a topic in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics unload \
  3.   persistent://test-tenant/ns1/tp1 \

PUT /admin/v2/:schema/:tenant/:namespace/:topic/unload

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.topics().unload(topic);

Get stats

You can check the following statistics of a given non-partitioned topic.

  • msgRateIn: The sum of all local and replication publishers’ publish rates (msg/s).

  • msgThroughputIn: The sum of all local and replication publishers’ publish rates (bytes/s).

  • msgRateOut: The sum of all local and replication consumers’ dispatch rates(msg/s).

  • msgThroughputOut: The sum of all local and replication consumers’ dispatch rates (bytes/s).

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

  • storageSize: The sum of the ledgers’ storage size for this topic. The space used to store the messages for the topic.

  • earliestMsgPublishTimeInBacklogs: The publish time of the earliest message in the backlog (ms).

  • bytesInCounter: Total bytes published to the topic.

  • msgInCounter: Total messages published to the topic.

  • bytesOutCounter: Total bytes delivered to consumers.

  • msgOutCounter: Total messages delivered to consumers.

  • msgChunkPublished: Topic has chunked message published on it.

  • backlogSize: Estimated total unconsumed or backlog size (in bytes).

  • offloadedStorageSize: Space used to store the offloaded messages for the topic (in bytes).

  • waitingPublishers: The number of publishers waiting in a queue in exclusive access mode.

  • deduplicationStatus: The status of message deduplication for the topic.

  • topicEpoch: The topic epoch or empty if not set.

  • nonContiguousDeletedMessagesRanges: The number of non-contiguous deleted messages ranges.

  • nonContiguousDeletedMessagesRangesSerializedSize: The serialized size of non-contiguous deleted messages ranges.

  • publishers: The list of all local publishers into the topic. The list ranges from zero to thousands.

    • accessMode: The type of access to the topic that the producer requires.

    • msgRateIn: The total rate of messages (msg/s) published by this publisher.

    • msgThroughputIn: The total throughput (bytes/s) of the messages published by this publisher.

    • averageMsgSize: The average message size in bytes from this publisher within the last interval.

    • chunkedMessageRate: The total rate of chunked messages published by this publisher.

    • producerId: The internal identifier for this producer on this topic.

    • producerName: The internal identifier for this producer, generated by the client library.

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

    • connectedSince: The timestamp when this producer is created or reconnected last time.

    • clientVersion: The client library version of this producer.

    • metadata: Metadata (key/value strings) associated with this publisher.

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

    • my-subscription: The name of this subscription. It is defined by the client.

      • msgRateOut: The total rate of messages (msg/s) delivered on this subscription.

      • msgThroughputOut: The total throughput (bytes/s) delivered on this subscription.

      • msgBacklog: The number of messages in the subscription backlog.

      • type: The subscription type.

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

      • lastExpireTimestamp: The timestamp of the last message expire execution.

      • lastConsumedFlowTimestamp: The timestamp of the last flow command received.

      • lastConsumedTimestamp: The latest timestamp of all the consumed timestamp of the consumers.

      • lastAckedTimestamp: The latest timestamp of all the acked timestamp of the consumers.

      • bytesOutCounter: Total bytes delivered to consumer.

      • msgOutCounter: Total messages delivered to consumer.

      • msgRateRedeliver: Total rate of messages redelivered on this subscription (msg/s).

      • chunkedMessageRate: Chunked message dispatch rate.

      • backlogSize: Size of backlog for this subscription (in bytes).

      • earliestMsgPublishTimeInBacklog: The publish time of the earliest message in the backlog for the subscription (ms).

      • msgBacklogNoDelayed: Number of messages in the subscription backlog that do not contain the delay messages.

      • blockedSubscriptionOnUnackedMsgs: Flag to verify if a subscription is blocked due to reaching threshold of unacked messages.

      • msgDelayed: Number of delayed messages currently being tracked.

      • unackedMessages: Number of unacknowledged messages for the subscription.

      • activeConsumerName: The name of the consumer that is active for single active consumer subscriptions. For example, failover or exclusive.

      • totalMsgExpired: Total messages expired on this subscription.

      • lastMarkDeleteAdvancedTimestamp: Last MarkDelete position advanced timestamp.

      • durable: Whether the subscription is durable or ephemeral (for example, from a reader).

      • replicated: Mark that the subscription state is kept in sync across different regions.

      • allowOutOfOrderDelivery: Whether out of order delivery is allowed on the Key_Shared subscription.

      • keySharedMode: Whether the Key_Shared subscription mode is AUTO_SPLIT or STICKY.

      • consumersAfterMarkDeletePosition: This is for Key_Shared subscription to get the recentJoinedConsumers in the Key_Shared subscription.

      • nonContiguousDeletedMessagesRanges: The number of non-contiguous deleted messages ranges.

      • nonContiguousDeletedMessagesRangesSerializedSize: The serialized size of non-contiguous deleted messages ranges.

      • consumers: The list of connected consumers for this subscription.

        • msgRateOut: The total rate of messages (msg/s) delivered to the consumer.

        • msgThroughputOut: The total throughput (bytes/s) delivered to the consumer.

        • consumerName: The internal identifier for this consumer, generated by the client library.

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

        • unackedMessages: The number of unacknowledged messages for the consumer.

        • blockedConsumerOnUnackedMsgs: The flag used to verify if the consumer is blocked due to reaching threshold of the unacknowledged messages.

        • lastConsumedTimestamp: The timestamp when the consumer reads a message the last time.

        • lastAckedTimestamp: The timestamp when the consumer acknowledges a message the last time.

        • address: The IP address and source port for the connection of this consumer.

        • connectedSince: The timestamp when this consumer is created or reconnected last time.

        • clientVersion: The client library version of this consumer.

        • bytesOutCounter: Total bytes delivered to consumer.

        • msgOutCounter: Total messages delivered to consumer.

        • msgRateRedeliver: Total rate of messages redelivered by this consumer (msg/s).

        • chunkedMessageRate: The total rate of chunked messages delivered to this consumer.

        • avgMessagesPerEntry: Number of average messages per entry for the consumer consumed.

        • readPositionWhenJoining: The read position of the cursor when the consumer joining.

        • keyHashRanges: Hash ranges assigned to this consumer if is Key_Shared sub mode.

        • metadata: Metadata (key/value strings) associated with this consumer.

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

    • msgRateIn: The total rate (msg/s) of messages received from the remote cluster.

    • msgThroughputIn: The total throughput (bytes/s) received from the remote cluster.

    • msgRateOut: The total rate of messages (msg/s) delivered to the replication-subscriber.

    • msgThroughputOut: The total throughput (bytes/s) delivered to the replication-subscriber.

    • msgRateExpired: The total rate of messages (msg/s) expired.

    • replicationBacklog: The number of messages pending to be replicated to remote cluster.

    • 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.

    • outboundConnection: The address of the outbound replication connection.

    • outboundConnectedSince: The timestamp of establishing outbound connection.

The following is an example of a topic status.

  2. {
  3.   "msgRateIn" : 0.0,
  4.   "msgThroughputIn" : 0.0,
  5.   "msgRateOut" : 0.0,
  6.   "msgThroughputOut" : 0.0,
  7.   "bytesInCounter" : 504,
  8.   "msgInCounter" : 9,
  9.   "bytesOutCounter" : 2296,
  10.   "msgOutCounter" : 41,
  11.   "averageMsgSize" : 0.0,
  12.   "msgChunkPublished" : false,
  13.   "storageSize" : 504,
  14.   "backlogSize" : 0,
  15.   "earliestMsgPublishTimeInBacklogs": 0,
  16.   "offloadedStorageSize" : 0,
  17.   "publishers" : [ {
  18.     "accessMode" : "Shared",
  19.     "msgRateIn" : 0.0,
  20.     "msgThroughputIn" : 0.0,
  21.     "averageMsgSize" : 0.0,
  22.     "chunkedMessageRate" : 0.0,
  23.     "producerId" : 0,
  24.     "metadata" : { },
  25.     "address" : "/127.0.0.1:65402",
  26.     "connectedSince" : "2021-06-09T17:22:55.913+08:00",
  27.     "clientVersion" : "2.9.0-SNAPSHOT",
  28.     "producerName" : "standalone-1-0"
  29.   } ],
  30.   "waitingPublishers" : 0,
  31.   "subscriptions" : {
  32.     "sub-demo" : {
  33.       "msgRateOut" : 0.0,
  34.       "msgThroughputOut" : 0.0,
  35.       "bytesOutCounter" : 2296,
  36.       "msgOutCounter" : 41,
  37.       "msgRateRedeliver" : 0.0,
  38.       "chunkedMessageRate" : 0,
  39.       "msgBacklog" : 0,
  40.       "backlogSize" : 0,
  41.       "earliestMsgPublishTimeInBacklog": 0,
  42.       "msgBacklogNoDelayed" : 0,
  43.       "blockedSubscriptionOnUnackedMsgs" : false,
  44.       "msgDelayed" : 0,
  45.       "unackedMessages" : 0,
  46.       "type" : "Exclusive",
  47.       "activeConsumerName" : "20b81",
  48.       "msgRateExpired" : 0.0,
  49.       "totalMsgExpired" : 0,
  50.       "lastExpireTimestamp" : 0,
  51.       "lastConsumedFlowTimestamp" : 1623230565356,
  52.       "lastConsumedTimestamp" : 1623230583946,
  53.       "lastAckedTimestamp" : 1623230584033,
  54.       "lastMarkDeleteAdvancedTimestamp" : 1623230584033,
  55.       "consumers" : [ {
  56.         "msgRateOut" : 0.0,
  57.         "msgThroughputOut" : 0.0,
  58.         "bytesOutCounter" : 2296,
  59.         "msgOutCounter" : 41,
  60.         "msgRateRedeliver" : 0.0,
  61.         "chunkedMessageRate" : 0.0,
  62.         "consumerName" : "20b81",
  63.         "availablePermits" : 959,
  64.         "unackedMessages" : 0,
  65.         "avgMessagesPerEntry" : 314,
  66.         "blockedConsumerOnUnackedMsgs" : false,
  67.         "lastAckedTimestamp" : 1623230584033,
  68.         "lastConsumedTimestamp" : 1623230583946,
  69.         "metadata" : { },
  70.         "address" : "/127.0.0.1:65172",
  71.         "connectedSince" : "2021-06-09T17:22:45.353+08:00",
  72.         "clientVersion" : "2.9.0-SNAPSHOT"
  73.       } ],
  74.       "allowOutOfOrderDelivery": false,
  75.       "consumersAfterMarkDeletePosition" : { },
  76.       "nonContiguousDeletedMessagesRanges" : 0,
  77.       "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
  78.       "durable" : true,
  79.       "replicated" : false
  80.     }
  81.   },
  82.   "replication" : { },
  83.   "deduplicationStatus" : "Disabled",
  84.   "nonContiguousDeletedMessagesRanges" : 0,
  85.   "nonContiguousDeletedMessagesRangesSerializedSize" : 0
  86. }

To get the status of a topic, you can use the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics stats \
  3.   persistent://test-tenant/ns1/tp1 \

GET /admin/v2/:schema/:tenant/:namespace/:topic/stats

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.topics().getStats(topic);

Get internal stats

You can get the detailed statistics of a topic.

  • entriesAddedCounter: Messages published since this broker loaded this topic.

  • numberOfEntries: The total number of messages being tracked.

  • totalSize: The total storage size in bytes of all messages.

  • currentLedgerEntries: The count of messages written to the ledger that is currently open for writing.

  • currentLedgerSize: The size in bytes of messages written to the ledger that is currently open for writing.

  • lastLedgerCreatedTimestamp: The time when the last ledger is created.

  • lastLedgerCreationFailureTimestamp: The time when the last ledger failed.

  • waitingCursorsCount: The number of cursors that are “caught up” and waiting for a new message to be published.

  • pendingAddEntriesCount: The number of messages that complete (asynchronous) write requests.

  • lastConfirmedEntry: The ledgerid:entryid of the last message that is written successfully. If the entryid is -1, then the ledger is open, yet no entries are written.

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

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

    • ledgerId: The ID of this ledger.

    • entries: The total number of entries belong to this ledger.

    • size: The size of messages written to this ledger (in bytes).

    • offloaded: Whether this ledger is offloaded.

    • metadata: The ledger metadata.

  • schemaLedgers: The ordered list of all ledgers for this topic schema.

    • ledgerId: The ID of this ledger.

    • entries: The total number of entries belong to this ledger.

    • size: The size of messages written to this ledger (in bytes).

    • offloaded: Whether this ledger is offloaded.

    • metadata: The ledger metadata.

  • compactedLedger: The ledgers holding un-acked messages after topic compaction.

    • ledgerId: The ID of this ledger.

    • entries: The total number of entries belong to this ledger.

    • size: The size of messages written to this ledger (in bytes).

    • offloaded: Whether this ledger is offloaded. The value is false for the compacted topic ledger.

  • cursors: The list of all cursors on this topic. Each subscription in the topic stats has a cursor.

    • markDeletePosition: All messages before the markDeletePosition are acknowledged by the subscriber.

    • 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 for new messages to be published.

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

    • messagesConsumedCounter: The 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 acknowledges are being done out of order, the ranges of messages acknowledged between the markDeletePosition and the read-position shows.

    • lastLedgerSwitchTimestamp: The last time the cursor ledger is rolled over.

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

The following is an example of the detailed statistics of a topic.

  2. {
  3.     "entriesAddedCounter":0,
  4.     "numberOfEntries":0,
  5.     "totalSize":0,
  6.     "currentLedgerEntries":0,
  7.     "currentLedgerSize":0,
  8.     "lastLedgerCreatedTimestamp":"2021-01-22T21:12:14.868+08:00",
  9.     "lastLedgerCreationFailureTimestamp":null,
  10.     "waitingCursorsCount":0,
  11.     "pendingAddEntriesCount":0,
  12.     "lastConfirmedEntry":"3:-1",
  13.     "state":"LedgerOpened",
  14.     "ledgers":[
  15.         {
  16.             "ledgerId":3,
  17.             "entries":0,
  18.             "size":0,
  19.             "offloaded":false,
  20.             "metadata":null
  21.         }
  22.     ],
  23.     "cursors":{
  24.         "test":{
  25.             "markDeletePosition":"3:-1",
  26.             "readPosition":"3:-1",
  27.             "waitingReadOp":false,
  28.             "pendingReadOps":0,
  29.             "messagesConsumedCounter":0,
  30.             "cursorLedger":4,
  31.             "cursorLedgerLastEntry":1,
  32.             "individuallyDeletedMessages":"[]",
  33.             "lastLedgerSwitchTimestamp":"2021-01-22T21:12:14.966+08:00",
  34.             "state":"Open",
  35.             "numberOfEntriesSinceFirstNotAckedMessage":0,
  36.             "totalNonContiguousDeletedMessagesRange":0,
  37.             "properties":{
  39.             }
  40.         }
  41.     },
  42.     "schemaLedgers":[
  43.         {
  44.             "ledgerId":1,
  45.             "entries":11,
  46.             "size":10,
  47.             "offloaded":false,
  48.             "metadata":null
  49.         }
  50.     ],
  51.     "compactedLedger":{
  52.         "ledgerId":-1,
  53.         "entries":-1,
  54.         "size":-1,
  55.         "offloaded":false,
  56.         "metadata":null
  57.     }
  58. }

To get the internal status of a topic, you can use the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics stats-internal \
  3.   persistent://test-tenant/ns1/tp1 \

GET /admin/v2/:schema/:tenant/:namespace/:topic/internalStats

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.topics().getInternalStats(topic);

Peek messages

You can peek a number of messages for a specific subscription of a given topic in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics peek-messages \
  3.   --count 10 --subscription my-subscription \
  4.   persistent://test-tenant/ns1/tp1 \
  6. Message ID: 315674752:0
  7. Properties:  {  "X-Pulsar-publish-time" : "2015-07-13 17:40:28.451"  }
  8. msg-payload

GET /admin/v2/:schema/:tenant/:namespace/:topic/subscription/:subName/position/:messagePosition

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. String subName = "my-subscription";
  4. int numMessages = 1;
  5. admin.topics().peekMessages(topic, subName, numMessages);

Get message by ID

You can fetch the message with the given ledger ID and entry ID in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ ./bin/pulsar-admin topics get-message-by-id \
  3.   persistent://public/default/my-topic \
  4.   -l 10 -e 0

GET /admin/v2/:schema/:tenant/:namespace/:topic/ledger/:ledgerId/entry/:entryId

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. long ledgerId = 10;
  4. long entryId = 10;
  5. admin.topics().getMessageById(topic, ledgerId, entryId);

Examine messages

You can examine a specific message on a topic by position relative to the earliest or the latest message.

  • pulsar-admin
  • REST API
  • Java
  2. ./bin/pulsar-admin topics examine-messages \
  3.   persistent://public/default/my-topic \
  4.   -i latest -m 1

GET /admin/v2/:schema/:tenant/:namespace/:topic/examinemessage?initialPosition=:initialPosition&messagePosition=:messagePosition

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.topics().examineMessage(topic, "latest", 1);

Get message ID

You can get message ID published at or just after the given datetime.

  • pulsar-admin
  • REST API
  • Java
  2. ./bin/pulsar-admin topics get-message-id \
  3.   persistent://public/default/my-topic \
  4.   -d 2021-06-28T19:01:17Z

GET /admin/v2/:schema/:tenant/:namespace/:topic/messageid/:timestamp

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. long timestamp = System.currentTimeMillis()
  4. admin.topics().getMessageIdByTimestamp(topic, timestamp);

Skip messages

You can skip a number of messages for a specific subscription of a given topic in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics skip \
  3.   --count 10 --subscription my-subscription \
  4.   persistent://test-tenant/ns1/tp1 \

POST /admin/v2/:schema/:tenant/:namespace/:topic/subscription/:subName/skip/:numMessages

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. String subName = "my-subscription";
  4. int numMessages = 1;
  5. admin.topics().skipMessages(topic, subName, numMessages);

Skip all messages

You can skip all the old messages for a specific subscription of a given topic.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics skip-all \
  3.   --subscription my-subscription \
  4.   persistent://test-tenant/ns1/tp1 \

POST /admin/v2/:schema/:tenant/:namespace/:topic/subscription/:subName/skip_all

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. String subName = "my-subscription";
  4. admin.topics().skipAllMessages(topic, subName);

Reset cursor

You can reset a subscription cursor position back to the position which is recorded X minutes before. It essentially calculates time and position of cursor at X minutes before and resets it at that position. You can reset the cursor in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics reset-cursor \
  3.   --subscription my-subscription --time 10 \
  4.   persistent://test-tenant/ns1/tp1 \

POST /admin/v2/:schema/:tenant/:namespace/:topic/subscription/:subName/resetcursor/:timestamp

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. String subName = "my-subscription";
  4. long timestamp = 2342343L;
  5. admin.topics().resetCursor(topic, subName, timestamp);

Lookup of topic

You can locate the broker URL which is serving the given topic in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics lookup \
  3.   persistent://test-tenant/ns1/tp1 \
  5.  "pulsar://broker1.org.com:4480"

GET /lookup/v2/topic/:schema/:tenant:namespace/:topic

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.lookup().lookupDestination(topic);

Lookup of partitioned topic

You can locate the broker URL of each partitioned topic which is serving the given topic in the following ways.

  • pulsar-admin
  • Java
  2. $ pulsar-admin topics partitioned-lookup \
  3.   persistent://test-tenant/ns1/my-topic \
  5.   "persistent://test-tenant/ns1/my-topic-partition-0   pulsar://localhost:6650"
  6.   "persistent://test-tenant/ns1/my-topic-partition-1   pulsar://localhost:6650"
  7.   "persistent://test-tenant/ns1/my-topic-partition-2   pulsar://localhost:6650"
  8.   "persistent://test-tenant/ns1/my-topic-partition-3   pulsar://localhost:6650"
  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.lookup().lookupPartitionedTopic(topic);

Lookup the partitioned topics sorted by broker URL

  2. $ pulsar-admin topics partitioned-lookup \
  3.   persistent://test-tenant/ns1/my-topic --sort-by-broker \
  5.   "pulsar://localhost:6650   [persistent://test-tenant/ns1/my-topic-partition-0, persistent://test-tenant/ns1/my-topic-partition-1, persistent://test-tenant/ns1/my-topic-partition-2, persistent://test-tenant/ns1/my-topic-partition-3]"

Get bundle

You can check the range of the bundle which contains given topic in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics bundle-range \
  3.   persistent://test-tenant/ns1/tp1 \
  5.  "0x00000000_0xffffffff"

GET /lookup/v2/topic/:topic_domain/:tenant/:namespace/:topic/bundle

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.lookup().getBundleRange(topic);

Get subscriptions

You can check all subscription names for a given topic in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics subscriptions \
  3.   persistent://test-tenant/ns1/tp1 \
  5.  my-subscription

GET /admin/v2/:schema/:tenant/:namespace/:topic/subscriptions

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.topics().getSubscriptions(topic);

Last Message Id

You can get the last committed message ID for a persistent topic. It is available since 2.3.0 release.

  • pulsar-admin
  • REST API
  • Java
  2. pulsar-admin topics last-message-id topic-name

Get /admin/v2/:schema/:tenant/:namespace/:topic/lastMessageId?version=2.10.0

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.topics().getLastMessage(topic);

Get backlog size

You can get the backlog size of a single partition topic or a non-partitioned topic with a given message ID (in bytes).

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics get-backlog-size \
  3.   -m 1:1 \
  4.   persistent://test-tenant/ns1/tp1-partition-0 \

PUT /admin/v2/:schema/:tenant/:namespace/:topic/backlogSize

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

Configure deduplication snapshot interval

Get deduplication snapshot interval

To get the topic-level deduplication snapshot interval, use one of the following methods.

  • Pulsar-admin API
  • REST API
  • Java API
  2. pulsar-admin topics get-deduplication-snapshot-interval options
  2. {@inject: endpoint|GET|/admin/v2/topics/:tenant/:namespace/:topic/deduplicationSnapshotInterval?version=2.10.0}
  2. admin.topics().getDeduplicationSnapshotInterval(topic)

Set deduplication snapshot interval

To set the topic-level deduplication snapshot interval, use one of the following methods.

Prerequisite brokerDeduplicationEnabled must be set to true.

  • Pulsar-admin API
  • REST API
  • Java API
  2. pulsar-admin topics set-deduplication-snapshot-interval options
  2. {@inject: endpoint|POST|/admin/v2/topics/:tenant/:namespace/:topic/deduplicationSnapshotInterval?version=2.10.0}
  2. {
  3.   "interval": 1000
  4. }
  2. admin.topics().setDeduplicationSnapshotInterval(topic, 1000)

Remove deduplication snapshot interval

To remove the topic-level deduplication snapshot interval, use one of the following methods.

  • Pulsar-admin API
  • REST API
  • Java API
  2. pulsar-admin topics remove-deduplication-snapshot-interval options
  2. {@inject: endpoint|DELETE|/admin/v2/topics/:tenant/:namespace/:topic/deduplicationSnapshotInterval?version=2.10.0}
  2. admin.topics().removeDeduplicationSnapshotInterval(topic)

Configure inactive topic policies

Get inactive topic policies

To get the topic-level inactive topic policies, use one of the following methods.

  • Pulsar-admin API
  • REST API
  • Java API
  2. pulsar-admin topics get-inactive-topic-policies options
  2. {@inject: endpoint|GET|/admin/v2/topics/:tenant/:namespace/:topic/inactiveTopicPolicies?version=2.10.0}
  2. admin.topics().getInactiveTopicPolicies(topic)

Set inactive topic policies

To set the topic-level inactive topic policies, use one of the following methods.

  • Pulsar-admin API
  • REST API
  • Java API
  2. pulsar-admin topics set-inactive-topic-policies options
  2. {@inject: endpoint|POST|/admin/v2/topics/:tenant/:namespace/:topic/inactiveTopicPolicies?version=2.10.0}
  2. admin.topics().setInactiveTopicPolicies(topic, inactiveTopicPolicies)

Remove inactive topic policies

To remove the topic-level inactive topic policies, use one of the following methods.

  • Pulsar-admin API
  • REST API
  • Java API
  2. pulsar-admin topics remove-inactive-topic-policies options
  2. {@inject: endpoint|DELETE|/admin/v2/topics/:tenant/:namespace/:topic/inactiveTopicPolicies?version=2.10.0}
  2. admin.topics().removeInactiveTopicPolicies(topic)

Configure offload policies

Get offload policies

To get the topic-level offload policies, use one of the following methods.

  • Pulsar-admin API
  • REST API
  • Java API
  2. pulsar-admin topics get-offload-policies options
  2. {@inject: endpoint|GET|/admin/v2/topics/:tenant/:namespace/:topic/offloadPolicies?version=2.10.0}
  2. admin.topics().getOffloadPolicies(topic)

Set offload policies

To set the topic-level offload policies, use one of the following methods.

  • Pulsar-admin API
  • REST API
  • Java API
  2. pulsar-admin topics set-offload-policies options
  2. {@inject: endpoint|POST|/admin/v2/topics/:tenant/:namespace/:topic/offloadPolicies?version=2.10.0}
  2. admin.topics().setOffloadPolicies(topic, offloadPolicies)

Remove offload policies

To remove the topic-level offload policies, use one of the following methods.

  • Pulsar-admin API
  • REST API
  • Java API
  2. pulsar-admin topics remove-offload-policies options
  2. {@inject: endpoint|DELETE|/admin/v2/topics/:tenant/:namespace/:topic/offloadPolicies?version=2.10.0}
  2. admin.topics().removeOffloadPolicies(topic)

Manage non-partitioned topics

You can use Pulsar admin API to create, delete and check status of non-partitioned topics.

Create

Non-partitioned topics must be explicitly created. When creating a new non-partitioned topic, you need to provide a name for the topic.

By default, 60 seconds after creation, topics are considered inactive and deleted automatically to avoid generating trash data. To disable this feature, set brokerDeleteInactiveTopicsEnabled to false. To change the frequency of checking inactive topics, set brokerDeleteInactiveTopicsFrequencySeconds to a specific value.

For more information about the two parameters, see here.

You can create non-partitioned topics in the following ways.

  • pulsar-admin
  • REST API
  • Java

When you create non-partitioned topics with the create command, you need to specify the topic name as an argument.

  2. $ bin/pulsar-admin topics create \
  3.   persistent://my-tenant/my-namespace/my-topic
note

When you create a non-partitioned topic with the suffix ‘-partition-‘ followed by numeric value like ‘xyz-topic-partition-x’ for the topic name, if a partitioned topic with same suffix ‘xyz-topic-partition-y’ exists, then the numeric value(x) for the non-partitioned topic must be larger than the number of partitions(y) of the partitioned topic. Otherwise, you cannot create such a non-partitioned topic.

PUT /admin/v2/:schema/:tenant/:namespace/:topic

  2. String topicName = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.topics().createNonPartitionedTopic(topicName);

Delete

You can delete non-partitioned topics in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ bin/pulsar-admin topics delete \
  3.   persistent://my-tenant/my-namespace/my-topic

DELETE /admin/v2/:schema/:tenant/:namespace/:topic

  2. admin.topics().delete(topic);

List

You can get the list of topics under a given namespace in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics list tenant/namespace
  3. persistent://tenant/namespace/topic1
  4. persistent://tenant/namespace/topic2

GET /admin/v2/:schema/:tenant/:namespace

  2. admin.topics().getList(namespace);

Stats

You can check the current statistics of a given topic. The following is an example. For description of each stats, refer to get stats.

  2. {
  3.   "msgRateIn": 4641.528542257553,
  4.   "msgThroughputIn": 44663039.74947473,
  5.   "msgRateOut": 0,
  6.   "msgThroughputOut": 0,
  7.   "averageMsgSize": 1232439.816728665,
  8.   "storageSize": 135532389160,
  9.   "publishers": [
  10.     {
  11.       "msgRateIn": 57.855383881403576,
  12.       "msgThroughputIn": 558994.7078932219,
  13.       "averageMsgSize": 613135,
  14.       "producerId": 0,
  15.       "producerName": null,
  16.       "address": null,
  17.       "connectedSince": null
  18.     }
  19.   ],
  20.   "subscriptions": {
  21.     "my-topic_subscription": {
  22.       "msgRateOut": 0,
  23.       "msgThroughputOut": 0,
  24.       "msgBacklog": 116632,
  25.       "type": null,
  26.       "msgRateExpired": 36.98245516804671,
  27.       "consumers": []
  28.     }
  29.   },
  30.   "replication": {}
  31. }

You can check the current statistics of a given topic and its connected producers and consumers in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics stats \
  3.   persistent://test-tenant/namespace/topic \
  4.   --get-precise-backlog

GET /admin/v2/:schema/:tenant/:namespace/:topic/stats

  2. admin.topics().getStats(topic, false /* is precise backlog */);

Manage partitioned topics

You can use Pulsar admin API to create, update, delete and check status of partitioned topics.

Create

Partitioned topics must be explicitly created. When creating a new partitioned topic, you need to provide a name and the number of partitions for the topic.

By default, 60 seconds after creation, topics are considered inactive and deleted automatically to avoid generating trash data. To disable this feature, set brokerDeleteInactiveTopicsEnabled to false. To change the frequency of checking inactive topics, set brokerDeleteInactiveTopicsFrequencySeconds to a specific value.

For more information about the two parameters, see here.

You can create partitioned topics in the following ways.

  • pulsar-admin
  • REST API
  • Java

When you create partitioned topics with the create-partitioned-topic command, you need to specify the topic name as an argument and the number of partitions using the -p or --partitions flag.

  2. $ bin/pulsar-admin topics create-partitioned-topic \
  3.   persistent://my-tenant/my-namespace/my-topic \
  4.   --partitions 4
note

If a non-partitioned topic with the suffix ‘-partition-‘ followed by a numeric value like ‘xyz-topic-partition-10’, you can not create a partitioned topic with name ‘xyz-topic’, because the partitions of the partitioned topic could override the existing non-partitioned topic. To create such partitioned topic, you have to delete that non-partitioned topic first.

PUT /admin/v2/:schema/:tenant/:namespace/:topic/partitions

  2. String topicName = "persistent://my-tenant/my-namespace/my-topic";
  3. int numPartitions = 4;
  4. admin.topics().createPartitionedTopic(topicName, numPartitions);

Create missed partitions

When topic auto-creation is disabled, and you have a partitioned topic without any partitions, you can use the create-missed-partitions command to create partitions for the topic.

  • pulsar-admin
  • REST API
  • Java

You can create missed partitions with the create-missed-partitions command and specify the topic name as an argument.

  2. $ bin/pulsar-admin topics create-missed-partitions \
  3.   persistent://my-tenant/my-namespace/my-topic \

POST /admin/v2/:schema/:tenant/:namespace/:topic

  2. String topicName = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.topics().createMissedPartitions(topicName);

Get metadata

Partitioned topics are associated with metadata, you can view it as a JSON object. The following metadata field is available.

FieldDescription
partitionsThe number of partitions into which the topic is divided.
  • pulsar-admin
  • REST API
  • Java

You can check the number of partitions in a partitioned topic with the get-partitioned-topic-metadata subcommand.

  2. $ pulsar-admin topics get-partitioned-topic-metadata \
  3.   persistent://my-tenant/my-namespace/my-topic
  4. {
  5.   "partitions": 4
  6. }

GET /admin/v2/:schema/:tenant/:namespace/:topic/partitions

  2. String topicName = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.topics().getPartitionedTopicMetadata(topicName);

Update

You can update the number of partitions for an existing partitioned topic if the topic is non-global. However, you can only add the partition number. Decrementing the number of partitions would delete the topic, which is not supported in Pulsar.

Producers and consumers can find the newly created partitions automatically.

  • pulsar-admin
  • REST API
  • Java

You can update partitioned topics with the update-partitioned-topic command.

  2. $ pulsar-admin topics update-partitioned-topic \
  3.   persistent://my-tenant/my-namespace/my-topic \
  4.   --partitions 8

POST /admin/v2/:schema/:tenant/:cluster/:namespace/:destination/partitions

  2. admin.topics().updatePartitionedTopic(topic, numPartitions);

Delete

You can delete partitioned topics with the delete-partitioned-topic command, REST API and Java.

  • pulsar-admin
  • REST API
  • Java
  2. $ bin/pulsar-admin topics delete-partitioned-topic \
  3.   persistent://my-tenant/my-namespace/my-topic

DELETE /admin/v2/:schema/:topic/:namespace/:destination/partitions

  2. admin.topics().delete(topic);

List

You can get the list of partitioned topics under a given namespace in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics list-partitioned-topics tenant/namespace
  3. persistent://tenant/namespace/topic1
  4. persistent://tenant/namespace/topic2

GET /admin/v2/:schema/:tenant/:namespace

  2. admin.topics().getPartitionedTopicList(namespace);

Stats

You can check the current statistics of a given partitioned topic. The following is an example. For description of each stats, refer to get stats.

Note that in the subscription JSON object, chuckedMessageRate is deprecated. Please use chunkedMessageRate. Both will be sent in the JSON for now.

  2. {
  3.   "msgRateIn" : 999.992947159793,
  4.   "msgThroughputIn" : 1070918.4635439808,
  5.   "msgRateOut" : 0.0,
  6.   "msgThroughputOut" : 0.0,
  7.   "bytesInCounter" : 270318763,
  8.   "msgInCounter" : 252489,
  9.   "bytesOutCounter" : 0,
  10.   "msgOutCounter" : 0,
  11.   "averageMsgSize" : 1070.926056966454,
  12.   "msgChunkPublished" : false,
  13.   "storageSize" : 270316646,
  14.   "backlogSize" : 200921133,
  15.   "publishers" : [ {
  16.     "msgRateIn" : 999.992947159793,
  17.     "msgThroughputIn" : 1070918.4635439808,
  18.     "averageMsgSize" : 1070.3333333333333,
  19.     "chunkedMessageRate" : 0.0,
  20.     "producerId" : 0
  21.   } ],
  22.   "subscriptions" : {
  23.     "test" : {
  24.       "msgRateOut" : 0.0,
  25.       "msgThroughputOut" : 0.0,
  26.       "bytesOutCounter" : 0,
  27.       "msgOutCounter" : 0,
  28.       "msgRateRedeliver" : 0.0,
  29.       "chuckedMessageRate" : 0,
  30.       "chunkedMessageRate" : 0,
  31.       "msgBacklog" : 144318,
  32.       "msgBacklogNoDelayed" : 144318,
  33.       "blockedSubscriptionOnUnackedMsgs" : false,
  34.       "msgDelayed" : 0,
  35.       "unackedMessages" : 0,
  36.       "msgRateExpired" : 0.0,
  37.       "lastExpireTimestamp" : 0,
  38.       "lastConsumedFlowTimestamp" : 0,
  39.       "lastConsumedTimestamp" : 0,
  40.       "lastAckedTimestamp" : 0,
  41.       "consumers" : [ ],
  42.       "isDurable" : true,
  43.       "isReplicated" : false
  44.     }
  45.   },
  46.   "replication" : { },
  47.   "metadata" : {
  48.     "partitions" : 3
  49.   },
  50.   "partitions" : { }
  51. }

You can check the current statistics of a given partitioned topic and its connected producers and consumers in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics partitioned-stats \
  3.   persistent://test-tenant/namespace/topic \
  4.   --per-partition

GET /admin/v2/:schema/:tenant/:namespace/:topic/partitioned-stats

  2. admin.topics().getPartitionedStats(topic, true /* per partition */, false /* is precise backlog */);

Internal stats

You can check the detailed statistics of a topic. The following is an example. For description of each stats, refer to get internal stats.

  2. {
  3.   "entriesAddedCounter": 20449518,
  4.   "numberOfEntries": 3233,
  5.   "totalSize": 331482,
  6.   "currentLedgerEntries": 3233,
  7.   "currentLedgerSize": 331482,
  8.   "lastLedgerCreatedTimestamp": "2016-06-29 03:00:23.825",
  9.   "lastLedgerCreationFailureTimestamp": null,
  10.   "waitingCursorsCount": 1,
  11.   "pendingAddEntriesCount": 0,
  12.   "lastConfirmedEntry": "324711539:3232",
  13.   "state": "LedgerOpened",
  14.   "ledgers": [
  15.     {
  16.       "ledgerId": 324711539,
  17.       "entries": 0,
  18.       "size": 0
  19.     }
  20.   ],
  21.   "cursors": {
  22.     "my-subscription": {
  23.       "markDeletePosition": "324711539:3133",
  24.       "readPosition": "324711539:3233",
  25.       "waitingReadOp": true,
  26.       "pendingReadOps": 0,
  27.       "messagesConsumedCounter": 20449501,
  28.       "cursorLedger": 324702104,
  29.       "cursorLedgerLastEntry": 21,
  30.       "individuallyDeletedMessages": "[(324711539:3134‥324711539:3136], (324711539:3137‥324711539:3140], ]",
  31.       "lastLedgerSwitchTimestamp": "2016-06-29 01:30:19.313",
  32.       "state": "Open"
  33.     }
  34.   }
  35. }

You can get the internal stats for the partitioned topic in the following ways.

  • pulsar-admin
  • REST API
  • Java
  2. $ pulsar-admin topics stats-internal \
  3.   persistent://test-tenant/namespace/topic

GET /admin/v2/:schema/:tenant/:namespace/:topic/internalStats

  2. admin.topics().getInternalStats(topic);

Publish to partitioned topics

By default, Pulsar topics are served by a single broker, which limits the maximum throughput of a topic. Partitioned topics can span multiple brokers and thus allow for higher throughput.

You can publish to partitioned topics using Pulsar client libraries. When publishing to partitioned topics, you must specify a routing mode. If you do not specify any routing mode when you create a new producer, the round robin routing mode is used.

Routing mode

You can specify the routing mode in the ProducerConfiguration object that you use to configure your producer. The routing mode determines which partition(internal topic) that each message should be published to.

The following MessageRoutingMode options are available.

ModeDescription
RoundRobinPartitionIf no key is provided, the producer publishes messages across all partitions in round-robin policy to achieve the maximum throughput. Round-robin is not done per individual message, round-robin is set to the same boundary of batching delay to ensure that batching is effective. If a key is specified on the message, the partitioned producer hashes the key and assigns message to a particular partition. This is the default mode.
SinglePartitionIf no key is provided, the producer picks a single partition randomly and publishes all messages into that partition. If a key is specified on the message, the partitioned producer hashes the key and assigns message to a particular partition.
CustomPartitionUse custom message router implementation that is called to determine the partition for a particular message. You can create a custom routing mode by using the Java client and implementing the MessageRouter interface.

The following is an example:

  2. String pulsarBrokerRootUrl = "pulsar://localhost:6650";
  3. String topic = "persistent://my-tenant/my-namespace/my-topic";
  5. PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(pulsarBrokerRootUrl).build();
  6. Producer<byte[]> producer = pulsarClient.newProducer()
  7.         .topic(topic)
  8.         .messageRoutingMode(MessageRoutingMode.SinglePartition)
  9.         .create();
  10. producer.send("Partitioned topic message".getBytes());

Custom message router

To use a custom message router, you need to provide an implementation of the MessageRouter interface, which has just one choosePartition method:

  2. public interface MessageRouter extends Serializable {
  3.     int choosePartition(Message msg);
  4. }

The following router routes every message to partition 10:

  2. public class AlwaysTenRouter implements MessageRouter {
  3.     public int choosePartition(Message msg) {
  4.         return 10;
  5.     }
  6. }

With that implementation, you can send

  2. String pulsarBrokerRootUrl = "pulsar://localhost:6650";
  3. String topic = "persistent://my-tenant/my-cluster-my-namespace/my-topic";
  5. PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(pulsarBrokerRootUrl).build();
  6. Producer<byte[]> producer = pulsarClient.newProducer()
  7.         .topic(topic)
  8.         .messageRouter(new AlwaysTenRouter())
  9.         .create();
  10. producer.send("Partitioned topic message".getBytes());

How to choose partitions when using a key

If a message has a key, it supersedes the round robin routing policy. The following example illustrates how to choose the partition when using a key.

  2. // If the message has a key, it supersedes the round robin routing policy
  3.         if (msg.hasKey()) {
  4.             return signSafeMod(hash.makeHash(msg.getKey()), topicMetadata.numPartitions());
  5.         }
  7.         if (isBatchingEnabled) { // if batching is enabled, choose partition on `partitionSwitchMs` boundary.
  8.             long currentMs = clock.millis();
  9.             return signSafeMod(currentMs / partitionSwitchMs + startPtnIdx, topicMetadata.numPartitions());
  10.         } else {
  11.             return signSafeMod(PARTITION_INDEX_UPDATER.getAndIncrement(this), topicMetadata.numPartitions());
  12.         }

Manage subscriptions

You can use Pulsar admin API to create, check, and delete subscriptions.

Create subscription

You can create a subscription for a topic using one of the following methods.

  • pulsar-admin
  • REST API
  • Java
  2. pulsar-admin topics create-subscription \
  3. --subscription my-subscription \
  4. persistent://test-tenant/ns1/tp1

PUT /admin/v2/persistent/:tenant/:namespace/:topic/subscription/:subscription

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. String subscriptionName = "my-subscription";
  4. admin.topics().createSubscription(topic, subscriptionName, MessageId.latest);

Get subscription

You can check all subscription names for a given topic using one of the following methods.

  • pulsar-admin
  • REST API
  • Java
  2. pulsar-admin topics subscriptions \
  3. persistent://test-tenant/ns1/tp1 \
  4. my-subscription

GET /admin/v2/:schema/:tenant/:namespace/:topic/subscriptions

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. admin.topics().getSubscriptions(topic);

Unsubscribe subscription

When a subscription does not process messages any more, you can unsubscribe it using one of the following methods.

  • pulsar-admin
  • REST API
  • Java
  2. pulsar-admin topics unsubscribe \
  3. --subscription my-subscription \
  4. persistent://test-tenant/ns1/tp1

DELETE /admin/v2/namespaces/:tenant/:namespace/:topic/subscription/:subscription

  2. String topic = "persistent://my-tenant/my-namespace/my-topic";
  3. String subscriptionName = "my-subscription";
  4. admin.topics().deleteSubscription(topic, subscriptionName);