Fetching Active DLQ Records

You can use the /openintegration/v1.0/dlq/messages endpoint to fetch active DLQ records.

Request:

Method: GET

Endpoint:

https://<OI Host>:<OI
        Port>/openintegration/v1.0/dlq/messages

Description:This API fetches active DLQ records.
Supported query parameters:
- dlqTopic: Specifies the DLQ Kafka topic to read records from.
- sourceTopic: Used to filter DLQ records by the original source topic.
- Limit: Specifies the maximum number of DLQ records to return in the response. For example, if limit is set to 3, the response returns up to three records.
- Cursor: Fetches the next page of DLQ records. Use the nextCursor value from the previous response as the cursor value in the next request.

Response:

The response includes:

DLQ messages envelope: The /dlq/messages response contains envelope fields that describe the page of DLQ records returned by the API. Each DLQ envelope can include the following fields:

Field	Description
`dlqTopic`	DLQ Kafka topic that was read. If the request does not pass `dlqTopic`, EPS resolves the configured DLQ topic when possible.
`sourceTopic`	Source topic filter used for the request. An empty value means no source topic filter is used.
`limit`	Page size used by the response after validation.
`maxLimit`	Server-side maximum allowed limit from `DLQMessagesMaxLimit`. Requests above this value return HTTP `400`.
`count`	Number of records returned in this page after filtering deleted records and `sourceTopic`.
`hasMore`	Indicates whether another page is available for the same snapshot.
`nextCursor`	Opaque cursor to pass as the `cursor` query parameter on the next `/dlq/messages` call. Do not parse or modify this value. An empty `nextCursor` with `hasMore=false` means the current snapshot has been fully read.

The cursor captures DLQ end offsets at the start of the first page. This prevents a reader from continuously following newly written DLQ records while other consumers are still writing to the DLQ topic. New DLQ records written after the first page can be returned by a later fresh read without a cursor.

DLQ records: Each DLQ record can include the following fields:

Field	Description
`dlqRecordId`	Stable identifier used for delete. Current format: `dlqTopic\|consumerGroup\|sourceTopic\|sourcePartition\|sourceOffset`.
`dlqVersion`	DLQ payload schema version. Current value: `1.0`.
`topic`, `partition`, `offset`	Kafka coordinates of the DLQ topic record.
`dlqPartition`, `dlqOffset`	Explicit aliases for the DLQ record partition and offset. These values are the same as `partition` and `offset` in the current REST output.
`sourceTopic`, `sourcePartition`, `sourceOffset`, `sourceTimestamp`	Original failed source record coordinates and source Kafka timestamp.
`key`	Kafka key of the DLQ topic record. EPS uses the original key when present. Otherwise, it uses `dlqRecordId`.
`data`	Payload shown for the failed record. For new DLQ records, this is normally the original source value.
`originalKey`, `originalValue`, `originalHeaders`	Original Kafka key, value, and headers captured in the DLQ payload. `originalHeaders` can be an empty array when the source record has no headers.
`failedAt`	Epoch-millisecond timestamp when EPS built the DLQ payload.
`failure`, `failureCategory`	High-level classification, such as retriable or non-retriable.
`failureType`	More specific failure code, such as `OM_EXCEPTION`, `QUEUE_ERROR`, `BAD_JSON`, or `UNKNOWN_ERROR`.
`failureReason`	Detailed processing failure text used for diagnostics.
`exception`, `exceptionMessage`	Exception detail when available. These values are usually the same in the current REST output.
`attempts`	Number of processing attempts in the final processing window before the record was written to DLQ.

The request can return the following response status codes:

200: Returns a page envelope with records.
400: The limit or cursor value is invalid.
401: The request is unauthenticated.
503: The state cache is initializing.
500: An unexpected runtime failure occurred.

Examples:

Sample Request:

GET https://<OI Host>:<OI Port>/openintegration/v1.0/dlq/messages?dlqTopic=testevent1-consumer1-
dlq&sourceTopic=testevent1&limit=3

Response:

{
  "dlqTopic": "testevent1-consumer1-dlq",
  "sourceTopic": "testevent1",
  "limit": 3,
  "maxLimit": 500,
  "count": 3,
  "hasMore": true,
  "nextCursor": "eyJkbHFUb3BpYyI6InRlc3RldmVudDEtY29uc3VtZXIxLWRscSIsIm9mZnNldHMiOnsiMCI6IjMiLCIxIjoiMCIsIjIiOiIwIn0sImVuZE9mZnNldHMiOnsiMCI6IjUiLCIxIjoiMCIsIjIiOiIxMCJ9fQ",
  "records": [
    {
      "dlqRecordId": "testevent1-consumer1-dlq|consumer6|testevent1|1|0",
      "dlqVersion": "1.0",
      "dlqTopic": "testevent1-consumer1-dlq",
      "consumerGroup": "consumer6",
      "topic": "testevent1-consumer1-dlq"
    }
  ]
}

Use the nextCursor value as the cursor query parameter to fetch the next page of records.

Sample Request:

GET https://<OI Host>:<OI POrt>/openintegration/v1.0/dlq/messages?dlqTopic=testevent1-consumer1-
dlq&sourceTopic=testevent1&limit=10&cursor=eyJkbHFUb3BpYyI6InRlc3RldmVudDEtY29uc3VtZXIxLWRscSIsIm9mZnNldHMiOnsiMCI6IjUiLCIxIjoiMCIsIjIiOiI4In0sImVuZE9mZnNldHMiOnsiMCI6IjUiLCIxIjoiMCIsIjIiOiIxMCJ9fQ

Response:

{
    "dlqTopic": "testevent1-consumer1-dlq",
    "sourceTopic": "testevent1",
    "limit": 10,
    "maxLimit": 500,
    "count": 2,
    "hasMore": false,
    "nextCursor": "",
    "records": [
        {
            "dlqRecordId": "testevent1-consumer1-dlq|consumer6|testevent1|2|6",
            "dlqVersion": "1.0",
            "dlqTopic": "testevent1-consumer1-dlq",
            "consumerGroup": "consumer6",
            "topic": "testevent1-consumer1-dlq",
       .......
         
        },
        {
            "dlqRecordId": "testevent1-consumer1-dlq|consumer6|testevent1|0|6",
            "dlqVersion": "1.0",
            "dlqTopic": "testevent1-consumer1-dlq",
            "consumerGroup": "consumer6",
            "topic": "testevent1-consumer1-dlq",
          ......
        }
    ]
}

In this example, the request uses the cursor from the previous response. Although the request sets limit to 10, only two records remain, so the response returns two records. Because no more records are available, hasMore is false and nextCursor is empty.