Topics API - Apache Pulsar

Overview

Topics are the channels through which messages are sent and received in Pulsar. The Topics API provides comprehensive management capabilities for both persistent and non-persistent topics, including partitioning, subscriptions, and topic-level policies. Base Path: /admin/v2/persistent (for persistent topics)

List Topics

Get all topics in a namespace.

GET /admin/v2/persistent/{tenant}/{namespace}

Path Parameters

tenant

string

required

The tenant name

namespace

string

required

The namespace name

Query Parameters

bundle

string

Specific bundle to query

includeSystemTopic

boolean

default:"false"

Include system topics in the list

Example

curl -X GET http://localhost:8080/admin/v2/persistent/my-tenant/my-namespace \
  -H "Authorization: Bearer <token>"

Response Example

[
  "persistent://my-tenant/my-namespace/my-topic-1",
  "persistent://my-tenant/my-namespace/my-topic-2",
  "persistent://my-tenant/my-namespace/events"
]

List Partitioned Topics

Get all partitioned topics in a namespace.

GET /admin/v2/persistent/{tenant}/{namespace}/partitioned

Example

curl -X GET http://localhost:8080/admin/v2/persistent/my-tenant/my-namespace/partitioned \
  -H "Authorization: Bearer <token>"

Create Non-Partitioned Topic

Create a new non-partitioned topic.

PUT /admin/v2/persistent/{tenant}/{namespace}/{topic}

Path Parameters

tenant

string

required

The tenant name

namespace

string

required

The namespace name

topic

string

required

The topic name

Query Parameters

authoritative

boolean

default:"false"

Whether this broker is authoritative

Request Body

properties

object

Key-value pairs for topic metadata

Example

curl -X PUT http://localhost:8080/admin/v2/persistent/my-tenant/my-namespace/my-topic \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{
    "app": "my-app",
    "owner": "team-a"
  }'

Response Codes

Code	Description
204	Topic created successfully
307	Current broker doesn’t serve this namespace
401	Don’t have permission
403	Don’t have admin permission
404	Tenant or namespace doesn’t exist
409	Topic already exists

Create Partitioned Topic

Create a partitioned topic with a specified number of partitions.

PUT /admin/v2/persistent/{tenant}/{namespace}/{topic}/partitions

Path Parameters

tenant

string

required

The tenant name

namespace

string

required

The namespace name

topic

string

required

The topic name

Request Body

numPartitions

integer

required

Number of partitions (must be > 0)

Query Parameters

createLocalTopicOnly

boolean

default:"false"

Create topic only on local cluster

Example

curl -X PUT http://localhost:8080/admin/v2/persistent/my-tenant/my-namespace/my-partitioned-topic/partitions \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '4'

Response Codes

Code	Description
204	Partitioned topic created successfully
401	Don’t have permission
403	Don’t have admin permission
404	Tenant or namespace doesn’t exist
406	Number of partitions exceeds maximum
409	Topic already exists

Update Partitions

Increase the number of partitions for an existing partitioned topic.

POST /admin/v2/persistent/{tenant}/{namespace}/{topic}/partitions

Request Body

numPartitions

integer

required

New number of partitions (must be greater than current)

Example

curl -X POST http://localhost:8080/admin/v2/persistent/my-tenant/my-namespace/my-partitioned-topic/partitions \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '8'

You can only increase the number of partitions. Decreasing partitions requires deleting and recreating the topic.

Topic Permissions

Get Permissions

Retrieve permissions for a topic.

GET /admin/v2/persistent/{tenant}/{namespace}/{topic}/permissions

Grant Permission

Grant permissions to a role on a topic.

POST /admin/v2/persistent/{tenant}/{namespace}/{topic}/permissions/{role}

Path Parameters

role

string

required

The role name

Request Body

actions

array

required

Array of actions: “produce”, “consume”, “functions”

Example

curl -X POST http://localhost:8080/admin/v2/persistent/my-tenant/my-namespace/my-topic/permissions/my-role \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '["produce", "consume"]'

Revoke Permission

Revoke permissions from a role on a topic.

DELETE /admin/v2/persistent/{tenant}/{namespace}/{topic}/permissions/{role}

Example

curl -X DELETE http://localhost:8080/admin/v2/persistent/my-tenant/my-namespace/my-topic/permissions/my-role \
  -H "Authorization: Bearer <token>"

Topic Policies

Offload Policies

Configure automatic offloading to long-term storage.

Set Offload Policies

POST /admin/v2/persistent/{tenant}/{namespace}/{topic}/offloadPolicies

Request Body

offloadThresholdInBytes

integer

Size threshold for offloading

offloadThresholdInSeconds

integer

Age threshold for offloading

offloadDriver

string

Storage driver: “aws-s3”, “google-cloud-storage”, “filesystem”

Example

curl -X POST http://localhost:8080/admin/v2/persistent/my-tenant/my-namespace/my-topic/offloadPolicies \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{
    "offloadThresholdInBytes": 10737418240,
    "offloadDriver": "aws-s3",
    "bucket": "my-pulsar-bucket",
    "region": "us-west-2"
  }'

Get Offload Policies

GET /admin/v2/persistent/{tenant}/{namespace}/{topic}/offloadPolicies

Query Parameters

applied

boolean

default:"false"

Get effective policies (including inherited)

Delete Offload Policies

DELETE /admin/v2/persistent/{tenant}/{namespace}/{topic}/offloadPolicies

Max Unacked Messages

Control the maximum number of unacknowledged messages.

Set Max Unacked Messages on Consumer

POST /admin/v2/persistent/{tenant}/{namespace}/{topic}/maxUnackedMessagesOnConsumer

Request Body

maxUnackedNum

integer

required

Maximum number of unacked messages per consumer

Example

curl -X POST http://localhost:8080/admin/v2/persistent/my-tenant/my-namespace/my-topic/maxUnackedMessagesOnConsumer \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '50000'

Get Max Unacked Messages on Consumer

GET /admin/v2/persistent/{tenant}/{namespace}/{topic}/maxUnackedMessagesOnConsumer

Delete Max Unacked Messages Policy

DELETE /admin/v2/persistent/{tenant}/{namespace}/{topic}/maxUnackedMessagesOnConsumer

Delayed Delivery

Configure delayed message delivery.

Set Delayed Delivery Policies

POST /admin/v2/persistent/{tenant}/{namespace}/{topic}/delayedDelivery

Request Body

tickTime

integer

required

Tick time for delayed delivery in milliseconds

active

boolean

required

Whether delayed delivery is enabled

Example

curl -X POST http://localhost:8080/admin/v2/persistent/my-tenant/my-namespace/my-topic/delayedDelivery \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{
    "tickTime": 1000,
    "active": true
  }'

Get Delayed Delivery Policies

GET /admin/v2/persistent/{tenant}/{namespace}/{topic}/delayedDelivery

Inactive Topic Policies

Configure automatic cleanup of inactive topics.

Set Inactive Topic Policies

POST /admin/v2/persistent/{tenant}/{namespace}/{topic}/inactiveTopicPolicies

Request Body

inactiveTopicDeleteMode

string

required

Delete mode: “delete_when_no_subscriptions” or “delete_when_subscriptions_caught_up”

maxInactiveDurationSeconds

integer

required

Maximum inactive duration before deletion

Example

curl -X POST http://localhost:8080/admin/v2/persistent/my-tenant/my-namespace/my-topic/inactiveTopicPolicies \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{
    "inactiveTopicDeleteMode": "delete_when_no_subscriptions",
    "maxInactiveDurationSeconds": 3600
  }'

Best Practices

Topic Naming

Use descriptive names that reflect the data type
Follow a consistent naming convention
Avoid special characters except hyphens and underscores
Consider using hierarchical names with separators

Partitioning Strategy

Use partitioned topics for high-throughput scenarios
Start with 1-2 partitions per broker core
Monitor partition load and adjust as needed
Plan partition count growth carefully (can only increase)

Policy Configuration

Set offload policies for cost-effective storage
Configure inactive topic cleanup to free resources
Use max unacked messages to prevent consumer overload
Enable delayed delivery for scheduled message processing

Performance Optimization

Use appropriate partition counts for parallelism
Configure batching for better throughput
Set message TTL to prevent unbounded growth
Monitor topic statistics regularly

Namespaces API

Configure namespace-level policies

Schemas API

Manage topic schemas

REST API

CLI Tools

​Overview

​List Topics

​Path Parameters

​Query Parameters

​Example

​Response Example

​List Partitioned Topics

​Example

​Create Non-Partitioned Topic

​Path Parameters

​Query Parameters

​Request Body

​Example

​Response Codes

​Create Partitioned Topic

​Path Parameters

​Request Body

​Query Parameters

​Example

​Response Codes

​Update Partitions

​Request Body

​Example

​Topic Permissions

​Get Permissions

​Grant Permission

​Path Parameters

​Request Body

​Example

​Revoke Permission

​Example

​Topic Policies

​Offload Policies

​Set Offload Policies

Request Body

Example

​Get Offload Policies

Query Parameters

​Delete Offload Policies

​Max Unacked Messages

​Set Max Unacked Messages on Consumer

Request Body

Example

​Get Max Unacked Messages on Consumer

​Delete Max Unacked Messages Policy

​Delayed Delivery

​Set Delayed Delivery Policies

Request Body

Example

​Get Delayed Delivery Policies

​Inactive Topic Policies

​Set Inactive Topic Policies

Request Body

Example

​Best Practices

​Topic Naming

​Partitioning Strategy

​Policy Configuration

​Performance Optimization

​Related Resources

Namespaces API

Schemas API

Build docs developers (and LLMs) love

Overview

List Topics

Path Parameters

Query Parameters

Example

Response Example

List Partitioned Topics

Example

Create Non-Partitioned Topic

Path Parameters

Query Parameters

Request Body

Example

Response Codes

Create Partitioned Topic

Path Parameters

Request Body

Query Parameters

Example

Response Codes

Update Partitions

Request Body

Example

Topic Permissions

Get Permissions

Grant Permission

Path Parameters

Request Body

Example

Revoke Permission

Example

Topic Policies

Offload Policies

Set Offload Policies

Get Offload Policies

Delete Offload Policies

Max Unacked Messages

Set Max Unacked Messages on Consumer

Get Max Unacked Messages on Consumer

Delete Max Unacked Messages Policy

Delayed Delivery

Set Delayed Delivery Policies

Get Delayed Delivery Policies

Inactive Topic Policies

Set Inactive Topic Policies

Best Practices

Topic Naming

Partitioning Strategy

Policy Configuration

Performance Optimization

Related Resources