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

[TsuyoshiUshio]

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. 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 ParameterBindingData with:

• source: "AzureKafkaRecord"
• content_type: "application/x-protobuf"
• content: Protobuf-encoded KafkaRecordProto

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

Protobuf Schema (shared across all languages)

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

// 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
• .NET Worker (done): Azure/azure-functions-dotnet-worker#3356