[Feature] Add azurefunctions-extensions-bindings-kafka package for KafkaRecord binding
评估
- 难度
- 4/5
- 预计耗时
- 3-5 天
- 新手友好度
- 68/100
- Issue 类型
- 功能
- 描述清晰度
- 描述清楚
- 活跃度
- 冷清
- 技术栈
- kafka, python
调研方向
首先遵循 azurefunctions-extensions-bindings-eventhub 的模式,并检查列出的 setup.cfg 或 pyproject.toml、kafkaRecord.py、kafkaRecordConverter.py、init.py 和 KafkaRecordProto.proto 文件。在检查单条记录和集合绑定用例的同时,运行 tests/test_kafka_record_converter.py。当新 package 暴露所请求的 record、header、timestamp 和 timestamp-type 对象,并且所有列出的 metadata 都可用时,即表示完成。
由索引模型根据 Issue 内容生成。
描述
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
- 主要语言
- Python
- 星标
- 7
- 派生
- 16
- 平均合并
- 20 小时 30 分钟
- 30 天内合并 PR
- 1
环境准备
- 没有 Dockerfile 或 Docker Compose 文件
- 没有 Pull Request 模板
- 阅读贡献指南
从这里开始
- 先读完整个 Issue,再读项目的贡献指南。
- 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
- Fork 仓库,在一个分支上完成修改。
- 提交 Pull Request,并在描述里引用这个 Issue 编号。
相似的 Issue
-
难度 2/5 1-3 小时 新手友好度 74/100
aws-samples/sample-per-model-bedrock#243 ·
维护者通常 1 天内回复
-
难度 2/5 1-3 小时 新手友好度 66/100
FuRongJun-1999/dsh-memory#404 ·
维护者通常 1 天内回复
-
难度 2/5 1-3 小时 新手友好度 72/100
维护者通常 1 天内回复
-
bug frontend
难度 2/5 1-3 小时 新手友好度 68/100
PennyLaneAI/catalyst#3337 ·
维护者通常 1 天内回复
-
Workflow.schedule stops when a cron deadline passes before sleep可能已有人在做 @HuaTNA 今天认领。 未关闭
难度 2/5 1-3 小时 新手友好度 82/100
维护者通常 1 天内回复