Managing partitioned topics

You can use Pulsar's admin API to create and manage partitioned topics.

In all of the instructions and commands below, the topic name structure is:

persistent://tenant/namespace/topic

Partitioned topics resources

Create

Partitioned topics in Pulsar must be explicitly created. When creating a new partitioned topic youneed to provide a name for the topic as well as the desired number of partitions.

pulsar-admin

You can create partitioned topics using the create-partitioned-topiccommand and specifying the topic name as an argument and the number of partitions using the -p or —partitions flag.Here's an example:

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

REST API

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

Java

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

Nonpartitioned topics resources

Create

Nonpartitioned topics in Pulsar must be explicitly created if allowAutoTopicCreation or createIfMissing is disabled.When creating a non-partitioned topic, you need to provide a topic name.

pulsar-admin

You can create non-partitioned topics using the createcommand and specifying the topic name as an argument. This is an example command:

$ bin/pulsar-admin topics create persistent://my-tenant/my-namespace/my-topic

REST API

PUTadmin/v2/persistent/:tenant/:namespace/:topic

Java

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

Get metadata

Partitioned topics have metadata associated with them that you can fetch as a JSON object.The following metadata fields are currently available:

pulsar-admin

You can see the number of partitions in a partitioned topic using theget-partitioned-topic-metadatasubcommand. Here's an example:

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

REST API

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

Java

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

Update

You can update the number of partitions on an existing partitioned topicif the topic is non-global. To update, the new number of partitions must be greaterthan the existing number.

Decrementing the number of partitions would deleting the topic, which is not supported in Pulsar.

Already created partitioned producers and consumers will automatically find the newly created partitions.

pulsar-admin

Partitioned topics can be updated using theupdate-partitioned-topic command.

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

REST API

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

Java

admin.persistentTopics().updatePartitionedTopic(persistentTopic, numPartitions);

Delete

pulsar-admin

Partitioned topics can be deleted using thedelete-partitioned-topic command, specifying the topic by name:

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

REST API

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

Java

admin.persistentTopics().delete(persistentTopic);

List

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

pulsar-admin

$ pulsar-admin topics list tenant/namespace
persistent://tenant/namespace/topic1
persistent://tenant/namespace/topic2

REST API

GET/admin/v2/persistent/:tenant/:namespace

Java

admin.persistentTopics().getList(namespace);

Stats

It shows current statistics of a given partitioned topic. Here's an example payload:

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

The following stats are available:

pulsar-admin

The stats for the partitioned topic and its connected producers and consumers can be fetched by using thepartitioned-stats command, specifying the topic by name:

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

REST API

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

Java

admin.persistentTopics().getStats(persistentTopic);

Internal stats

It shows detailed statistics of a topic.

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

pulsar-admin

The internal stats for the partitioned topic can be fetched by using thestats-internal command, specifying the topic by name:

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

REST API

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

Java

admin.persistentTopics().getInternalStats(persistentTopic);