Azure / Azure/azure-functions-python-extensions

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

Đang mở
#155 0 bình luận 0 reaction 0 người được giao Xem trên GitHub
Ngôn ngữ chính
Python
Star
7
Fork
15
Merge trung bình
2 ngày 7 giờ
Pull request đã merge (30 ngày)
4

Mô tả

## 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](https://github.com/Azure/azure-functions-kafka-extension/issues/612). The host-side Kafka Extension 4.3.1 and .NET Isolated Worker ([PR #3356](https://github.com/Azure/azure-functions-dotnet-worker/pull/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

```protobuf
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

```ini
[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

```python
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](https://github.com/Azure/azure-functions-kafka-extension/issues/612)
- .NET Worker (done): [Azure/azure-functions-dotnet-worker#3356](https://github.com/Azure/azure-functions-dotnet-worker/pull/3356)

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Hướng nghiên cứu

Bắt đầu bằng cách làm theo mẫu azurefunctions-extensions-bindings-eventhub và xem xét các tệp setup.cfg hoặc pyproject.toml, kafkaRecord.py, kafkaRecordConverter.py, __init__.py và KafkaRecordProto.proto được liệt kê. Chạy tests/test_kafka_record_converter.py đồng thời kiểm tra các trường hợp binding bản ghi đơn và collection. Hoàn tất khi package mới cung cấp các đối tượng record, header, timestamp và timestamp-type được yêu cầu, với tất cả metadata được liệt kê đều khả dụng.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Đánh giá

Công nghệ
kafka, python
Lĩnh vực
stream-processing
Loại issue
Tính năng
Độ khó
4/5
Thời gian dự kiến
3-5 ngày
Mức độ hoạt động
Ít trao đổi
Độ rõ ràng
Đặc tả rõ ràng
Mức phù hợp với người mới
68/100

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.