Azure / Azure/azure-functions-nodejs-extensions

[Feature] Add @azure/functions-extensions-kafka package for KafkaRecord binding

Open
#103 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
TypeScript
Stars
2
Forks
8
PR merge metrics
No merged PRs in 30d

Description

## Summary

Add a new `@azure/functions-extensions-kafka` package that enables Node.js users to bind to raw Apache Kafka records (`KafkaRecord` type) with full message metadata access (topic, partition, offset, key/value as Buffers, headers, timestamp, leader epoch).

This is the Node.js 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 Node.js library already has a `ResourceFactoryResolver` singleton pattern (used by ServiceBus extensions) that auto-dispatches `ModelBindingData` to registered factories based on `source`. No changes needed to the library or worker — only 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: `azure-functions-nodejs-extensions-kafka/`

Follow the `azure-functions-nodejs-extensions-servicebus` pattern exactly:

```
azure-functions-nodejs-extensions-kafka/
├── package.json
├── src/
│ ├── index.ts # Auto-registers factory on import
│ ├── kafka/
│ │ ├── kafkaRecordFactory.ts # Protobuf decode + KafkaRecord build
│ │ └── registerKafkaRecordFactory.ts # Register "AzureKafkaRecord" with ResourceFactoryResolver
│ └── util/
│ └── kafkaRecordDecoder.ts # Protobuf binary decoder using protobufjs
├── types/
│ └── index.d.ts # KafkaRecord, KafkaHeader, KafkaTimestamp exports
└── test/
└── kafkaRecordFactory.test.ts
```

| File | Description |
|------|-------------|
| `registerKafkaRecordFactory.ts` | Register factory under key `"AzureKafkaRecord"` with `ResourceFactoryResolver` |
| `kafkaRecordFactory.ts` | Decode Protobuf bytes from `ModelBindingData.content`, build `KafkaRecord` objects. Handle single and batch (array) |
| `kafkaRecordDecoder.ts` | Use `protobufjs` to decode `KafkaRecordProto` from binary |
| `types/index.d.ts` | TypeScript interfaces: `KafkaRecord`, `KafkaHeader`, `KafkaTimestamp`, `KafkaTimestampType` |
| `index.ts` | Call `registerKafkaRecordFactory()` on module load (auto-registration pattern) |

### Dependencies

```json
{
"dependencies": {
"@azure/functions-extensions-base": "^0.3.0",
"protobufjs": "^7.0.0"
}
}
```

### No changes needed in other repos

| Repo | Changes |
|------|---------|
| `azure-functions-nodejs-library` | None — `fromRpcTypedData.ts` already dispatches `modelBindingData` via `ResourceFactoryResolver` |
| `azure-functions-nodejs-worker` | None — worker is trigger-type agnostic |

## User Experience

```typescript
import { KafkaRecord } from '@azure/functions-extensions-kafka';

app.generic('KafkaTrigger', {
trigger: {
type: 'kafkaTrigger',
topic: 'my-topic',
brokerList: '%BrokerList%',
consumerGroup: '$Default',
},
handler: async (record: KafkaRecord, context) => {
context.log(`Topic: ${record.topic}`);
context.log(`Partition: ${record.partition}`);
context.log(`Offset: ${record.offset}`);
context.log(`Key: ${Buffer.from(record.key).toString('utf-8')}`);
context.log(`Timestamp: ${new Date(record.timestamp.unixTimestampMs)}`);

for (const header of record.headers) {
context.log(`Header: ${header.key} = ${Buffer.from(header.value).toString('utf-8')}`);
}
},
});
```

## Breaking Changes

None. This is a new package. All existing Kafka bindings (string, generic) continue to work.

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