HTTP Endpoints

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

Authentication and Authorization

These rules apply to all HTTP endpoints on this page. Each API section below only documents what is specific to it, such as the required ACL permission or a different credential header.

Authentication

All endpoints support optional HTTP Basic Auth carrying your WarpStream credentials (the same ones used for Kafka SASL/PLAIN). When the Agent is configured to require SASL authentication, credentials are mandatory; otherwise they can be omitted.

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

The Datadog log intake endpoints are the one exception: they accept the same credentials via the DD-API-KEY header instead of Basic Auth. See Datadog HTTP Log Intake for details.

mTLS authentication is not supported for any of the HTTP endpoints.

Authorization (ACLs)

All endpoints enforce the same Kafka ACLs as the native Kafka protocol:

• The ACL principal is derived from the authenticated username as User:<username>. For example, if you authenticate as ccun_abc123, the ACL principal will be User:ccun_abc123.
• If authentication is optional and no credentials are provided, the request is evaluated as the anonymous principal.
• The required permission depends on the API: the fetch endpoints require READ on each topic being fetched, while the produce and Datadog log intake endpoints require WRITE on each destination topic.

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

See Authentication and Authorization. Specific to the fetch endpoints:

• They 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

HTTP Produce APIs

Requires Agent version v825+.

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

Two API styles are available:

• Confluent REST Proxy v2-compatible endpoints (POST /topics/{topic} and POST /topics/{topic}/partitions/{partition}) — drop-in compatible with existing Confluent REST Proxy v2 tooling. Single topic per request, automatic partitioning.
• WarpStream-native endpoint (POST /v1/kafka/produce) — mirrors the Kafka Produce protocol. Supports multiple topics and partitions per request with explicit partition assignment, per-record timestamps, and headers.

Authentication and Authorization

See Authentication and Authorization. Specific to the produce endpoints:

• They require WRITE permission on each destination topic.
• If the authenticated principal does not have write access, the affected records fail with an authorization error: 403 TOPIC_AUTHORIZATION_FAILED for the native endpoint, or error code 40301 for the v2-compatible endpoints.

Common Headers

Header	Description	Default
kafka-client-id	Identifies the client for logging and diagnostics	http-produce-client / http-produce-v2-client
Content-Encoding	Optional request compression (gzip or zstd)	none

The request body limit is 64 MiB after decompression.

Confluent REST Proxy v2-Compatible Endpoints

These endpoints implement the Confluent REST Proxy v2 produce API for the binary and json embedded formats. Schema-based formats (avro, jsonschema, protobuf) are not supported and return 415 Unsupported Media Type.

Content Types

The Content-Type request header selects the embedded format:

Content-Type	Embedded format
application/vnd.kafka.binary.v2+json	binary: key and value are base64-encoded strings
application/vnd.kafka.json.v2+json	json: key and value are arbitrary JSON, stored as their JSON serialization
application/vnd.kafka.v2+json	Treated as binary
application/json	Treated as binary
application/octet-stream	Treated as binary

Responses always use Content-Type: application/vnd.kafka.v2+json.

POST /topics/{topic}

Produce records to a topic, optionally specifying keys or partitions per record.

Request Body (JSON)

Field	Type	Required	Description
records	array	yes	List of records to produce (must not be empty)
records[].key	object	no	Record key, formatted according to the embedded format, or null to omit
records[].value	object	no	Record value, formatted according to the embedded format
records[].partition	integer	no	Partition to store the record in
records[].headers	array	no	Record headers: {"key": string, "value": base64 string} (WarpStream extension)

Partitioning

Each record's partition is chosen with the same semantics as Kafka's default partitioner:

1. If records[].partition is set, it is used directly.
2. Otherwise, if the record has a key, the partition is murmur2(serialized key) % partition count — identical to the Kafka Java client and Confluent REST Proxy, so records produced over HTTP land on the same partitions as records produced with a Kafka client.
3. Otherwise (keyless records), all keyless records in the request share one sticky partition, which rotates round-robin across requests.

Example Requests

Binary format (base64 key and value):

curl -X POST 'http://localhost:8080/topics/my-topic' \
  -H 'Content-Type: application/vnd.kafka.binary.v2+json' \
  -u 'username:password' \
  -d '{"records": [{"key": "a2V5", "value": "dmFsdWU="}]}'

JSON format (arbitrary JSON key and value):

curl -X POST 'http://localhost:8080/topics/my-topic' \
  -H 'Content-Type: application/vnd.kafka.json.v2+json' \
  -u 'username:password' \
  -d '{"records": [{"value": {"name": "testUser"}}]}'

Success Response (200 OK)

offsets contains one entry per record, in the same order as the request:

{
  "offsets": [
    {
      "partition": 0,
      "offset": 42,
      "error_code": null,
      "error": null
    }
  ],
  "key_schema_id": null,
  "value_schema_id": null
}

If an individual record fails, its entry has null partition and offset and a non-null error_code and error:

{
  "offsets": [
    {
      "partition": null,
      "offset": null,
      "error_code": 50003,
      "error": "..."
    }
  ],
  "key_schema_id": null,
  "value_schema_id": null
}

Request-Level Error Response

{
  "error_code": 40401,
  "message": "topic: my-topic does not exist"
}

---

POST /topics/{topic}/partitions/{partition}

Produce records directly to a single partition. Identical to POST /topics/{topic} except that all records are written to the partition from the path, and per-record partition fields are ignored.

Path Parameters

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

Example Request

curl -X POST 'http://localhost:8080/topics/my-topic/partitions/0' \
  -H 'Content-Type: application/vnd.kafka.binary.v2+json' \
  -u 'username:password' \
  -d '{"records": [{"value": "dmFsdWU="}]}'

If the partition does not exist, the request fails with 404 and error code 40402.

---

v2 Error Codes

Error codes follow the Confluent REST Proxy v2 numeric scheme:

HTTP Status	error_code	Meaning
400	400	Serialization failure (e.g. invalid base64 in binary format)
401	40101	Missing or invalid credentials
403	40301	ACL denied write access to the topic
404	40401	Topic does not exist
404	40402	Partition does not exist
413	413	Request body exceeds 64 MiB after decompression, or record too large
415	415	Unsupported embedded format (e.g. avro, protobuf)
422	422	Request validation failure (e.g. empty records list)
500	50002	Non-retriable Kafka error
500	50003	Retriable Kafka error; the produce might succeed if retried

Per-record failures inside a 200 OK response use 50002, 50003, or 40301 in the offsets[].error_code field. If any record fails with an authorization error, the overall HTTP status is 403.

WarpStream-Native Endpoint

POST /v1/kafka/produce

Full-featured produce endpoint that mirrors the Kafka Produce protocol. Supports producing to multiple topics and partitions in a single request with explicit partition assignment.

Request Body (JSON)

Field	Type	Required	Description
topics	array	yes	List of topics to produce to
topics[].topic	string	yes	Topic name
topics[].partitions	array	yes	List of partitions to produce to
topics[].partitions[].partition	integer	yes	Partition index (≥ 0)
topics[].partitions[].records	array	yes	List of records for this partition
topics[].partitions[].records[].key	string	no	Base64-encoded record key, or null for no key
topics[].partitions[].records[].value	string	no	Base64-encoded record value, or null for a tombstone
topics[].partitions[].records[].headers	array	no	Record headers: {"key": string, "value": base64 string}
topics[].partitions[].records[].timestamp	integer	no	Record timestamp in milliseconds since the Unix epoch. If 0 or omitted, the Agent assigns the current time.
timeout_ms	integer	no	Produce timeout in milliseconds. Default: 10000

Example Request

curl -X POST \
  'http://localhost:8080/v1/kafka/produce' \
  -H 'Content-Type: application/json' \
  -u 'username:password' \
  -d '{
    "topics": [
      {
        "topic": "my-topic",
        "partitions": [
          {
            "partition": 0,
            "records": [
              {
                "key": "dGVzdC1rZXk=",
                "value": "dGVzdC12YWx1ZQ==",
                "headers": [
                  {
                    "key": "header-name",
                    "value": "aGVhZGVyLXZhbHVl"
                  }
                ]
              }
            ]
          }
        ]
      }
    ]
  }'

Success Response (200 OK)

The response mirrors the Kafka Produce response. Per-partition failures are reported in the body with a 200 status, so check each partition's error_code:

{
  "throttle_time_ms": 0,
  "topics": [
    {
      "topic": "my-topic",
      "partitions": [
        {
          "partition": 0,
          "error_code": "NONE",
          "base_offset": 42,
          "log_append_time": -1
        }
      ]
    }
  ]
}

base_offset is the offset assigned to the first record in the partition's batch; subsequent records occupy consecutive offsets.

Error Response (400 Bad Request / other statuses)

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

---

Native Endpoint Error Codes

The code field in error responses maps to Kafka protocol error codes, in the same style as the HTTP fetch endpoints:

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 write access to the topic
404	UNKNOWN_TOPIC_OR_PARTITION	Topic or partition does not exist
413	MESSAGE_TOO_LARGE	Request body exceeds 64 MiB after decompression
500	KAFKA_STORAGE_ERROR	Internal error producing data

Datadog HTTP Log Intake

Datadog Log Intake APIs

Request version v772+ of the Agent.

The WarpStream Agent exposes Datadog-compatible HTTP/JSON endpoints for accepting log batches from the Datadog Agent and producing them into Kafka topics. All endpoints are served on the Agent's HTTP port (8080 by default).

Authentication and Authorization

See Authentication and Authorization. Specific to the log intake endpoints:

• These endpoints use the DD-API-KEY header rather than HTTP Basic Auth. WarpStream interprets DD-API-KEY as Kafka SASL/PLAIN credentials encoded as $SASL_USERNAME:$SASL_PASSWORD.
• If DD-API-KEY is present but malformed and SASL auth is optional, it is ignored.
• If DD-API-KEY is missing or malformed and SASL auth is required, the request fails with 401 SASL_AUTHENTICATION_FAILED.
• They require WRITE permission on the destination topic. If the authenticated principal does not have write access to the topic, the request fails with 403 TOPIC_AUTHORIZATION_FAILED.

Example:

curl -X POST \'http://localhost:8080/dd/my-topic/api/v2/logs' \-H 'Content-Type: application/json' \-H 'DD-API-KEY: YOUR_USERNAME:YOUR_PASSWORD' \-d '[{"message":"hello from datadog","service":"my-service"}]'

Common Headers

Header	Description	Default
DD-API-KEY	WarpStream Kafka SASL/PLAIN credentials encoded as username:password	omitted
kafka-client-id	Identifies the client for logging and diagnostics	http-datadog-logs-client
Content-Type	Must be application/json	none
Content-Encoding	Optional request compression	none

Datadog Agent Configuration

When configuring the Datadog Agent, point logs_dd_url at the path prefix only. Do not include /api/v2/logs or /v1/input in the configured URL, because the Datadog Agent appends the intake suffix itself.

Example Datadog Agent configuration for the primary endpoint:

logs_enabled: true
logs_config:
    use_v2_api: true
    logs_dd_url: https://warpstream.example.com/dd/$TOPIC_NAME

If you want Datadog to use the compatibility endpoint instead, set:

logs_enabled: true
logs_config:
    use_v2_api: false
    logs_dd_url: https://warpstream.example.com/dd/$TOPIC_NAME

Common Behavior

These endpoints share the following behavior:

• The $TOPIC_NAME path parameter is the exact Kafka topic name to produce to.
• The topic must already exist.
• The topic name must be a valid Kafka topic name.
• The request body limit is 64 MiB after decompression.
• Each JSON object in the submitted array becomes one Kafka record.
• The JSON object is stored as the Kafka record value exactly as submitted.
• No Kafka key is set.
• No Datadog-specific fields are interpreted or transformed.
• All records from a single HTTP request are written to the same partition.
• Across requests, the Agent rotates partitions over time for rough byte-based balancing.

Raw Endpoint Specs

The raw endpoint specifications is not required to use this integration. Simply follow the instructions above to configure the Datadog Agent and you'll be good to go. That said, the endpoints are documented for posterity.

POST /dd/$TOPIC_NAME/api/v2/logs

Primary Datadog-compatible log intake endpoint.

This is the preferred route for Datadog Agents configured with use_v2_api: true.

Path Parameters

Parameter	Type	Required	Description
topic	string	yes	Destination Kafka topic name

Request Body (JSON)

The request body must be one of:

Shape	Description
[{...}, {...}]	A JSON array of log objects. Each array element becomes one Kafka record value.
{}	Connectivity probe. Accepted but does not produce any records.

The Agent treats each array element as an opaque JSON object. Common Datadog fields like message, service, ddsource, ddtags, hostname, status, and timestamp are preserved exactly as sent.

Example Request

curl -X POST \'http://localhost:8080/dd/my-topic/api/v2/logs' \-H 'Content-Type: application/json' \-H 'DD-API-KEY: YOUR_USERNAME:YOUR_PASSWORD' \-d '[{"message": "application started","service": "payments","ddsource": "kubernetes","ddtags": "env:staging,team:data"},{"message": "worker ready","service": "payments","ddsource": "kubernetes","ddtags": "env:staging,team:data"}]'

Success Response (200 OK)

{}

POST /dd/$TOPIC_NAME/v1/input

Compatibility alias for Datadog Agents configured with use_v2_api: false.

This endpoint has the same authentication, authorization, request-body, partitioning, and response semantics as /dd/$TOPIC_NAME/api/v2/logs.

Example Request

curl -X POST \'http://localhost:8080/dd/my-topic/v1/input' \-H 'Content-Type: application/json' \-H 'DD-API-KEY: YOUR_USERNAME:YOUR_PASSWORD' \-d '[{"message": "legacy intake example","service": "payments"}]'

Success Response (200 OK)

{}

Connectivity Probe

The Datadog Agent may send an empty JSON object to check HTTP connectivity:

{}

This is accepted and returns 200 OK with an empty JSON response body:

{}

No Kafka records are produced for connectivity probes.

Request Compression

These endpoints support compressed request bodies via Content-Encoding.

Supported encodings for Datadog-style traffic:

• gzip
• zstd

The 64 MiB request-size limit is enforced after decompression.

Error Codes

The code field in error responses maps either to Kafka protocol error codes or endpoint-specific validation errors.

HTTP Status	Code	Meaning
400	INVALID_REQUEST	Malformed JSON, unsupported content type, non-empty top-level object, trailing JSON tokens, or other request-shape validation failure
400	INVALID_TOPIC_EXCEPTION	Invalid topic name or topic does not exist
401	SASL_AUTHENTICATION_FAILED	Missing, malformed, or invalid DD-API-KEY when authentication is required
403	TOPIC_AUTHORIZATION_FAILED	ACL denied write access to the topic
413	PAYLOAD_TOO_LARGE	Request body exceeds 64 MiB after decompression
413	MESSAGE_TOO_LARGE	One or more produced records exceed the Agent's maximum record size
500	KAFKA_STORAGE_ERROR	Internal metadata lookup, authorization, or produce failure

Example Error Response

{"code": "INVALID_REQUEST","message": "payload must be either an empty JSON object or a JSON array"}

Notes

• These endpoints are intended for Datadog-compatible log ingestion. For general-purpose Kafka-over-HTTP produce, see the HTTP Produce APIs above.
• Partitioning is request-scoped: all records in one HTTP request go to one partition.
• Partition selection starts from a random partition per topic and rotates round-robin over time based on accumulated request bytes.
• If you need different routing, use a different $TOPIC_NAME path prefix for each Datadog sender configuration.