StreamNative Console supports a RESTful interface to Pulsar clusters. You can produce and consume messages without using the native Pulsar protocol or clients.

The Message Rest API supports both non-partitioned and partitioned topics as well as basic and Avro base struct schema.

Example use cases include:

  • Send data to Pulsar from any frontend application built in any language
  • Integrate Pulsar with existing automation tools
  • Ingest Pulsar data into corporate dashboards and monitoring systems
  • Provide instant access to data in motion for data scientist notebooks
  • Ingest messages into a stream processing framework that may not support Pulsar

​
Produce

Description:

Send a single message to the topic.

Method: POST

Path: /admin/rest/topics/v1/<tenant>/<namespace>/<topic>/message

Request headers(* means required):

  • Accept:application/json * Accept the response body in JSON format
  • Content-Type: application/octet-stream * Send the request body in binary format
  • X-Pulsar-Long-Schema-Version Schema version in long type
  • X-Pulsar-Sequence-Id Message sequence ID
  • X-Pulsar-Event-Time Message event time
  • X-Pulsar-Property-<key> Message property, key-value pair
  • X-Pulsar-Partition-Key Message key

Request body:

Binary data

Response headers:

  • Content-Type: application/json * Send the response body in JSON format

Response body:

String: message ID in string format

Status code:

  • 201: Send single message success
  • 400: Wrong request params
  • 401: Unauthorized
  • 404: Topic/Schema doesn’t exist
  • 500: Internal server error

Error:

JsonObject:

reason(string): error information

​
Consume

Description:

Consume a single message from the topic.

Method: POST

Path: /admin/rest/topics/v1/<tenant>/<namespace>/<topic>/<subscriptionName>/message

Request headers(* means required):

  • Accept:application/octet-stream * Accept the response body in JSON format
  • Content-Type: application/json * Send the request body in binary format

Request body:

  • JsonObject :
    • timeoutMillis (int): 3000

Response headers( * means required):

  • Content-Type: application/octet-stream * Send the response body in JSON format
  • X-Pulsar-Base64-Schema-Version Schema version in base64 encoded format
  • X-Pulsar-Long-Schema-Version Schema version in long type
  • X-Pulsar-Message-Id * Message ID in base64 encoded format
  • X-Pulsar-Message-String-Id * Message ID in string format
  • X-Pulsar-Sequence-Id Message sequence ID
  • X-Pulsar-Event-Time Message event time
  • X-Pulsar-Property-<key> Message property, key-value pair
  • X-Pulsar-Partition-Key Message key

Response body:

Binary data

Status code:

  • 200: Get single message success
  • 204: Get null message by timeout
  • 400: Wrong request params
  • 401: Unauthorized
  • 404: Topic doesn’t exist
  • 500: Internal server error

Error:

  • JsonObject:
    • reason(string): error information

​
Acknowledge

Description:

Acknowledge a single message from the topic.

Method: PUT

Path: /admin/rest/topics/v1/<tenant>/<namespace>/<topic>/<subscriptionName>/message

Request headers(* means required):

  • Accept: application/json * Accept the response body in JSON format
  • Content-Type: application/json * Send the request body in binary format

Request body:

String: message ID in base64 encoded format

Response body:

no content

Status code:

  • 204: Acknowledge success
  • 400: Wrong request params
  • 401: Unauthorized
  • 404: Topic doesn’t exist
  • 500: Internal server error

Error:

JsonObject:

reason(string): error information

​
Working with Schemas

pulsar-rest supports all schemas and is compatible with the Pulsar client in other languages. If you want to produce/consume a topic with schemas through pulsar-rest, see the following example.

Pulsar schema uses the Pulsar Admin API to create. see manage schemas to learn the schema API or see create schema for topics to learn topic schema management on the StreamNative Console.

​
Binary Data

Because the HTTP client needs to send binary data, use the cURL tool to encode the string Hi Pulsar into UTF-8 format bytes. If you need to send bytes in another format, specify the appropriate file. For more information see the cURL documentation.

​
Avro Base Struct Schema

Pulsar uses Avro Specification to declare the schema definition for AvroBaseStructSchema, which supports AvroSchema, JsonSchema, and ProtobufSchema.

​
Multi-version schema

For basic schema types like String and JSON, pulsar-rest will automatically take the schema of the topic and put it into the metadata of the message. The client in other languages will serialize/deserialize the data based on this metadata.

​
String Schema

Create topic schema

curl -X POST https://<endpoint>:<port>/admin/v2/schemas/public/default/pulsar-rest-string/schema \
	--header "Authorization: Bearer <Access Token>" \
 	--header "Accept: application/json" \
 	--header "Content-Type: application/json" \
 	--data-raw '{"schema":"","type":"STRING","properties":{}}'

Expected output:

{"version":{"version":0}}

Create a subscription on the topic

curl -X PUT https://<endpoint>:<port>/admin/v2/persistent/public/default/pulsar-rest-string/subscription/rest-sub \
	--header 'Authorization: Bearer <Access Token>' \
 	--header 'Accept: application/json' \
 	--header 'Content-Type: application/json'

Expected output:

# No content

Produce messages with String schema

curl -X POST https://<endpoint>:<port>/admin/rest/topics/v1/persistent/public/default/pulsar-rest-string/message \
	-H "Authorization: Bearer <Access Token>" \
 	-H "Accept: application/json" \
 	-H "Content-Type: application/octet-stream" \
 	--data-binary 'Hi, Pulsar'

Expected output:

#string format message id
10:0:-1:0

Consume messages with String schema

curl -X POST https://<endpoint>:<port>/admin/rest/topics/v1/persistent/public/default/pulsar-rest-string/rest-sub/message \
	--header 'Authorization: Bearer <Access Token>' \
 	--header 'Accept: application/octet-stream' \
 	--data-raw '{"timeoutMillis":3000}' \
 	--header 'Content-Type: application/json' -v

Expected output:

# Headers
X-Pulsar-Message-Id: CAoQACAAMAE=
X-Pulsar-Message-String-Id: 10:0:-1:0
X-Pulsar-Sequence-Id: 0

# Body
Hi, Pulsar

Acknowledge messages

curl -X PUT https://<endpoint>:<port>/admin/rest/topics/v1/persistent/public/default/pulsar-rest-string/rest-sub/message \
	--header 'Authorization: Bearer <Access Token>' \
 	--header 'Accept: application/json' \
 	--header 'Content-Type: application/json' \
  --data-raw 'CAoQACAAMAE'

Expected output:

# No content

Negative acknowledge messages

curl -X PUT https://<endpoint>:<port>/admin/rest/topics/v2/persistent/public/default/pulsar-rest-string/rest-sub/message \
	--header 'Authorization: Bearer <Access Token>' \
 	--header 'Accept: application/json' \
 	--header 'Content-Type: application/json' \
    --data-raw '{"encodedMessageId":"CAoQACAAMAE=","negativeAck":"true"}'

Expected output:

# No content

​
JSON Schema

Create topic schema

curl -X POST https://<endpoint>:<port>/admin/v2/schemas/public/default/pulsar-rest-json/schema \
	--header "Authorization: Bearer <Access Token>" \
 	--header "Accept: application/json" \
 	--header "Content-Type: application/json" \
 	--data-raw '{"schema":"{\"type\":\"record\",\"name\":\"Student\",\"fields\":[{\"name\":\"age\",\"type\":\"int\"},{\"name\":\"name\",\"type\":[\"null\",\"string\"]}]}","type":"JSON","properties":{}}'

Expected output:

{"version":{"version":0}}

Create a subscription on the topic

curl -X PUT https://<endpoint>:<port>/admin/v2/persistent/public/default/pulsar-rest-json/subscription/rest-sub \
	--header 'Authorization: Bearer <Access Token>' \
 	--header 'Accept: application/json' \
 	--header 'Content-Type: application/json'

Expected output:

# No content

Produce messages with JSON schema

curl -X POST https://<endpoint>:<port>/admin/rest/topics/v1/persistent/public/default/pulsar-rest-json/message \
	--header 'Authorization: Bearer <Access Token>' \
  --header 'Accept: application/json' \
  --header 'Content-Type: application/octet-stream' \
 	--data-binary '{"name":"f8a42816-6183-4e3e-9f5d-51b36c17de5c","age":79}'

Expected output:

#string format message id
10:0:-1:0

Consume messages with JSON schema

curl -X POST https://<endpoint>:<port>/admin/rest/topics/v1/persistent/public/default/pulsar-rest-json/rest-sub/message \
	--header 'Authorization: Bearer <Access Token>' \
  --header 'Accept: application/octet-stream' \
  --data-raw '{"timeoutMillis":3000}' \
  --header 'Content-Type: application/json' -v

Expected output:

# Headers
< X-Pulsar-Base64-Schema-Version: AAAAAAAAAAA=
< X-Pulsar-Long-Schema-Version: 0
< X-Pulsar-Message-Id: CIUCEAAgADAB
< X-Pulsar-Message-String-Id: 261:0:-1:0
< X-Pulsar-Sequence-Id: 0

# Body
{"name":"f8a42816-6183-4e3e-9f5d-51b36c17de5c","age":79}

Acknowledge messages

curl -X PUT https://<endpoint>:<port>/admin/rest/topics/v1/persistent/public/default/pulsar-rest-json/rest-sub/message \
	--header "Authorization: Bearer <Access Token>" \
	--header "Accept: application/json" \
	--header "Content-Type: application/json" \
	--data-raw 'CIUCEAAgADAB'

Expected output:

# No content

​
Avro Schema

Create topic schema

curl -X POST https://<endpoint>:<port>/admin/v2/schemas/public/default/pulsar-rest-avro/schema \
	--header "Authorization: Bearer <Access Token>" \
 	--header "Accept: application/json" \
 	--header "Content-Type: application/json" \
 	--data-raw '{"schema":"{\"type\":\"record\",\"name\":\"Student\",\"fields\":[{\"name\":\"age\",\"type\":\"int\"},{\"name\":\"name\",\"type\":[\"null\",\"string\"]}]}","type":"AVRO","properties":{}}'

Expected output:

{"version":{"version":0}}

Create a subscription on the topic

curl -X PUT https://<endpoint>:<port>/admin/v2/persistent/public/default/pulsar-rest-avro/subscription/rest-sub \
	--header 'Authorization: Bearer <Access Token>' \
 	--header 'Accept: application/json' \
 	--header 'Content-Type: application/json'

Expected output:

# No content

Prepare Avro binary data

Create ~/test-avro-student.avsc

{
  "type": "record",
  "name": "Student",
  "fields": [
    {
      "name": "age",
      "type": "int"
    },
    {
      "name": "name",
      "type": ["null", "string"]
    }
  ]
}

Create ~/test-avro-student.json

{ "age": 8, "name": { "string": "1ffc0cee-a3d9-42f2-b96d-efa09f54e782" } }

Renders a JSON-encoded Avro datum as binary.

avro-tools jsontofrag --schema-file ~/test-avro-student.avsc   ~/test-avro-student.json > ~/test-avro-student.bin

If you don’t have avro-tools installed, see the Avro product page.

Produce messages with Avro schema

When you want to send a message to a topic with a multi-version schema, put the version of the schema in the header.

X-Pulsar-Long-Schema-Version Schema version in long type

cat ~/test-avro-student.bin | \
curl -X POST https://<endpoint>:<port>/admin/rest/topics/v1/persistent/public/default/pulsar-rest-avro/message \
	--header 'Authorization: Bearer <Access Token>' \
  --header 'Accept: application/json' \
  --header 'Content-Type: application/octet-stream' \
	--header 'X-Pulsar-Long-Schema-Version: 0' \
 	--data-binary @-

Expected output:

#string format message id
10:0:-1:0

Consume messages with Avro schema

curl -X POST https://<endpoint>:<port>/admin/rest/topics/v1/persistent/public/default/pulsar-rest-avro/rest-sub/message \
	--header 'Authorization: Bearer <Access Token>' \
  --header 'Accept: application/octet-stream' \
  --header 'Content-Type: application/json' \
	--dump-header ~/test-avro-student-recv-header.txt \
 	--data-raw '{"timeoutMillis":3000}' \
  --output ~/test-avro-student-recv.bin

Verify the consumed message

avro-tools fragtojson --schema-file ~/test-avro-student.avsc ~/test-avro-student-recv.bin

Expected output:

{
  "age" : 8,
  "name" : {
    "string" : "1ffc0cee-a3d9-42f2-b96d-efa09f54e782"
  }
}

Acknowledge messages

curl -X PUT https://<endpoint>:<port>/admin/rest/topics/v1/persistent/public/default/pulsar-rest-avro/rest-sub/message \
	--header "Authorization: Bearer <Access Token>" \
	--header "Accept: application/json" \
	--header "Content-Type: application/json" \
	--data-raw 'CIUCEAAgADAB'

Expected output:

# no content

​
Protobuf Schema

Create topic schema

curl -X POST https://<endpoint>:<port>/admin/v2/schemas/public/default/pulsar-rest-proto/schema \
	--header "Authorization: Bearer <Access Token>" \
 	--header "Accept: application/json" \
 	--header "Content-Type: application/json" \
 	--data-raw '{"schema":"{\"type\":\"record\",\"name\":\"TestMessage\",\"fields\":[{\"name\":\"stringField\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"},\"default\":\"\"},{\"name\":\"doubleField\",\"type\":\"double\",\"default\":0},{\"name\":\"intField\",\"type\":\"int\",\"default\":0}]}","type":"PROTOBUF","properties":{}}'

Expected output:

{"version":{"version":0}}

Create a subscription on the topic

curl -X PUT https://<endpoint>:<port>/admin/v2/persistent/public/default/pulsar-rest-proto/subscription/rest-sub \
	--header 'Authorization: Bearer <Access Token>' \
 	--header 'Accept: application/json' \
 	--header 'Content-Type: application/json'

Expected output:

# No content

Prepare protobuf data

Create ~/test-proto-test.proto

syntax = "proto3";

message TestMessage {
    string stringField = 1;
    double doubleField = 2;
    int32 intField = 3;
}

Create ~/test-proto-message.txt

stringField:"string filed"
doubleField: 3.14
intField: 8

Read a text-format message of the given type from standard input and write it in binary to standard output.

cat ~/test-proto-message.txt |  protoc --proto_path=<your user directory, ~> --encode='TestMessage' ~/test-proto-test.proto > ~/test-proto-message.bin

Produce messages with Protobuf schema

When you want to send a message to a topic with a multi-version schema, put the version of the schema in the header.

X-Pulsar-Long-Schema-Version Schema version in long type

cat ~/test-proto-message.bin | \
curl -X POST https://<endpoint>:<port>/admin/rest/topics/v1/persistent/public/default/pulsar-rest-proto/message \
	--header 'Authorization: Bearer <Access Token>' \
  --header 'Accept: application/json' \
  --header 'Content-Type: application/octet-stream' \
	--header 'X-Pulsar-Long-Schema-Version: 0' \
 	--data-binary '@-'

Expected output:

#string format message id
10:0:-1:0

Consume messages with Protobuf schema

curl -X POST https://<endpoint>:<port>/admin/rest/topics/v1/persistent/public/default/pulsar-rest-proto/rest-sub/message \
	--header 'Authorization: Bearer <Access Token>' \
  --header 'Accept: application/octet-stream' \
  --header 'Content-Type: application/json' \
	--dump-header ~/test-proto-message-recv-header.txt \
 	--data-raw '{"timeoutMillis":3000}' \
  --output ~/test-proto-message-recv.bin

Verify the consumed data

cat ~/test-proto-message-recv.bin |  protoc --proto_path=<your user directory> --decode='TestMessage' ~/test-proto-test.proto > ~/test-proto-message-recv.txt && cat ~/test-proto-message-recv.txt

Expected output:

stringField: "string filed"
doubleField: 3.14
intField: 8

Acknowledge messages

curl -X PUT https://<endpoint>:<port>/admin/rest/topics/v1/persistent/public/default/pulsar-rest-ptoto/rest-sub/message \
	--header "Authorization: Bearer <Access Token>" \
	--header "Accept: application/json" \
	--header "Content-Type: application/json" \
	--data-raw 'CIUCEAAgADAB'

Expected output:

# No content