> ## Documentation Index
> Fetch the complete documentation index at: https://docs.startree.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Pulsar KeyValue Avro Message Decoder

> Decode Apache Pulsar KeyValue messages whose key and value are Avro records.

Use the Pulsar KeyValue Avro decoder when a Pulsar topic uses a KeyValue Avro schema and the consumed payload contains both key bytes and value bytes in the format expected by StarTree Pinot.

### Class

`ai.startree.pinot.plugin.inputformat.pulsar.PulsarKeyValueAvroMessageDecoder`

## When to Use

Use this decoder when:

* Your realtime table consumes from Apache Pulsar.
* The Pulsar topic schema is a KeyValue schema.
* Both the key and value schemas are Avro schemas.
* The consumed message payload contains the key and value bytes stitched together.
* You want Pinot rows to include fields from the Pulsar message key and fields from the Pulsar message value.

Use a standard Avro decoder instead when each message contains a single Avro record and you do not need to decode a Pulsar KeyValue payload.

## Payload Layout

The decoder expects the byte array passed to Pinot to contain both key and value data in this exact layout:

| Offset          | Field        | Description                               |
| --------------- | ------------ | ----------------------------------------- |
| `0`             | key length   | 4-byte integer key length.                |
| `4`             | key bytes    | Avro binary payload for the key record.   |
| `4 + keyLength` | value length | 4-byte integer value length.              |
| `8 + keyLength` | value bytes  | Avro binary payload for the value record. |

Conceptually:

```text theme={null}
[4-byte key length][key Avro bytes][4-byte value length][value Avro bytes]
```

If your Pulsar client or connector exposes only the value bytes, this decoder will not have enough data to populate key fields. Make sure the stream layer passes the stitched key/value payload to Pinot.

## Configuration Example

Configure the decoder in the table `streamConfig`. The exact Pulsar consumer properties depend on your environment, but the decoder properties are:

| Property                                         | Required | Description                                          |
| ------------------------------------------------ | -------- | ---------------------------------------------------- |
| `stream.pulsar.decoder.class.name`               | Yes      | Decoder class name.                                  |
| `stream.pulsar.decoder.prop.pulsar.admin.url`    | Yes      | Pulsar Admin HTTP URL used to read the topic schema. |
| `stream.pulsar.decoder.prop.authenticationToken` | Yes      | Token used by the Pulsar Admin client.               |

Example:

```json theme={null}
{
  "tableName": "orders_REALTIME",
  "tableType": "REALTIME",
  "segmentsConfig": {
    "schemaName": "orders"
  },
  "streamConfig": {
    "streamType": "pulsar",
    "stream.pulsar.topic.name": "persistent://commerce/prod/orders",
    "stream.pulsar.bootstrap.servers": "pulsar://pulsar-broker:6650",
    "stream.pulsar.consumer.type": "lowlevel",
    "stream.pulsar.decoder.class.name": "ai.startree.pinot.plugin.inputformat.pulsar.PulsarKeyValueAvroMessageDecoder",
    "stream.pulsar.decoder.prop.pulsar.admin.url": "https://pulsar-admin.example.com",
    "stream.pulsar.decoder.prop.authenticationToken": "${PULSAR_ADMIN_TOKEN}"
  }
}
```

## Example Schemas

Pulsar key Avro schema:

```json theme={null}
{
  "type": "record",
  "name": "OrderKey",
  "fields": [
    {"name": "tenantId", "type": "string"},
    {"name": "orderId", "type": "string"}
  ]
}
```

Pulsar value Avro schema:

```json theme={null}
{
  "type": "record",
  "name": "OrderValue",
  "fields": [
    {"name": "status", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "eventTimeMs", "type": "long"}
  ]
}
```

Pinot schema:

```json theme={null}
{
  "schemaName": "orders",
  "dimensionFieldSpecs": [
    {"name": "tenantId", "dataType": "STRING"},
    {"name": "orderId", "dataType": "STRING"},
    {"name": "status", "dataType": "STRING"}
  ],
  "metricFieldSpecs": [
    {"name": "amount", "dataType": "DOUBLE"}
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "eventTimeMs",
      "dataType": "TIMESTAMP",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}
```

Decoded row:

```json theme={null}
{
  "tenantId": "tenant-a",
  "orderId": "O-1001",
  "status": "COMPLETE",
  "amount": 42.15,
  "eventTimeMs": 1741021200000
}
```

## Decoder Behavior

* Uses the Pulsar Admin API to fetch schema information for the configured topic.
* Decodes the Pulsar KeyValue schema into separate key and value Avro schemas.
* Extracts every field from the key schema.
* Extracts value fields that are requested by Pinot and are not key fields.
* Writes key fields and value fields into the same Pinot `GenericRow`.
* Throws an exception when the payload cannot be parsed as the expected key/value Avro byte layout.

## Field Selection

The decoder always extracts all fields from the key schema. For value fields, it honors Pinot's requested field set when Pinot provides one.

If a field exists in both key and value schemas, the key field wins because value extraction removes fields that are already part of the key schema. Avoid duplicate names unless the key value is the intended Pinot column value.

## Authentication Notes

`authenticationToken` is used to create a Pulsar token authentication client for the Pulsar Admin API. Treat this token as a secret:

* Store it in your deployment secret manager.
* Inject it into the table config or runtime configuration through your approved secret flow.
* Do not commit production tokens in table config files.

## Troubleshooting

| Symptom                                                                     | Cause                                                                                                        | Fix                                                                                                                                 |
| --------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------------------- |
| `Caught exception while decoding Avro message`                              | Payload does not match the expected key/value byte layout, or the schema does not match the bytes.           | Verify that the stream layer sends `[key length][key bytes][value length][value bytes]` and that the topic schema is KeyValue Avro. |
| `NullPointerException` or admin-client errors during decoder initialization | Missing `pulsar.admin.url`, missing `authenticationToken`, or the token cannot read topic schemas.           | Set both decoder properties and grant schema-read access to the token.                                                              |
| Key fields are missing                                                      | The consumed payload contains only value bytes.                                                              | Configure the Pulsar source path to pass both key and value bytes to Pinot.                                                         |
| A value field is not populated                                              | The Pinot schema or field selection did not request that field, or the field name collides with a key field. | Add the field to the Pinot schema and avoid duplicate key/value field names.                                                        |
| Schema parsing fails                                                        | The topic schema is not a KeyValue schema with Avro key and value schema definitions.                        | Use a Pulsar KeyValue Avro schema or choose a decoder that matches the topic format.                                                |