jk-shah opened a new issue, #41225:
URL: https://github.com/apache/beam/issues/41225
### Description
Consuming from Kafka topics serialized with Confluent Schema Registry (magic
byte `0x00` + 4-byte schema ID) is a ubiquitous industry pattern.
Currently in Beam, consuming from Schema Registry requires writing custom
Java deserializers or complex manual Python Avro decode steps. In Beam YAML,
`ReadFromKafka` only natively supports raw bytes or JSON.
### Proposed Architecture
Add first-class Schema Registry support directly in `ReadFromKafka` (both
Python wrapper and Java cross-language provider in `standard_io.yaml`):
```yaml
pipeline:
transforms:
- type: ReadFromKafka
config:
topic: orders
bootstrap_servers: "kafka:9092"
schema_registry: "http://schema-registry:8081";
format: avro # automatically resolves schema ID from wire format
```
### Key Capabilities
1. **Wire Format Parsing**: Parses Confluent wire format header (magic byte
0 + 4-byte Big-Endian schema ID).
2. **Schema Cache**: Fetches and caches Avro/Protobuf schema definitions
from Schema Registry by ID.
3. **Beam Schema Mapping**: Automatically converts deserialized records to
strongly-typed Beam Schema `Row` objects.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
