HTTP Endpoints

This pages describes the HTTP endpoints available in the WarpStream Kafka product.

HTTP Fetch APIs

Requires Agent version v755+.

The WarpStream Agent exposes HTTP/JSON endpoints for fetching records from Kafka topics without a native Kafka client. All endpoints are served on the Agent's HTTP port (8080 by default).

Authentication and Authorization

Authentication

All endpoints support optional HTTP Basic Auth. When the Agent is configured with SASL authentication, you must include your WarpStream credentials (the same ones used for Kafka SASL/PLAIN) as HTTP Basic Auth.

With curl, use the -u flag:

curl -u 'ccun_YOUR_USERNAME:ccp_YOUR_PASSWORD' ...

This is equivalent to setting the Authorization header manually with the base64-encoded username:password pair:

curl -H 'Authorization: Basic Y2N1bl9ZT1VSX1VTRVJOQU1FOmNjcF9ZT1VSX1BBU1NXT1JE' ...

If the Agent does not require authentication, these headers can be omitted.

Authorization (ACLs)

The HTTP fetch endpoints enforce the same Kafka ACLs as the native Kafka protocol. The ACL principal is derived from the Basic Auth username as User:<username>. For example, if you authenticate as ccun_abc123, the ACL principal will be User:ccun_abc123.

The endpoints require READ permission on each topic being fetched. If the authenticated user does not have READ access to a topic, the partition will be returned with a TOPIC_AUTHORIZATION_FAILED error code (for /v1/kafka/fetch) or a 400 error response (for the single-record endpoints).

Common Headers

Header	Description	Default
kafka-client-id	Identifies the client for logging and diagnostics	http-fetch-client / http-fetch-single-record-client

Endpoints

GET or POST /v1/kafka/fetch

Full-featured fetch endpoint that mirrors the Kafka Fetch protocol. Supports fetching from multiple topics and partitions in a single request.

Request Body (JSON)

Field	Type	Required	Description
topics	array	yes	List of topics to fetch from
topics[].topic	string	yes	Topic name
topics[].partitions	array	yes	List of partitions to fetch from
topics[].partitions[].partition	integer	yes	Partition index (≥ 0)
topics[].partitions[].fetch_offset	integer	yes	Offset to start fetching from (≥ 0)
topics[].partitions[].partition_max_bytes	integer	yes	Max bytes to fetch for this partition (> 0)
max_bytes	integer	no	Max bytes for the entire fetch response (≥ 0). Default: 0 (agent-managed)
max_records	integer	no	Max total records to return across all partitions (≥ 0). 0 means unlimited. Parsing stops early once the budget is reached.
isolation_level	string	no	"read_uncommitted" (default) or "read_committed"
long_poll	boolean	no	If true, the agent waits up to 30s for new data instead of returning immediately. Default: false

Example Request

curl -X GET \
  'http://localhost:8080/v1/kafka/fetch' \
  -H 'Content-Type: application/json' \
  -u 'username:password' \
  -d '{
    "max_bytes": 1048576,
    "max_records": 10,
    "topics": [
      {
        "topic": "my-topic",
        "partitions": [
          {
            "partition": 0,
            "fetch_offset": 0,
            "partition_max_bytes": 1048576
          }
        ]
      }
    ]
  }'

Success Response (200 OK)

{
  "throttle_time_ms": 0,
  "topics": [
    {
      "topic": "my-topic",
      "partitions": [
        {
          "partition": 0,
          "error_code": "NONE",
          "high_watermark": 150,
          "last_stable_offset": 150,
          "log_start_offset": 0,
          "records": [
            {
              "offset": 0,
              "timestamp": 1707744000000,
              "key": "dGVzdC1rZXk=",
              "value": "dGVzdC12YWx1ZQ==",
              "headers": [
                {
                  "key": "header-name",
                  "value": "aGVhZGVyLXZhbHVl"
                }
              ]
            }
          ]
        }
      ]
    }
  ]
}

Note: key, value, and header value fields are base64-encoded byte arrays. A JSON null indicates a nil key/value.

Error Response (400 Bad Request / 500 Internal Server Error)

{
  "code": "INVALID_REQUEST",
  "message": "invalid request: topics must not be empty"
}

---

GET /v1/kafka/fetch_single_record

Convenience endpoint for fetching a single record at a specific offset using query parameters.

Query Parameters

Parameter	Type	Required	Description
topic	string	yes	Topic name
partition	integer	yes	Partition index (≥ 0)
offset	integer	yes	Exact offset to fetch (≥ 0)

Example Request

curl -X GET \
  'http://localhost:8080/v1/kafka/fetch_single_record?topic=my-topic&partition=0&offset=42' \
  -u 'username:password'

Success Response (200 OK)

Returns the record directly as a JSON object (not wrapped in the topic/partition structure):

{
  "offset": 42,
  "timestamp": 1707744000000,
  "key": "dGVzdC1rZXk=",
  "value": "dGVzdC12YWx1ZQ==",
  "headers": []
}

Not Found Response (404 Not Found)

Returned when no record exists at the requested offset (e.g., the offset is beyond the high watermark or below the low watermark):

{
  "code": "OFFSET_OUT_OF_RANGE",
  "message": "The requested offset is not within the range of offsets maintained by the server.: no record found at topic=my-topic partition=0 offset=999999"
}

Error Response (400 Bad Request)

{
  "code": "INVALID_REQUEST",
  "message": "missing required query parameter: topic"
}

---

GET /v1/kafka/topics/{topic}/partitions/{partition}/records/{offset}

REST-style convenience endpoint for fetching a single record at a specific offset using path parameters. Functionally identical to /v1/kafka/fetch_single_record.

Path Parameters

Parameter	Type	Required	Description
topic	string	yes	Topic name
partition	integer	yes	Partition index (≥ 0)
offset	integer	yes	Exact offset to fetch (≥ 0)

Example Request

curl -X GET \
  'http://localhost:8080/v1/kafka/topics/my-topic/partitions/0/records/42' \
  -u 'username:password'

Responses

Identical to /v1/kafka/fetch_single_record:

• 200 OK — The record as a JSON object.

• 404 Not Found — No record at the requested offset (OFFSET_OUT_OF_RANGE).

• 400 Bad Request — Invalid path parameters.

---

Error Codes

The code field in error responses maps to Kafka protocol error codes:

HTTP Status	Code	Meaning
400	INVALID_REQUEST	Malformed request (bad JSON, missing fields, invalid values)
401	SASL_AUTHENTICATION_FAILED	Missing or invalid credentials
403	TOPIC_AUTHORIZATION_FAILED	ACL denied read access to the topic
404	OFFSET_OUT_OF_RANGE	No record exists at the requested offset
500	KAFKA_STORAGE_ERROR	Internal error fetching data