Hacktoberfest 2026: los issues que los mantenedores marcaron para octubre, abiertos y aptos para principiantes. Explorar issues de Hacktoberfest

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

Abierto
  • ¿Algo no está bien?
#868 0 comentarios 0 reacciones 0 asignados Ver en GitHub

@TsuyoshiUshio ya está trabajando en esto.

Desde el 24/4/2026.

  • #869 de @TsuyoshiUshio — abierto

Evaluación

Dificultad
4/5
Tiempo estimado
3-5 días
Aptitud para principiantes
50/100
Tipo de issue
Nueva funcionalidad
Claridad
Bien especificado
Estado de actividad
Tranquilo
Stack tecnológico
java
Área
api, backend

Línea de trabajo

Comienza con las incorporaciones de KafkaRecord.java, KafkaHeader.java, KafkaTimestamp.java y KafkaTimestampType.java en azure-functions-java-library; después, inspecciona KafkaRecordProto.proto, RpcModelBindingDataSource.java y KafkaRecordProtoDeserializer.java en azure-functions-java-worker. Verifica la generación de protobuf y el dispatch tanto de application/json como de application/x-protobuf; después, comprueba que los bindings existentes sigan funcionando y que los raw records expongan los metadatos especificados.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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

Lenguaje dominante
Java
Estrellas
103
Forks
74
Merge medio
12 h 49 min
PR fusionados (30 d)
2

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de Azure/azure-functions-java-worker

Todos los issues de Azure/azure-functions-java-worker

Issues similares

Más issues de Java

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.