Hacktoberfest 2026:维护者为十月标记出来的 issue,仍然开放、适合新手。 浏览 Hacktoberfest issue

[Feature] Add azurefunctions-extensions-bindings-kafka package for KafkaRecord binding

未关闭
#155 0 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看

@TsuyoshiUshio 已经在做这个了。

开始于 2026年4月24日。

  • #156 来自 @TsuyoshiUshio —— 未关闭

评估

难度
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-encoded KafkaRecordProto

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

主要语言
Python
星标
7
派生
16
平均合并
20 小时 30 分钟
30 天内合并 PR
1

环境准备

  • 没有 Dockerfile 或 Docker Compose 文件
  • 没有 Pull Request 模板
  • 阅读贡献指南

从这里开始

  1. 先读完整个 Issue,再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库,在一个分支上完成修改。
  4. 提交 Pull Request,并在描述里引用这个 Issue 编号。

相似的 Issue

更多 Python Issue

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。