Pulsar KeyValue Avro Message Decoder

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:

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

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

{
  "type": "record",
  "name": "OrderKey",
  "fields": [
    {"name": "tenantId", "type": "string"},
    {"name": "orderId", "type": "string"}
  ]
}

Pulsar value Avro schema:

{
  "type": "record",
  "name": "OrderValue",
  "fields": [
    {"name": "status", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "eventTimeMs", "type": "long"}
  ]
}

Pinot schema:

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

{
  "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.

Get Started

Ingestion

External Tables

Query Data

Manage Data

Observability

Visualize Data

Cluster Operations

Manage Security

Release Notes

Reference

Pulsar KeyValue Avro Message Decoder

Class

When to Use

Payload Layout

Configuration Example

Example Schemas

Decoder Behavior

Field Selection

Authentication Notes

Troubleshooting

​Class

​When to Use

​Payload Layout

​Configuration Example

​Example Schemas

​Decoder Behavior

​Field Selection

​Authentication Notes

​Troubleshooting

Class

When to Use

Payload Layout

Configuration Example

Example Schemas

Decoder Behavior

Field Selection

Authentication Notes

Troubleshooting