[Feature] Add azurefunctions-extensions-bindings-kafka package for KafkaRecord binding
Evaluación
- Dificultad
- 4/5
- Tiempo estimado
- 3-5 días
- Aptitud para principiantes
- 68/100
- Tipo de issue
- Nueva funcionalidad
- Claridad
- Bien especificado
- Estado de actividad
- Tranquilo
- Stack tecnológico
- kafka, python
- Área
- stream-processing
Línea de trabajo
Comienza siguiendo el patrón de azurefunctions-extensions-bindings-eventhub y revisa los archivos setup.cfg o pyproject.toml, kafkaRecord.py, kafkaRecordConverter.py, init.py y KafkaRecordProto.proto indicados. Ejecuta tests/test_kafka_record_converter.py mientras compruebas los casos de binding de un solo registro y de colecciones. Se considera terminado cuando el nuevo paquete expone los objetos de registro, encabezado, marca de tiempo y tipo de marca de tiempo solicitados, con todos los metadatos indicados disponibles.
Escrito por el modelo de indexación a partir del texto del issue.
Descripción
Summary
Add a new azurefunctions-extensions-bindings-kafka package that enables Python users to bind to raw Apache Kafka records (KafkaRecord type) with full message metadata access (topic, partition, offset, key/value as bytes, headers, timestamp, leader epoch).
This is the Python implementation of Azure/azure-functions-kafka-extension#612. The host-side Kafka Extension 4.3.1 and .NET Isolated Worker (PR #3356) are already complete.
Background
The host-side Kafka Extension (4.3.1) serializes IKafkaEventData to Protobuf and sends it as ModelBindingData with:
source:"AzureKafkaRecord"content_type:"application/x-protobuf"content: Protobuf-encodedKafkaRecordProto
The Python worker already supports deferred bindings via DEFERRED_BINDING_REGISTRY which auto-discovers extension packages. The existing KafkaEvent / KafkaTriggerConverter in azure-functions-python-library handles simple types (string, bytes) — the new KafkaRecord will use the deferred binding path via a new extensions package.
Protobuf Schema
message KafkaRecordProto {
string topic = 1;
int32 partition = 2;
int64 offset = 3;
optional bytes key = 4;
optional bytes value = 5;
KafkaTimestampProto timestamp = 6;
repeated KafkaHeaderProto headers = 7;
optional int32 leader_epoch = 8;
reserved 9 to 15;
}
message KafkaTimestampProto {
int64 unix_timestamp_ms = 1;
int32 type = 2; // 0=NotAvailable, 1=CreateTime, 2=LogAppendTime
}
message KafkaHeaderProto {
string key = 1;
optional bytes value = 2;
}
Required Changes
New package: azurefunctions-extensions-bindings-kafka/
Follow the azurefunctions-extensions-bindings-eventhub pattern:
azurefunctions-extensions-bindings-kafka/
├── setup.cfg (or pyproject.toml)
├── azurefunctions/
│ └── extensions/
│ └── bindings/
│ └── kafka/
│ ├── __init__.py # Export KafkaRecord, KafkaHeader, etc.
│ ├── kafkaRecord.py # SdkType wrapper — deserializes Protobuf
│ └── kafkaRecordConverter.py # InConverter (binding='kafkaTrigger')
├── proto/
│ └── KafkaRecordProto.proto # For protoc code generation
└── tests/
└── test_kafka_record_converter.py
| File | Description |
|---|---|
kafkaRecord.py |
KafkaRecord(SdkType) — wraps ModelBindingData, deserializes Protobuf on get_sdk_type(). Properties: topic, partition, offset, key (bytes), value (bytes), timestamp, headers, leader_epoch |
kafkaRecordConverter.py |
KafkaRecordConverter(InConverter, binding='kafkaTrigger') — handles model_binding_data and collection_model_binding_data cases |
__init__.py |
Export public types: KafkaRecord, KafkaHeader, KafkaTimestamp, KafkaTimestampType |
Dependencies
[options]
install_requires =
azurefunctions-extensions-base>=1.1.0
protobuf>=4.0.0,<6.0
No changes needed in other repos
| Repo | Changes |
|---|---|
azure-functions-python-library |
None — existing KafkaEvent / KafkaTriggerConverter unchanged (simple types path) |
azure-functions-python-worker |
None — DEFERRED_BINDING_REGISTRY auto-discovers the new extension |
User Experience
import azure.functions as func
from azurefunctions.extensions.bindings.kafka import KafkaRecord
app = func.FunctionApp()
@app.kafka_trigger(
arg_name="record",
topic="my-topic",
broker_list="%BrokerList%",
consumer_group="$Default")
def kafka_trigger(record: KafkaRecord):
logging.info(f"Topic: {record.topic}")
logging.info(f"Partition: {record.partition}")
logging.info(f"Offset: {record.offset}")
logging.info(f"Key: {record.key.decode('utf-8')}")
logging.info(f"Timestamp: {record.timestamp.datetime}")
for header in record.headers:
logging.info(f"Header: {header.key} = {header.value.decode('utf-8')}")
# Batch mode
@app.kafka_trigger(..., cardinality=func.Cardinality.MANY)
def kafka_batch(records: list[KafkaRecord]):
for record in records:
logging.info(f"{record.topic}:{record.partition}:{record.offset}")
Breaking Changes
None. This is a new package. All existing Kafka bindings (str, bytes, KafkaEvent) continue to work via the existing library path.
Related Issues
- Parent: Azure/azure-functions-kafka-extension#612
- .NET Worker (done): Azure/azure-functions-dotnet-worker#3356
- Lenguaje dominante
- Python
- Estrellas
- 7
- Forks
- 16
- Merge medio
- 20 h 30 min
- PR fusionados (30 d)
- 1
Preparar el entorno
- Sin Dockerfile ni archivo de Docker Compose
- Sin plantilla de pull request
- Leer la guía de contribución
Primeros pasos
- Lee el issue completo y luego la guía de contribución del proyecto.
- Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
- Haz un fork del repositorio y trabaja en una rama.
- Abre un pull request que haga referencia al número del issue.
Issues similares
-
approved Good First Issue scope: docs
Dificultad 1/5 Menos de una hora Aptitud para principiantes 85/100
hiero-ledger/hiero-sdk-python#2730 · 3 comentarios ·
Los mantenedores suelen responder en 1 día
-
Dificultad 2/5 1-3 horas Aptitud para principiantes 62/100
wemake-services/wemake-django-template#3178 · 1 reacción ·
Los mantenedores suelen responder en 1 día
-
md/asciidoc/csv: an explicit `encoding="utf-8"` keeps a leading BOM, the auto-detect path drops itAbierto
Dificultad 2/5 1-3 horas Aptitud para principiantes 82/100
docling-project/docling#4846 ·
Los mantenedores suelen responder en 1 día
-
Dificultad 2/5 1-3 horas Aptitud para principiantes 78/100
neurostuff/NiMARE#1181 ·
Los mantenedores suelen responder en 1 día
-
Dificultad 2/5 1-3 horas Aptitud para principiantes 78/100
pytest-dev/pytest#15180 ·
Los mantenedores suelen responder en 1 día