AWS Glue Schema Registry Avro Decoder

​

Class

​

When to Use

• 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

​

How Registry Mode Works

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

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.

​

Configuration Properties

​

Glue Credential Resolution Order

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

"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\":[...]}"
}

"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"
}

"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": "<access-key>",
  "stream.kafka.decoder.prop.glue.secretAccessKey": "<secret-key>"
}

"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": "<external-id>"
}

​

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