You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
Commit 2f69b41
Browse filesBrowse the repository at this point in the historyBrowse files
Copy file name to clipboardExpand all lines: docs/utilities/kafka.md
+40Lines changed: 40 additions & 0 deletions
Display the source diff
Display the rich diff
Original file line number
Diff line number
Diff line change
@@ -29,6 +29,7 @@ flowchart LR
29
29
* Support for key and value deserialization
30
30
* Support for custom output serializers (e.g., dataclasses, Pydantic models)
31
31
* Support for ESM with and without Schema Registry integration
32
+
* Support for offline Avro schemas with schema-registry wire-format prefixes (Confluent only)
32
33
* Proper error handling for deserialization issues
33
34
34
35
## Terminology
@@ -255,6 +256,45 @@ Each Kafka record contains important metadata that you can access alongside the
255
256
|`value_schema_metadata`| Metadata about the value schema like `schemaId` and `dataFormat`| Data format and schemaId propagated when integrating with Schema Registry |
256
257
|`key_schema_metadata`| Metadata about the key schema like `schemaId` and `dataFormat`| Data format and schemaId propagated when integrating with Schema Registry |
257
258
259
+
### Using an offline Avro schema with a schema-registry wire-format prefix
260
+
261
+
When Confluent serializes messages with its schema-registry-aware Avro serializer (for example, `KafkaAvroSerializer`), each payload carries a wire-format header before the Avro body.
262
+
The header is 5 bytes long: 1-byte magic byte (`0x00`) followed by a 4-byte big-endian schema ID.
263
+
264
+
When the ESM Schema Registry integration is enabled, Lambda strips those bytes and populates the record's schema metadata. When you use an **offline Avro schema** without the ESM Schema Registry integration, the header reaches the function and prevents plain Avro deserialization.
265
+
266
+
Set `value_schema_wire_format` or `key_schema_wire_format` on `SchemaConfig` to `"CONFLUENT"`. Powertools validates the magic byte and strips the 5-byte header before running the Avro decoder.
267
+
268
+
???+ info "When do I need this?"
269
+
Use this option when you supply the Avro schema and the producer uses the Confluent wire format. If ESM Schema Registry integration has already removed the header, leave the option as `None`.
270
+
271
+
=== "Offline Avro schema with a Confluent prefix"
272
+
273
+
```python hl_lines="10"
274
+
from aws_lambda_powertools.utilities.kafka import SchemaConfig, kafka_consumer
275
+
from aws_lambda_powertools.utilities.kafka.consumer_records import ConsumerRecords
276
+
from aws_lambda_powertools.utilities.typing import LambdaContext
# with the validated 5-byte wire-format header removed.
292
+
...
293
+
```
294
+
295
+
???+ warning "Scope"
296
+
`value_schema_wire_format` and `key_schema_wire_format` apply only to **Avro** payloads. Leave them as `None` when ESM Schema Registry integration has already removed the wire-format header.
297
+
258
298
### Custom output serializers
259
299
260
300
Transform deserialized data into your preferred object types using output serializers. This can help you integrate Kafka data with your domain models and application architecture, providing type hints, validation, and structured data access.
0 commit comments