Azure / Azure/azure-functions-java-worker

[Feature] Support raw Kafka record (KafkaRecord) binding with Protobuf deserialization

オープン
#868 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る
主要言語
Java
スター
103
フォーク
74
平均マージ
4日 8時間
マージ済み PR(30日)
2

説明

## Summary

Add support for binding to raw Apache Kafka records (`KafkaRecord` type) in the Java worker, enabling users to access full Kafka message metadata (topic, partition, offset, key/value as raw bytes, headers, timestamp, leader epoch).

This is the Java 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 `ParameterBindingData` with:
- `source`: `"AzureKafkaRecord"`
- `content_type`: `"application/x-protobuf"`
- `content`: Protobuf-encoded `KafkaRecordProto`

Currently, Java users can only bind to `String`, `byte[]`, or `Map` — they cannot access structured metadata like headers, timestamps, or partition info.

## Protobuf Schema (shared across all languages)

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

### Part A: New POJO types (in `azure-functions-java-library`)

| File | Description |
|------|-------------|
| `KafkaRecord.java` | Main POJO: topic, partition, offset, key (byte[]), value (byte[]), timestamp, headers, leaderEpoch (Integer) |
| `KafkaHeader.java` | Header: key (String) + value (byte[]) + `getValueAsString()` helper |
| `KafkaTimestamp.java` | Timestamp: unixTimestampMs (long) + type (enum) + `getDateTime()` -> OffsetDateTime |
| `KafkaTimestampType.java` | Enum: NotAvailable(0), CreateTime(1), LogAppendTime(2) |

No annotation changes needed — existing `@KafkaTrigger` works as-is.

### Part B: Worker-side Protobuf deserializer (in `azure-functions-java-worker`)

| File | Change |
|------|--------|
| `KafkaRecordProto.proto` | New: Add proto schema, configure `protobuf-maven-plugin` for code generation |
| `RpcModelBindingDataSource.java` | Modify: Add `content_type` dispatch — `application/json` -> existing JSON path, `application/x-protobuf` -> new Protobuf deserialization |
| `KafkaRecordProtoDeserializer.java` | New: Map `KafkaRecordProto` -> `KafkaRecord` POJO |

Note: `protobuf-java` 3.25.5 is already in pom.xml. Maven protobuf plugin setup is needed for `.proto` compilation.

### Part C: No changes to `azure-functions-java-additions`

`KafkaRecord` is a data container, not an Azure SDK client — the SdkType/Hydrator pattern is not applicable.

## User Experience

```java
// Existing (continues to work)
@FunctionName("ExistingTrigger")
public void run(@KafkaTrigger(...) String message) { }

// NEW: Full record access
@FunctionName("KafkaRecordTrigger")
public void run(
@KafkaTrigger(name = "record", topic = "my-topic",
brokerList = "%BrokerList%", consumerGroup = "$Default")
KafkaRecord record,
final ExecutionContext context) {

context.getLogger().info("Topic: " + record.getTopic());
context.getLogger().info("Partition: " + record.getPartition());
context.getLogger().info("Offset: " + record.getOffset());
context.getLogger().info("Key: " + new String(record.getKey()));
context.getLogger().info("Timestamp: " + record.getTimestamp().getDateTime());

for (KafkaHeader header : record.getHeaders()) {
context.getLogger().info("Header: " + header.getKey() + " = " + header.getValueAsString());
}
}

// NEW: Batch mode
@FunctionName("KafkaBatchTrigger")
public void run(
@KafkaTrigger(..., cardinality = Cardinality.MANY)
KafkaRecord[] records) { ... }
```

## Breaking Changes

None. This is purely additive. All existing binding types (`String`, `byte[]`, POJO) continue to work.

## Implementation Order

1. POJO types in `java-library` (no dependency on host release)
2. Protobuf deserializer in `java-worker` (requires host extension 4.3.1 NuGet — already released)

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

コントリビューションガイド

コントリビューションガイドを開く

調査の方向性

まず azure-functions-java-library の KafkaRecord.java、KafkaHeader.java、KafkaTimestamp.java、KafkaTimestampType.java の追加を確認し、次に azure-functions-java-worker の KafkaRecordProto.proto、RpcModelBindingDataSource.java、KafkaRecordProtoDeserializer.java を調査します。protobuf の生成と、application/json および application/x-protobuf の両方の dispatch を検証し、その後、既存の binding が引き続き動作し、raw records が指定されたメタデータを公開することをテストします。

索引モデルが issue の本文から書いたものです。

評価

技術スタック
java
領域
api, backend
issue の種類
機能追加
難易度
4/5
見積もり時間
3〜5日
活発さ
静か
明瞭さ
明確に書かれている
初心者へのやさしさ
50/100

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。