[Feature] Add azurefunctions-extensions-bindings-kafka package for KafkaRecord binding
Avaliação
- Dificuldade
- 4/5
- Tempo estimado
- 3-5 dias
- Facilidade para iniciantes
- 68/100
- Tipo de issue
- Funcionalidade
- Clareza
- Claramente especificada
- Status de atividade
- Pouca atividade
- Stack de tecnologia
- kafka, python
- Domínio
- stream-processing
Direção de pesquisa
Comece seguindo o padrão de azurefunctions-extensions-bindings-eventhub e revise os arquivos setup.cfg ou pyproject.toml, kafkaRecord.py, kafkaRecordConverter.py, init.py e KafkaRecordProto.proto listados. Execute tests/test_kafka_record_converter.py enquanto verifica os casos de binding de registro único e de coleção. Está concluído quando o novo pacote expuser os objetos solicitados de registro, cabeçalho, timestamp e tipo de timestamp, com todos os metadados listados disponíveis.
Escrita pelo modelo de indexação a partir do texto da issue.
Descrição
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
- Linguagem predominante
- Python
- Estrelas
- 7
- Forks
- 16
- Merge médio
- 20h 30min
- PRs com merge (30d)
- 1
Preparar o ambiente
- Sem Dockerfile nem arquivo Docker Compose
- Sem modelo de pull request
- Ler o guia de contribuição
Primeiros passos
- Leia a issue inteira e depois o guia de contribuição do projeto.
- Comente na issue dizendo que vai assumir — evita que duas pessoas façam o mesmo trabalho.
- Faça um fork do repositório e trabalhe em uma branch.
- Abra um pull request que referencie o número da issue.
Issues semelhantes
-
Questions crashes on a real SQuAD2 no-answer predictionTalvez já em andamento @HuaTNA assumiu hoje. Aberta
Dificuldade 2/5 1-3 horas Facilidade para iniciantes 72/100
Mantenedores costumam responder em até 1 dia
-
GST is read as Gulf Standard Time (UTC+4) instead of Guam Standard Time (UTC+10)Talvez já em andamento @yang00333 assumiu hoje. Aberta
Dificuldade 2/5 1-3 horas Facilidade para iniciantes 78/100
DOI-USGS/dataretrieval-python#454 ·
Mantenedores costumam responder em até 4 dias
-
area:core-platform effort:S priority:P3 status:in-review type:test
Dificuldade 2/5 Menos de uma hora Facilidade para iniciantes 80/100
randomparity/kdive#3246 · 1 comentário ·
Mantenedores costumam responder em até 1 dia
-
Requantification library converts m/z to mass with the hydrogen-atom mass instead of the proton massAbertabug
Dificuldade 1/5 Menos de uma hora Facilidade para iniciantes 85/100
nf-core/metaboigniter#130 ·
Mantenedores costumam responder em até 1 dia
-
netbox status: needs owner type: documentation
Dificuldade 1/5 Menos de uma hora Facilidade para iniciantes 75/100
netbox-community/netbox#23412 · 1 reação ·
Mantenedores costumam responder em até 1 dia