Azure / Azure/azure-functions-python-extensions

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

Open
#155 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
7
Forks
15
Avg merge
2d 7h
Merged PRs (30d)
4

Description

## 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)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.