> ## 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.
# AWS Glue Schema Registry Avro Decoder
The Glue decoder is used for **real-time ingestion** in Pinot when consuming **Avro-encoded messages** from Kafka (including Amazon MSK) topics whose schemas are managed by the **AWS Glue Schema Registry**.
Unlike the [Confluent Avro decoder](/corecapabilities/ingestdata/adv-concepts/realtime/decoders/avro), Glue Schema Registry wraps each message with its own binary header rather than Confluent's magic-byte + schema-ID format, so a dedicated decoder is required.
### Class
`ai.startree.pinot.plugin.inputformat.glue.avro.GlueSchemaRegistryAvroMessageDecoder`
## When to Use
Use this decoder when:
* Kafka (or MSK) messages are serialized using **Avro** and wrapped with the **AWS Glue Schema Registry** wire format.
* Schemas are registered and versioned in **AWS Glue Schema Registry** rather than Confluent Schema Registry.
* The topic's producers may write with **different schema versions over time** (schema evolution) and messages from any version must decode correctly.
## Operating Modes
The decoder supports two mutually exclusive modes, selected automatically by whether the `schema` config property is set.
| Mode | Trigger | Behavior |
| ----------------- | ---------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Legacy mode** | `schema` is provided | Uses the inline schema you supply as both the writer and reader schema. No Glue API calls are made. This is the original decoder behavior. |
| **Registry mode** | `schema` is absent, `region` is provided | Resolves the actual writer schema per message from AWS Glue Schema Registry using the schema-version UUID embedded in the message header. Handles schema evolution transparently. |
Legacy mode requires the inline schema to **exactly match** the writer schema used to produce every message on the topic. If the topic's schema evolves (for example, a producer starts writing with 13 fields instead of 10), messages written with a different version than the inline schema fail to decode — this can surface as `Sampled: 0 Expected min: 1` from the table preview API, or as silent row loss during ingestion. Switch to registry mode to support schema evolution.
## How Registry Mode Works
Each Glue Schema Registry message is wrapped with an 18-byte header:
```
[1 byte header version] [1 byte compression] [16 bytes schema-version UUID] [Avro payload...]
```
* **Header version** must be `0x03` — the current AWS Glue Schema Registry wire format. Messages with any other version byte are skipped.
* **Compression** is either `0x00` (none) or `0x05` (ZLIB). Any other value is skipped. ZLIB-compressed payloads are inflated up to the `maxDecompressedBytes` limit before decoding; payloads that would exceed it are dropped and logged.
* **Schema-version UUID** identifies the exact writer schema used for that message.
For each message, the decoder:
1. **Resolves the writer schema per message** — extracts the UUID from the header and looks up the matching Avro schema from its in-memory cache, or fetches it from Glue on a cache miss.
2. **Pre-warms the cache at init**, if `registry.name` and `schemaName` are both set — all known schema versions for that Glue schema are paginated via `listSchemaVersions` and fetched via `getSchemaVersion` before the first message arrives, avoiding a Glue round-trip on cold start.
3. **Falls back on demand** for UUIDs not already cached (whether pre-warming was configured or not) — the schema is fetched from Glue and cached for subsequent messages with the same UUID.
4. **Does not cache failed fetches** — if a Glue API call fails (throttling, transient network error, permissions), the UUID is not cached as a failure, so the next message with that UUID retries the fetch rather than being permanently unable to decode.
5. **Bounds the cache with LRU eviction** — a single cache holds the writer schema, its Avro `DatumReader`, and a reusable `GenericData.Record` per schema-version UUID, bounded by `glue.maxCachedSchemas` (default `1000`). Because writer schema, reader, and reused record live in one entry, eviction is atomic — there's no risk of one map evicting a UUID while a companion map still holds stale state for it.
Each stream partition consumer creates its own decoder instance, and therefore its own Glue client and schema cache. `glue.maxCachedSchemas` bounds the cache **per partition consumer**, not per table — size it with the number of partitions in mind if a table has many.
## Configuration Properties
All properties are set via `stream.kafka.decoder.prop.*` (or the equivalent prefix for other stream types).
| Property | Required | Default | Description |
| ----------------------- | ------------------------------------- | ------------------- | -------------------------------------------------------------------------------------------------------- |
| `schema` | Mutually exclusive with registry mode | — | Inline Avro schema JSON. If present, the decoder runs in **legacy mode** and makes no Glue API calls. |
| `region` | Required for registry mode | — | AWS region for the Glue client, e.g. `ap-south-1`. Required when `schema` is absent. |
| `registry.name` | Optional | — | Glue registry name. Combine with `schemaName` to pre-warm the schema cache at init. |
| `schemaName` | Optional | — | Glue schema name. Combine with `registry.name` to pre-warm the schema cache at init. |
| `glue.endpoint` | Optional | — | Custom Glue API endpoint URL — for example, a VPC endpoint or a cross-region override. |
| `glue.accessKeyId` | Optional (paired) | — | Static AWS access key ID for the Glue client. Requires `glue.secretAccessKey`. |
| `glue.secretAccessKey` | Optional (paired) | — | Static AWS secret access key for the Glue client. Requires `glue.accessKeyId`. |
| `glue.roleArn` | Optional | — | IAM role ARN to assume for Glue API calls — useful for cross-account access or scoping down permissions. |
| `glue.roleSessionName` | Optional | `pinot-glue-sr` | Session name used when assuming `glue.roleArn`. |
| `glue.roleExternalId` | Optional | — | External ID required by the assumed role's trust policy, for cross-account setups. |
| `glue.maxCachedSchemas` | Optional | `1000` | Maximum writer schemas cached per decoder instance, with LRU eviction once exceeded. |
| `maxDecompressedBytes` | Optional | `16777216` (16 MiB) | Upper bound on ZLIB-decompressed payload size; larger payloads are dropped rather than fully inflated. |
### Glue Credential Resolution Order
The Glue client's credentials are resolved independently of your Kafka/MSK credentials (for example, MSK IAM auth configured via `sasl.jaas.config`) — they follow their own chain:
1. If `glue.accessKeyId` and `glue.secretAccessKey` are both set → static credentials.
2. If `glue.roleArn` is also set → assume that role via STS, using the static credentials from step 1 as the base identity if present, otherwise the default credential chain.
3. Otherwise → the AWS SDK v2 default credential chain (environment variables, system properties, instance profile, IRSA, etc.).
## Configuration Examples
Inline schema, no Glue API calls — identical to the decoder's original behavior:
```json theme={null}
"streamConfig": {
"streamType": "kafka",
"stream.kafka.topic.name": "orders_topic",
"stream.kafka.consumer.type": "lowlevel",
"stream.kafka.decoder.class.name": "ai.startree.pinot.plugin.inputformat.glue.avro.GlueSchemaRegistryAvroMessageDecoder",
"stream.kafka.broker.list": "broker:9092",
"stream.kafka.decoder.prop.schema": "{\"type\":\"record\",\"name\":\"Order\",\"fields\":[...]}"
}
```
Pinot's ambient IAM role or instance profile already has Glue read access, so no explicit credentials are needed. `registry.name` + `schemaName` pre-warm the cache at startup:
```json theme={null}
"streamConfig": {
"streamType": "kafka",
"stream.kafka.topic.name": "orders_topic",
"stream.kafka.consumer.type": "lowlevel",
"stream.kafka.decoder.class.name": "ai.startree.pinot.plugin.inputformat.glue.avro.GlueSchemaRegistryAvroMessageDecoder",
"stream.kafka.broker.list": "broker:9092",
"stream.kafka.decoder.prop.region": "ap-south-1",
"stream.kafka.decoder.prop.registry.name": "MySchemaRegistry",
"stream.kafka.decoder.prop.schemaName": "ORDER_EVENT"
}
```
Glue Schema Registry lives in a different AWS account, accessed with a dedicated IAM user's access keys:
```json theme={null}
"streamConfig": {
"streamType": "kafka",
"stream.kafka.topic.name": "orders_topic",
"stream.kafka.consumer.type": "lowlevel",
"stream.kafka.decoder.class.name": "ai.startree.pinot.plugin.inputformat.glue.avro.GlueSchemaRegistryAvroMessageDecoder",
"stream.kafka.broker.list": "broker:9092",
"stream.kafka.decoder.prop.region": "ap-south-1",
"stream.kafka.decoder.prop.registry.name": "MySchemaRegistry",
"stream.kafka.decoder.prop.schemaName": "ORDER_EVENT",
"stream.kafka.decoder.prop.glue.accessKeyId": "",
"stream.kafka.decoder.prop.glue.secretAccessKey": ""
}
```
Pinot assumes an IAM role in the target account that has Glue permissions:
```json theme={null}
"streamConfig": {
"streamType": "kafka",
"stream.kafka.topic.name": "orders_topic",
"stream.kafka.consumer.type": "lowlevel",
"stream.kafka.decoder.class.name": "ai.startree.pinot.plugin.inputformat.glue.avro.GlueSchemaRegistryAvroMessageDecoder",
"stream.kafka.broker.list": "broker:9092",
"stream.kafka.decoder.prop.region": "ap-south-1",
"stream.kafka.decoder.prop.registry.name": "MySchemaRegistry",
"stream.kafka.decoder.prop.schemaName": "ORDER_EVENT",
"stream.kafka.decoder.prop.glue.roleArn": "arn:aws:iam::123456789012:role/GlueSchemaRegistryReadOnly",
"stream.kafka.decoder.prop.glue.roleSessionName": "pinot-glue-sr",
"stream.kafka.decoder.prop.glue.roleExternalId": ""
}
```
## Decoder Behavior
* Converts Avro records into Pinot-compatible rows, dropping fields not present in the Pinot schema.
* Reuses the underlying Avro `GenericData.Record` across decode calls (one instance in legacy mode, one per cached schema-version UUID in registry mode) to reduce per-message allocations on high-throughput topics.
* Avoid nested records — flatten them before ingestion if needed.
* Use `nullHandlingEnabled` if fields might be missing or optional.
* Malformed or unrecognized payloads (bad header version, unsupported compression byte, truncated payload, decode failure) are logged and skipped rather than failing the consumer.
## Troubleshooting
This is the classic symptom of legacy mode's inline schema no longer matching every message on the topic — typically because the Glue schema evolved (fields added, types changed) after the inline `schema` was captured. Switch to registry mode: remove `schema` and set `region` (plus `registry.name` + `schemaName` if you want pre-warming), and each message decodes with its own correct writer schema.
Registry mode requires `region`. If you intended legacy mode instead, set `schema` to the inline Avro schema JSON.
`glue.accessKeyId` and `glue.secretAccessKey` must be set together — supplying only one throws at init. Either provide both, or omit both and rely on `glue.roleArn` or the default AWS credential chain.
Set `registry.name` and `schemaName` so all known schema versions are pre-warmed into the cache at decoder init, avoiding a per-partition Glue round-trip on the first message of each previously-unseen schema version.