> For the complete documentation index, see [llms.txt](https://docs.warpstream.com/warpstream/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.warpstream.com/warpstream/kafka/reference/protocol-and-feature-support/http-endpoints.md).

# HTTP Endpoints

## 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:

```bash
curl -u 'ccun_YOUR_USERNAME:ccp_YOUR_PASSWORD' ...
```

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

```bash
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](#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

{% hint style="info" %}
Requires Agent version v755+.
{% endhint %}

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](#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**

```bash
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`)**

```json
{
  "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`)**

```json
{
  "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**

```bash
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):

```json
{
  "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):

```json
{
  "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`)**

```json
{
  "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**

```bash
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

{% hint style="info" %}
Requires Agent version v825+.
{% endhint %}

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](#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](https://docs.confluent.io/platform/current/kafka-rest/api.html) 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):

{% code overflow="wrap" %}

```bash
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="}]}'
```

{% endcode %}

JSON format (arbitrary JSON key and value):

{% code overflow="wrap" %}

```bash
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"}}]}'
```

{% endcode %}

**Success Response (`200 OK`)**

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

```json
{
  "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`:

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

**Request-Level Error Response**

```json
{
  "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**

{% code overflow="wrap" %}

```bash
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="}]}'
```

{% endcode %}

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**

```bash
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`:

```json
{
  "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)**

```json
{
  "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

{% hint style="info" %}
Request version v772+ of the Agent.
{% endhint %}

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](#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:

{% code overflow="wrap" %}

```bash
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"}]'
```

{% endcode %}

### 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:

{% code overflow="wrap" %}

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

{% endcode %}

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

{% code overflow="wrap" %}

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

{% endcode %}

### 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

{% hint style="info" %}
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.
{% endhint %}

#### `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**

{% code overflow="wrap" %}

```bash
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"}]'
```

{% endcode %}

**Success Response (`200 OK`)**

{% code overflow="wrap" %}

```
{}
```

{% endcode %}

#### `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**

{% code overflow="wrap" %}

```yaml
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"}]'
```

{% endcode %}

**Success Response (`200 OK`)**

{% code overflow="wrap" %}

```
{}
```

{% endcode %}

### Connectivity Probe

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

{% code overflow="wrap" %}

```
{}
```

{% endcode %}

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

{% code overflow="wrap" %}

```
{}
```

{% endcode %}

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 overflow="wrap" %}

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

{% endcode %}

### Notes

* These endpoints are intended for Datadog-compatible log ingestion. For general-purpose Kafka-over-HTTP produce, see the [HTTP Produce APIs](#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.


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://docs.warpstream.com/warpstream/kafka/reference/protocol-and-feature-support/http-endpoints.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.