Skip to main content
Version: Next

Manage topics

tip

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

  • 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. A persistent topic is a logical endpoint for publishing and consuming messages. The topic name structure for persistent topics is:

persistent://tenant/namespace/topic

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

non-persistent://tenant/namespace/topic
note

Topic naming: due to backward compatibility, some special characters, e.g., "/," are allowed as a part of the topic name. But it's recommended not to use special characters as a part of the topic name.

Manage topic resources​

Whether it is a 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/persistent/{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 topics list my-tenant/my-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 topics grant-permission \
    --actions produce,consume \
    --role application1 \
    persistent://test-tenant/ns1/tp1

Get permission​

You can fetch permission in the following ways.

pulsar-admin topics permissions persistent://test-tenant/ns1/tp1

Example output:

application1    [consume, produce]

Revoke permission​

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

pulsar-admin topics revoke-permission \
    --role application1 \
    persistent://test-tenant/ns1/tp1

Delete topic​

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

pulsar-admin topics delete persistent://test-tenant/ns1/tp1

Unload topic​

You can unload a topic in the following ways.

pulsar-admin topics unload persistent://test-tenant/ns1/tp1

Truncate topic​

You can truncate a topic in the following ways. The truncate operation moves all cursors to the end of the topic and deletes all inactive ledgers, freeing the storage used by messages that have already been consumed.

pulsar-admin topics truncate persistent://test-tenant/ns1/tp1

Get stats​

For the detailed statistics of a topic, see Pulsar statistics.

The following is an example of a topic status.

{
  "msgRateIn" : 0.0,
  "msgThroughputIn" : 0.0,
  "msgRateOut" : 0.0,
  "msgThroughputOut" : 0.0,
  "bytesInCounter" : 504,
  "msgInCounter" : 9,
  "bytesOutCounter" : 2296,
  "msgOutCounter" : 41,
  "averageMsgSize" : 0.0,
  "msgChunkPublished" : false,
  "storageSize" : 504,
  "backlogSize" : 0,
  "filteredEntriesCount" : 100,
  "earliestMsgPublishTimeInBacklogs": 0,
  "offloadedStorageSize" : 0,
  "publishers" : [ {
    "accessMode" : "Shared",
    "msgRateIn" : 0.0,
    "msgThroughputIn" : 0.0,
    "averageMsgSize" : 0.0,
    "chunkedMessageRate" : 0.0,
    "producerId" : 0,
    "metadata" : { },
    "address" : "/127.0.0.1:65402",
    "connectedSince" : "2021-06-09T17:22:55.913+08:00",
    "clientVersion" : "5.0.0-M2",
    "producerName" : "standalone-1-0"
  } ],
  "waitingPublishers" : 0,
  "subscriptions" : {
    "sub-demo" : {
      "msgRateOut" : 0.0,
      "msgThroughputOut" : 0.0,
      "bytesOutCounter" : 2296,
      "msgOutCounter" : 41,
      "msgRateRedeliver" : 0.0,
      "chunkedMessageRate" : 0,
      "msgBacklog" : 0,
      "backlogSize" : 0,
      "earliestMsgPublishTimeInBacklog": 0,
      "msgBacklogNoDelayed" : 0,
      "blockedSubscriptionOnUnackedMsgs" : false,
      "msgDelayed" : 0,
      "unackedMessages" : 0,
      "type" : "Exclusive",
      "activeConsumerName" : "20b81",
      "msgRateExpired" : 0.0,
      "totalMsgExpired" : 0,
      "lastExpireTimestamp" : 0,
      "lastConsumedFlowTimestamp" : 1623230565356,
      "lastConsumedTimestamp" : 1623230583946,
      "lastAckedTimestamp" : 1623230584033,
      "lastMarkDeleteAdvancedTimestamp" : 1623230584033,
      "filterProcessedMsgCount": 100,
      "filterAcceptedMsgCount": 100,
      "filterRejectedMsgCount": 0,
      "filterRescheduledMsgCount": 0,
      "consumers" : [ {
        "msgRateOut" : 0.0,
        "msgThroughputOut" : 0.0,
        "bytesOutCounter" : 2296,
        "msgOutCounter" : 41,
        "msgRateRedeliver" : 0.0,
        "chunkedMessageRate" : 0.0,
        "consumerName" : "20b81",
        "availablePermits" : 959,
        "unackedMessages" : 0,
        "avgMessagesPerEntry" : 314,
        "blockedConsumerOnUnackedMsgs" : false,
        "lastAckedTimestamp" : 1623230584033,
        "lastConsumedTimestamp" : 1623230583946,
        "metadata" : { },
        "address" : "/127.0.0.1:65172",
        "connectedSince" : "2021-06-09T17:22:45.353+08:00",
        "clientVersion" : "5.0.0-M2"
      } ],
      "allowOutOfOrderDelivery": false,
      "consumersAfterMarkDeletePosition" : { },
      "nonContiguousDeletedMessagesRanges" : 0,
      "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
      "durable" : true,
      "replicated" : false
    }
  },
  "replication" : { },
  "deduplicationStatus" : "Disabled",
  "nonContiguousDeletedMessagesRanges" : 0,
  "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
  "ownerBroker" : "localhost:8080"
}

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

pulsar-admin topics stats persistent://test-tenant/ns1/tp1

Get internal stats​

For the detailed internal statistics inside a topic, see Pulsar statistics.

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

{
    "entriesAddedCounter":0,
    "numberOfEntries":0,
    "totalSize":0,
    "currentLedgerEntries":0,
    "currentLedgerSize":0,
    "lastLedgerCreatedTimestamp":"2021-01-22T21:12:14.868+08:00",
    "lastLedgerCreationFailureTimestamp":null,
    "waitingCursorsCount":0,
    "pendingAddEntriesCount":0,
    "lastConfirmedEntry":"3:-1",
    "state":"LedgerOpened",
    "ledgers":[
        {
            "ledgerId":3,
            "entries":0,
            "size":0,
            "offloaded":false,
            "metadata":null
        }
    ],
    "cursors":{
        "test":{
            "markDeletePosition":"3:-1",
            "readPosition":"3:-1",
            "waitingReadOp":false,
            "pendingReadOps":0,
            "messagesConsumedCounter":0,
            "cursorLedger":4,
            "cursorLedgerLastEntry":1,
            "individuallyDeletedMessages":"[]",
            "lastLedgerSwitchTimestamp":"2021-01-22T21:12:14.966+08:00",
            "state":"Open",
            "numberOfEntriesSinceFirstNotAckedMessage":0,
            "totalNonContiguousDeletedMessagesRange":0,
            "properties":{

            }
        }
    },
    "schemaLedgers":[
        {
            "ledgerId":1,
            "entries":11,
            "size":10,
            "offloaded":false,
            "metadata":null
        }
    ],
    "compactedLedger":{
        "ledgerId":-1,
        "entries":-1,
        "size":-1,
        "offloaded":false,
        "metadata":null
    }
}

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

pulsar-admin topics stats-internal persistent://test-tenant/ns1/tp1

Add --metadata to include BookKeeper ledger metadata. The response also includes metadata for the compacted ledger and schema ledgers when they are present.

Analyze a subscription backlog​

Use analyze-backlog to scan the backlog of a persistent topic subscription. --backlog-scan-max-entries enables a client-side loop that resumes after a server scan reaches its limit and combines the results:

pulsar-admin topics analyze-backlog persistent://test-tenant/ns1/tp1 \
  --subscription my-subscription --backlog-scan-max-entries 100000

The entry threshold must be positive. It is checked between scan responses, so the final count can exceed the threshold. The loop also stops when the scan completes or cannot make progress. Without this option, the command performs a single server scan, which can return a partial result.

Use --position ledgerId:entryId to choose a starting position, --quiet to suppress intermediate progress results, and --plain to emit compact newline-delimited JSON. For scripts that need only the final result, combine --quiet --plain.

Peek messages​

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

The REST response separates transaction state from the topic's read boundary:

  • X-Pulsar-txn-uncommitted is present for transactional messages and indicates whether that transaction is still ongoing. It no longer means that the message lies beyond the maximum readable position.
  • X-Pulsar-txn-aborted identifies an aborted transaction's message.
  • X-Pulsar-txn-consumable indicates whether the message position is at or before the topic's maximum readable position. It does not by itself exclude aborted messages. A committed or nontransactional message can still lie beyond this boundary while an earlier transaction is ongoing.

The Java admin client exposes X-Pulsar-txn-consumable as a property of the peeked messages, and includes X-Pulsar-txn-uncommitted and X-Pulsar-txn-aborted properties when their values are true. Its READ_COMMITTED isolation mode excludes aborted or ongoing transaction messages and messages beyond the consumable boundary. Update diagnostic tools that used the old txn-uncommitted header as a position check.

Message inspection endpoints return message properties in HTTP response headers. If large properties exceed the header limit, configure httpMaxResponseHeaderSize on the broker and, when used, the proxy. The default is 8192 bytes. Increasing this limit also increases the memory available to response headers.

For persistent topics, a missing subscription is created at the earliest position only when automatic subscription creation is allowed. If it is disabled, peeking a nonexistent subscription fails with HTTP 412 Precondition Failed (Java admin PreconditionFailedException). Create the subscription explicitly or use an existing one before peeking.

pulsar-admin topics peek-messages \
    --count 10 --subscription my-subscription \
    persistent://test-tenant/ns1/tp1

Example output:

Message ID: 77:2
Publish time: 1668674963028
Event time: 0
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 68 65 6c 6c 6f 2d 31                            |hello-1         |
+--------+-------------------------------------------------+----------------+

Get message by ID​

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

pulsar-admin topics get-message-by-id \
    -l 10 -e 0 persistent://public/default/my-topic

Examine messages​

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

pulsar-admin topics examine-messages \
    -i latest -m 1 persistent://public/default/my-topic

Get message ID​

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

pulsar-admin topics get-message-id \
    persistent://public/default/my-topic \
    -d 2021-06-28T19:01:17Z

Skip messages​

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

pulsar-admin topics skip \
    --count 10 --subscription my-subscription \
    persistent://test-tenant/ns1/tp1

Skip all messages​

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

pulsar-admin topics clear-backlog \
    --subscription my-subscription \
    persistent://test-tenant/ns1/tp1

Reset cursor​

You can reset a subscription cursor position back to the position which is recorded X seconds (or with other time units, e.g 100m, 3h, 2d, 5w) before. It essentially calculates the time and position of the cursor at X seconds before and resets it at that position. You can reset the cursor in the following ways.

pulsar-admin topics reset-cursor \
    --subscription my-subscription --time 10 \
    persistent://test-tenant/ns1/tp1

Look up topic's owner broker​

You can locate the owner broker of the given topic in the following ways.

pulsar-admin topics lookup persistent://test-tenant/ns1/tp1

Example output:

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

Look up partitioned topic's owner broker​

You can locate the owner broker of the given partitioned topic in the following ways.

pulsar-admin topics partitioned-lookup persistent://test-tenant/ns1/my-topic

Example output:

"persistent://test-tenant/ns1/my-topic-partition-0   pulsar://localhost:6650"
"persistent://test-tenant/ns1/my-topic-partition-1   pulsar://localhost:6650"
"persistent://test-tenant/ns1/my-topic-partition-2   pulsar://localhost:6650"
"persistent://test-tenant/ns1/my-topic-partition-3   pulsar://localhost:6650"

Lookup the partitioned topics sorted by broker URL

pulsar-admin topics partitioned-lookup \
    persistent://test-tenant/ns1/my-topic --sort-by-broker

Example output:

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 get the range of the bundle that the given topic belongs to in the following ways.

pulsar-admin topics bundle-range persistent://test-tenant/ns1/tp1

Example output:

"0x00000000_0xffffffff"

Get subscriptions​

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

pulsar-admin topics subscriptions persistent://test-tenant/ns1/tp1

Example output:

my-subscription

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 topics last-message-id topic-name

Example output:

{
  "ledgerId" : 97,
  "entryId" : 9,
  "partitionIndex" : -1
}

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 topics get-backlog-size \
  -m 1:1 \
  persistent://test-tenant/ns1/tp1-partition-0

Configure deduplication snapshot interval​

Get deduplication snapshot interval​

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

pulsar-admin topics get-deduplication-snapshot-interval my-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 topics set-deduplication-snapshot-interval my-topic -i 1000

Remove deduplication snapshot interval​

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

pulsar-admin topics remove-deduplication-snapshot-interval my-topic

Close inactive topics without deleting data​

Pulsar can close inactive topics to release broker memory while preserving their persistent data and metadata. The next producer or consumer connection reloads the topic. This is useful for large fleets of infrequently used topics: unloaded topics no longer contribute their per-topic metric series or hold their managed-ledger cache in memory.

To enable this mode, configure each broker in conf/broker.conf:

brokerDeleteInactiveTopicsEnabled=false
brokerCloseInactiveTopicsEnabled=true
brokerDeleteInactiveTopicsMode=delete_when_no_subscriptions
brokerDeleteInactiveTopicsFrequencySeconds=60
brokerDeleteInactiveTopicsMaxInactiveDurationSeconds=300

This example checks every 60 seconds and closes eligible topics after at least 300 seconds of inactivity. The close feature defaults to false. Its inactivity check uses the existing inactive-topic policy: in delete_when_no_subscriptions mode, a topic must have no subscriptions or active producers. A disconnected durable subscription still counts as a subscription; an empty backlog alone does not make the topic eligible.

The broker refuses to start if both inactive-topic deletion and closing are enabled, or if closing is combined with the broker mode delete_when_subscriptions_caught_up. Apply a consistent configuration before restarting brokers. While close mode is enabled, it takes precedence over namespace- or topic-level deleteWhileInactive settings. For persistent topics, a namespace or topic policy that overrides the mode to delete_when_subscriptions_caught_up causes the close check to skip that topic.

Closing does not delete the topic's BookKeeper data, schema, policies, or partitioned-topic metadata. Retention settings do not prevent closing because the operation preserves the data. This mode does not turn non-persistent topics into durable storage. Scalable-topic backing segments are excluded from this inactive-topic cleanup; the scalable-topic controller manages their lifecycle.

Observe the broker log message Topic closed successfully due to inactivity and the loaded-topic count when evaluating the feature. Allow for topic reload latency when an idle workload resumes.

Configure inactive topic policies​

Get inactive topic policies​

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

pulsar-admin topics get-inactive-topic-policies my-topic

Set inactive topic policies​

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

pulsar-admin topics set-inactive-topic-policies my-topic

Remove inactive topic policies​

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

pulsar-admin topics remove-inactive-topic-policies my-topic

Configure offload policies​

Get offload policies​

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

pulsar-admin topics get-offload-policies my-topic

Set offload policies​

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

pulsar-admin topics set-offload-policies my-topic

Remove offload policies​

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

pulsar-admin topics remove-offload-policies my-topic

Manage non-partitioned topics​

You can use Pulsar admin API to create, delete and check the 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.

Inactive-topic deletion is enabled by default and checked every 60 seconds. Eligibility depends on the effective inactive-topic policy, producer/subscription activity, inactivity duration, and retention settings; creation time alone does not determine deletion. Set brokerDeleteInactiveTopicsEnabled=false to disable the broker default, or use close-on-inactivity to release broker resources while preserving data.

For more information about the two parameters, see here.

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

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

pulsar-admin topics create \
    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.

Delete​

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

pulsar-admin topics delete \
    persistent://my-tenant/my-namespace/my-topic

List​

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

pulsar-admin topics list tenant/namespace

Example output:

persistent://tenant/namespace/topic1
persistent://tenant/namespace/topic2

Stats​

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

pulsar-admin topics stats \
    persistent://test-tenant/namespace/topic \
    --get-precise-backlog

The following is an example. For the description of topic stats, see Pulsar statistics.

{
  "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": {}
}

Internal stats​

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

{
  "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"
    }
  }
}

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

pulsar-admin topics stats-internal \
    persistent://test-tenant/namespace/topic

Manage partitioned topics​

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

Create​

When creating a new partitioned topic, you need to provide a name and the number of partitions for the topic.

note

Partitions follow the effective inactive-topic policy. An empty backlog alone does not make a partition eligible for deletion in the default delete_when_no_subscriptions mode. Partitioned-topic metadata is preserved by default (brokerDeleteInactivePartitionedTopicMetadataEnabled=false). See close-on-inactivity for an alternative that unloads topics without deleting their data.

For more information about the two parameters, see here.

You can create partitioned topics in the following ways.

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.

pulsar-admin topics create-partitioned-topic \
    persistent://my-tenant/my-namespace/my-topic \
    --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.

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.

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

pulsar-admin topics create-missed-partitions \
    persistent://my-tenant/my-namespace/my-topic

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.

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

pulsar-admin topics get-partitioned-topic-metadata \
    persistent://my-tenant/my-namespace/my-topic

Example output:

{
  "partitions" : 4,
  "deleted" : false
}

Update​

You can update the number of partitions for an existing partitioned topic. However, you can only increase the number of partitions. 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.

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

pulsar-admin topics update-partitioned-topic \
    persistent://my-tenant/my-namespace/my-topic \
    --partitions 8

Delete​

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

pulsar-admin topics delete-partitioned-topic \
    persistent://my-tenant/my-namespace/my-topic

List​

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

pulsar-admin topics list-partitioned-topics tenant/namespace

Example output:

persistent://tenant/namespace/topic1
persistent://tenant/namespace/topic2

Stats​

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

pulsar-admin topics partitioned-stats \
    persistent://test-tenant/namespace/topic \
    --per-partition

The following is an example. For the description of each topic stats, see Pulsar statistics.

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

{
  "msgRateIn" : 999.992947159793,
  "msgThroughputIn" : 1070918.4635439808,
  "msgRateOut" : 0.0,
  "msgThroughputOut" : 0.0,
  "bytesInCounter" : 270318763,
  "msgInCounter" : 252489,
  "bytesOutCounter" : 0,
  "msgOutCounter" : 0,
  "averageMsgSize" : 1070.926056966454,
  "msgChunkPublished" : false,
  "storageSize" : 270316646,
  "backlogSize" : 200921133,
  "publishers" : [ {
    "msgRateIn" : 999.992947159793,
    "msgThroughputIn" : 1070918.4635439808,
    "averageMsgSize" : 1070.3333333333333,
    "chunkedMessageRate" : 0.0,
    "producerId" : 0
  } ],
  "subscriptions" : {
    "test" : {
      "msgRateOut" : 0.0,
      "msgThroughputOut" : 0.0,
      "bytesOutCounter" : 0,
      "msgOutCounter" : 0,
      "msgRateRedeliver" : 0.0,
      "chuckedMessageRate" : 0,
      "chunkedMessageRate" : 0,
      "msgBacklog" : 144318,
      "msgBacklogNoDelayed" : 144318,
      "blockedSubscriptionOnUnackedMsgs" : false,
      "msgDelayed" : 0,
      "unackedMessages" : 0,
      "msgRateExpired" : 0.0,
      "lastExpireTimestamp" : 0,
      "lastConsumedFlowTimestamp" : 0,
      "lastConsumedTimestamp" : 0,
      "lastAckedTimestamp" : 0,
      "consumers" : [ ],
      "isDurable" : true,
      "isReplicated" : false
    }
  },
  "replication" : { },
  "metadata" : {
    "partitions" : 3
  },
  "partitions" : { }
}

Internal stats​

You can check the detailed statistics of a partitioned topic. The following is an example. For the description of each internal topic stats, see Pulsar statistics.

{
  "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"
    }
  }
}

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

pulsar-admin topics partitioned-stats-internal \
    persistent://test-tenant/namespace/topic

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 topics create-subscription \
    --subscription my-subscription \
    persistent://test-tenant/ns1/tp1

Get subscription​

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

pulsar-admin topics subscriptions persistent://test-tenant/ns1/tp1

Example output:

my-subscription

Unsubscribe subscription​

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

pulsar-admin topics unsubscribe \
    --subscription my-subscription \
    persistent://test-tenant/ns1/tp1

Shadow topics​

Shadow topics provide separate subscriptions to a persistent source topic while sharing its underlying message storage. They are not independent data copies or backups. Applications publish to the source topic and consume from a shadow topic.

Shadow-topic creation, loading, and replication are disabled by default. To use them, set the following on every broker that may own source or shadow topics and restart those brokers:

enableShadowTopics=true

For an upgrade of an existing shadow-topic deployment, apply this setting to the replacement brokers before rolling them. Leaving the default false prevents shadow topics from loading and source topics from starting shadow replication. This setting requires a broker restart; it is not a dynamic configuration option.

For example, after creating a persistent source topic, create and attach a shadow topic:

pulsar-admin topics create-shadow-topic persistent://public/default/orders-shadow \
  --source persistent://public/default/orders
pulsar-admin topics set-shadow-topics persistent://public/default/orders \
  --topics persistent://public/default/orders-shadow

set-shadow-topics replaces the source topic's shadow-topic list. Include all required shadow topics in the comma-separated --topics value when updating an existing configuration.

Inspect the relationship with:

pulsar-admin topics get-shadow-topics persistent://public/default/orders
pulsar-admin topics get-shadow-source persistent://public/default/orders-shadow

To remove the source's shadow replication configuration, use pulsar-admin topics remove-shadow-topics persistent://public/default/orders. This removes the association; it does not delete the shadow topic.

Was this helpful?
💡 Suggest changes🛟 Get support