[Feature Request] Support raw Kafka record (ConsumeResult<TKey, TValue>) in .NET Isolated Worker

[TsuyoshiUshio]

[Feature Request] Support raw Kafka record (ConsumeResult<TKey, TValue>) in .NET Isolated Worker

Summary

Add support for binding to raw Kafka records with full metadata access in the .NET Isolated Worker model. Currently, the KafkaTrigger only supports binding to string, byte[], or our abstracted KafkaEventData types, which limits access to important Kafka message metadata.

Problem Statement

Customer Scenario

Customers using the .NET Isolated Worker (the recommended hosting model) need access to the complete Kafka message metadata, similar to what Confluent.Kafka.ConsumeResult<TKey, TValue> provides. Current limitations:

Available Today	Missing
Key, Value	Full Message<TKey, TValue> object
Offset, Partition, Topic	IsPartitionEOF flag
Timestamp	LeaderEpoch
Headers (copied)	Original Confluent.Kafka.Headers instance
	TopicPartitionOffset as a single object

Why This Matters

1. Custom header processing - Correlation IDs, trace context, custom routing metadata
2. Precise offset management - Manual commit scenarios, idempotency patterns
3. Partition-aware logic - Sharding strategies based on partition info
4. Integration with existing code - Organizations often have internal libraries that expect ConsumeResult types

Comparison with Other Triggers

EventHubs and ServiceBus extensions already support binding to their raw SDK types:

Trigger	Abstracted Type	Raw SDK Type (Supported)
EventHubs	Custom data types	Azure.Messaging.EventHubs.EventData ✅
ServiceBus	Custom data types	Azure.Messaging.ServiceBus.ServiceBusReceivedMessage ✅
Kafka	KafkaEventData<TKey, TValue>	Confluent.Kafka.ConsumeResult<TKey, TValue> ❌

---

Proposed Architecture

Option A: Custom KafkaConsumeResult Type (Recommended for Initial Implementation)

Since Confluent.Kafka.ConsumeResult<TKey, TValue> lacks a factory method to reconstruct from serialized bytes (unlike Azure SDK types which use AMQP), we propose creating a new type that mirrors its structure.

┌─────────────────────────────────────────────────────────────────────────┐
│                        HOST PROCESS (WebJobs Extension)                 │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │  KafkaListener                                                   │   │
│  │  ┌─────────────────┐    ┌──────────────────────────────────┐   │   │
│  │  │ Consume()       │───►│ ConsumeResult<TKey, TValue>      │   │   │
│  │  │ (librdkafka)    │    │ (Confluent.Kafka)                │   │   │
│  │  └─────────────────┘    └──────────────┬───────────────────┘   │   │
│  │                                         │                       │   │
│  │                         ┌───────────────▼───────────────────┐   │   │
│  │                         │ Serialize to ModelBindingData    │   │   │
│  │                         │ (MessagePack/Proto format)        │   │   │
│  │                         │ - All metadata preserved          │   │   │
│  │                         └───────────────┬───────────────────┘   │   │
│  └─────────────────────────────────────────┼───────────────────────┘   │
│                                             │ gRPC                      │
└─────────────────────────────────────────────┼───────────────────────────┘
                                              │
┌─────────────────────────────────────────────┼───────────────────────────┐
│                     WORKER PROCESS (Isolated Extension)                 │
│                                             │                           │
│  ┌──────────────────────────────────────────▼──────────────────────┐   │
│  │  KafkaConsumeResultConverter : IInputConverter                   │   │
│  │  [SupportsDeferredBinding]                                       │   │
│  │  [SupportedTargetType(typeof(KafkaConsumeResult<,>))]           │   │
│  │                                                                   │   │
│  │  ┌─────────────────┐    ┌──────────────────────────────────┐    │   │
│  │  │ ModelBindingData│───►│ Deserialize to                   │    │   │
│  │  │ (bytes)         │    │ KafkaConsumeResult<TKey, TValue> │    │   │
│  │  └─────────────────┘    └──────────────────────────────────┘    │   │
│  └──────────────────────────────────────────┬──────────────────────┘   │
│                                              │                          │
│  ┌───────────────────────────────────────────▼─────────────────────┐   │
│  │  User Function                                                   │   │
│  │                                                                   │   │
│  │  [Function("KafkaTrigger")]                                      │   │
│  │  public void Run(                                                │   │
│  │      [KafkaTrigger(...)]                                         │   │
│  │      KafkaConsumeResult<string, MyEvent> record)                 │   │
│  │  {                                                               │   │
│  │      var offset = record.Offset;                                 │   │
│  │      var partition = record.Partition;                           │   │
│  │      var headers = record.Message.Headers;                       │   │
│  │      var value = record.Message.Value;                           │   │
│  │      // Full metadata access!                                    │   │
│  │  }                                                               │   │
│  └──────────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────────┘

Option B: Upstream Confluent.Kafka Enhancement (Long-term)

There are active proposals in the Confluent.Kafka repository that could simplify this:

Issue	Description	Status
confluentinc/confluent-kafka-dotnet#2157	Factory pattern for custom consume result types	Open (Dec 2023)
confluentinc/confluent-kafka-dotnet#2543	Allocation-free API with MessageReader	Open (Nov 2025)

If #2157 is adopted:

// In WebJobs Extension - could directly create our type without intermediate serialization
TConsumeResult Consume<TConsumeResult>(
    int millisecondsTimeout, 
    IConsumeResultFactory<TKey, TValue, TConsumeResult> factory);

This would eliminate the need for custom serialization, as we could implement IConsumeResultFactory to create KafkaConsumeResult directly.

---

Required Changes

Repository: Azure/azure-functions-kafka-extension (WebJobs Extension)

File/Area	Change
KafkaEventData.cs	Add serialization support for full metadata
Trigger/KafkaTriggerBindingStrategy.cs	Support ModelBindingData creation with full record
New: Serialization/KafkaRecordSerializer.cs	MessagePack/Proto serializer for ConsumeResult
Microsoft.Azure.WebJobs.Extensions.Kafka.csproj	Add serialization dependencies

Repository: Azure/azure-functions-dotnet-worker (Isolated Worker Extension)

File/Area	Change
New: KafkaConsumeResult.cs	New type mirroring ConsumeResult<TKey, TValue> structure
New: KafkaMessage.cs	New type mirroring Message<TKey, TValue>
New: KafkaConsumeResultConverter.cs	IInputConverter implementation
New: KafkaHeaders.cs	Headers collection type
Worker.Extensions.Kafka.csproj	Add serialization dependencies

Shared: Serialization Schema

Define a shared schema (MessagePack or Protocol Buffers):

// Example Proto definition
message KafkaConsumeResultProto {
    string topic = 1;
    int32 partition = 2;
    int64 offset = 3;
    optional int32 leader_epoch = 4;
    KafkaMessageProto message = 5;
    bool is_partition_eof = 6;
}

message KafkaMessageProto {
    bytes key = 1;
    bytes value = 2;
    repeated KafkaHeaderProto headers = 3;
    KafkaTimestampProto timestamp = 4;
}

message KafkaHeaderProto {
    string key = 1;
    bytes value = 2;
}

message KafkaTimestampProto {
    int64 unix_timestamp_ms = 1;
    int32 type = 2;  // CreateTime, LogAppendTime, etc.
}

---

User Experience

New API (Non-Breaking Addition)

Users can opt-in to the new type while existing code continues to work:

// ✅ Existing (continues to work)
[Function("ExistingTrigger")]
public void Run([KafkaTrigger(...)] string message) { }

[Function("ExistingTriggerBytes")]  
public void Run([KafkaTrigger(...)] byte[] message) { }

// ✅ NEW: Full record access
[Function("NewTrigger")]
public void Run([KafkaTrigger(...)] KafkaConsumeResult<string, MyEvent> record)
{
    // Access everything
    Console.WriteLine($"Topic: {record.Topic}");
    Console.WriteLine($"Partition: {record.Partition}");
    Console.WriteLine($"Offset: {record.Offset}");
    Console.WriteLine($"LeaderEpoch: {record.LeaderEpoch}");
    Console.WriteLine($"Key: {record.Message.Key}");
    Console.WriteLine($"Value: {record.Message.Value}");
    Console.WriteLine($"Timestamp: {record.Message.Timestamp}");
    
    // Headers with full API
    foreach (var header in record.Message.Headers)
    {
        Console.WriteLine($"Header: {header.Key} = {Encoding.UTF8.GetString(header.GetValueBytes())}");
    }
}

// ✅ NEW: Batch mode
[Function("BatchTrigger")]
public void Run([KafkaTrigger(...)] KafkaConsumeResult<string, MyEvent>[] records)
{
    foreach (var record in records)
    {
        // Process each record with full metadata
    }
}

Breaking Changes

None - This is an additive feature. All existing binding types (string, byte[], KafkaEventData<T>) continue to work.

---

Estimated Effort

Task	Estimate
Serialization schema design	3 days
WebJobs Extension changes	5 days
Isolated Worker Extension changes	5 days
End-to-end testing	5 days
Documentation	2 days
Total	~4 weeks

---

Next Steps

1. Engage with Confluent - Comment on #2157 to express Azure Functions' interest in the factory pattern. If adopted, it would significantly simplify our implementation.

2. Design Review - Finalize serialization format (MessagePack vs Proto) and KafkaConsumeResult API surface.

3. Prototype - Create a proof-of-concept in a feature branch to validate the approach.

4. Customer Validation - Share the proposed API with customers to confirm it meets their requirements.

5. Implementation - Proceed with full implementation following design review approval.

---

Related Issues

• This issue addresses the request for raw Kafka record support mentioned in customer feedback
• Related upstream issues: confluentinc/confluent-kafka-dotnet#2157, #2543

/cc @Azure/azure-functions-kafka

[TsuyoshiUshio]

Implementation Plan

This issue will be implemented in multiple phases across two repositories:

Phase 1: Host-side metadata & serialization (WebJobs Extension) ✅

PR: #613

• Add LeaderEpoch (int?) and IsPartitionEOF (bool) to IKafkaEventData interface
• Preserve these fields from ConsumeResult in all KafkaEventData constructors
• New KafkaRecordSerializer - JSON serialization for ModelBindingData transmission
• New ParameterBindingData converter in KafkaEventDataConvertManager
  • Source: "AzureKafkaConsumeResult", ContentType: "application/json"
• Expanded trigger binding contract (LeaderEpoch, IsPartitionEOF in trigger_metadata)
• Non-breaking: All existing bindings (string, byte[], KafkaEventData) unchanged
• 15 new unit tests

Phase 2: .NET Isolated Worker POCO types & Converter

Target repo: azure-functions-dotnet-worker

• New KafkaConsumeResult<TKey, TValue> POCO type
• New KafkaMessage<TKey, TValue>, KafkaHeaders, KafkaHeader, KafkaTimestamp types
• KafkaConsumeResultConverter : IInputConverter with [SupportsDeferredBinding]
• Update KafkaTriggerAttribute with [InputConverter] attribute
• Unit tests for deserialization

Phase 3: End-to-End tests & documentation

• E2E tests: single dispatch, batch dispatch, backward compatibility
• Sample code demonstrating KafkaConsumeResult usage
• README and CHANGELOG updates

Phase 4+: Cross-language support (Future)

Analysis completed - all approaches are non-breaking and additive:

Language	Approach	Effort
Node.js	New @azure/functions-extensions-kafka npm package (ServiceBus pattern)	Medium
Java	KafkaConsumeResult POJO + SdkType implementation	Medium
Python	Update KafkaTriggerConverter or deferred binding extension	Small-Medium
PowerShell	New metadata auto-available via $TriggerMetadata (Phase 1 already covers this)	None

[daankets]

This proposal seems to match our needs as well. I would however propose one more addition: proper support for handling failures. Today, when we have a failure to process a message, we write it to a dead-letter queue. However, that may fail as well in case of a technical incident. In such a case, we are losing messages today due to the at-most-once delivery guarantee of the current trigger solution. An at-least-once solution that does not commit in case of failure would be a welcome solution. Today (in-process), our only option is to use FailFast to exit the process and therefore the kill trigger (and lose the unsaved commit offset). However, we can't use that solution with the isolated worker setup, where failfast would only kill the worker.

[TsuyoshiUshio]

Thank you for reporting @daankets I'll have a look when we release next version with this feature. I'll update.

[TsuyoshiUshio]

Hi @daankets I wrote a spec. Are you good for it?

[daankets]

Hey @TsuyoshiUshio I'll take a look! I greatly appreciate you taking the time to look into this!

[TsuyoshiUshio]

I've got a customer feedback about this issue. Let's discuss on this. I'll do some research for the following comment.

---

Proposal: Redefine Scope of KafkaTrigger Feature Request

I believe a few concepts are currently mixed up on the referenced page. This issue proposes redefining the scope of the KafkaTrigger feature to avoid unnecessary complexity and better align with the Apache Kafka specification.

---

Rationale / Analysis

A “raw Kafka record” is not a ConsumeResult<TKey, TValue>

Raw Kafka record

A raw Kafka record:

• Represents one record as stored/produced on Kafka infrastructure
• Contains:
  • Key and value (when present), each defined as a sequence of bytes
  • Headers
  • Partition
  • Offset
  • Topic
  • Timestamp

This is the fundamental Kafka data model.

ConsumeResult<TKey, TValue> (Confluent Kafka library)

A ConsumeResult<TKey, TValue>:

• Represents the result of a fetch attempt, not necessarily a raw Kafka record
• Carries metadata about the consumer state, including:
  • Whether the consumer was at the end of the topic at read time (i.e., consumer lag == 0)
• May or may not carry a Kafka record:
  • Contains a record when one is available
  • Contains no record when there is nothing left to consume
• Uses generics (TKey, TValue), which:
  • Do not support topics using the TopicRecordName strategy (i.e., multiple schemas on the same topic)
  • Require the implementation to correctly deserialize the data

This introduces additional concerns:

• The need for Schema Registry support:
  • Which registry systems must be supported? (Microsoft, Confluent, others?)
  • Which serialization formats? (JSON, Avro, Protobuf are currently supported by Confluent Schema Registry)
  • Whether to support the Confluent Wire Format
• Tight coupling to Confluent’s type system and APIs

---

Proposal

1. Scope limitation

   • Extend the isolated worker KafkaTrigger functionality to support the pure Apache Kafka specification only

2. New data model

   • Introduce a dedicated KafkaRecord C# type
   • Do not reuse or modify the existing EventData type, as it expresses explicit event semantics
     • A Kafka record does not necessarily represent an “event” (though the inverse can be true)

   (See attached C# code snippet.)

3. Trigger binding

   • Add support for a KafkaTriggerBinding with a method signature similar to the one attached (or equivalent)

   (See attached C# code snippet.)

---

Benefits of This Approach

• No added complexity or assumptions around Schema Registry support
• Enables use of TopicRecordName strategy
• Reduced development effort
• No coupling to Confluent’s type system at the interface layer (specifically ConsumeResult<TKey, TValue>)
  • Avoids issues with the non-exposed constructor of ConsumeResult<TKey, TValue>
• Less dependency on the planning, evolution, and implementation details of the Confluent Kafka library

---

If anything is unclear or raises questions, feel free to reach out—I’m happy to elaborate beyond my two cents.

[TsuyoshiUshio]

Cross-Language Analysis: KafkaRecord Type Support

Following the previous proposal from a customer to redefine the scope around a pure Apache Kafka spec-aligned KafkaRecord type (separate from ConsumeResult), I investigated all language workers to assess what changes are needed.

Design Approach

• Non-breaking: KafkaEventData (existing) is maintained. KafkaRecord (new, bytes-based) is added in parallel
• Users opt in via function signature type (same pattern as EventHubs/ServiceBus), not configuration
• IsPartitionEOF is intentionally excluded from KafkaRecord (it is consumer state, not record metadata per Apache Kafka spec)
• Host-side sends KafkaRecord as ModelBindingData with key/value as base64-encoded bytes - one host change supports all languages

Per-Language Assessment

Language	Current Kafka Support	Extension Package Pattern	KafkaRecord Effort	Priority
.NET Isolated	KafkaEventData in Worker.Extensions.Kafka NuGet	Worker.Extensions.Kafka NuGet	Medium - new POCO + IInputConverter	1 (customer env)
Node.js	No types (uses app.generic())	azure-functions-nodejs-extensions repo (ServiceBus pattern)	Medium - new extensions-kafka package	2
Java	KafkaTrigger annotation only, no SDK type	azure-functions-java-library Maven	Small-Medium - add KafkaRecord POJO	3
Python	KafkaEvent class in azure-functions-python-library	Converter registry in library	Medium - new KafkaRecord class + converter	4
PowerShell	No types (Hashtable via generic pipeline)	TriggerMetadata already exposes all metadata	Small - TriggerMetadata already works	5

Architecture

All languages follow the same pattern - the differences are only in language-specific type conversion.

Host (Kafka Extension): ConsumeResult to KafkaRecord (key/value as byte[]) to Serialize to ModelBindingData JSON (base64 key/value) to gRPC to Worker.
Worker (language-specific): Deserialize ModelBindingData JSON to Convert base64 to native bytes type to Expose as KafkaRecord.

Each language has an existing extensions package pattern that Kafka can follow.

Breaking Change Assessment

Change	Breaking?
Add KafkaRecord type to each language	Non-breaking (additive)
Existing KafkaEventData / KafkaEvent types	Unchanged (maintained)
Existing string / byte[] bindings	Unchanged
Host-side ModelBindingData for KafkaRecord	Non-breaking (new path)

All changes are non-breaking across all languages.

Helper Methods

To reduce user burden with raw bytes, each language package will provide helpers for common operations like JSON deserialization, string conversion, and header access.

[daankets]

@TsuyoshiUshio I support the latest change to the proposal (working for same customer). It's important to support the basic Kafka record. We can always layer additional processing on top if the base structure is complete and spec compliant. I would like to note that helper methods should not pull in additional dependencies that consumers might not use. These should be optional and it should be up to the consumer to include them if needed (for example to support protobuf or Avro schemas)

[KristofVDV]

@TsuyoshiUshio Your latest comment is indeed what would be beneficial to obtain the raw bytes of a Kafka record (or a batch of those).

However, one thing I'm not sure of: is the conversion using Base64 encoding required as stated in
Worker (language-specific): Deserialize ModelBindingData JSON to Convert base64 to native bytes type to Expose as KafkaRecord.?

Maybe this penalty of extra CPU-cycles and memory consumption can be avoided altogether, depending on the current internal workings and implementation of the existing KakaTrigger?

[TsuyoshiUshio]

Thank you @daankets for the support and @KristofVDV for the great question about Base64 overhead!

Serialization Format Decision: Protobuf (no Base64 needed)

@KristofVDV raises a valid concern. After investigating the internal architecture, I have good news: Base64 encoding is not required.

Why Base64 was in the original proposal

The original design used JSON as the serialization format inside ModelBindingData. Since JSON cannot represent raw binary, key/value bytes had to be Base64-encoded. This adds ~33% size overhead plus CPU cycles for encode/decode on every message.

The solution: use Protobuf instead of JSON

The gRPC ModelBindingData proto definition is:

message ModelBindingData {
    string version = 1;
    string source = 2;
    string content_type = 3;
    bytes content = 4;       // binary payload, format-agnostic
}

The content field is bytes - the Host treats it as an opaque pass-through. It does not inspect or parse the content. This means we can use any serialization format, not just JSON.

With Protobuf, key and value are native bytes fields - no Base64, no string conversion:

message KafkaRecordProto {
    string topic = 1;
    int32 partition = 2;
    int64 offset = 3;
    bytes key = 4;                    // native bytes, zero overhead
    bytes value = 5;                  // native bytes, zero overhead
    KafkaTimestampProto timestamp = 6;
    repeated KafkaHeaderProto headers = 7;
    optional int32 leader_epoch = 8;
}

message KafkaTimestampProto {
    int64 unix_timestamp_ms = 1;
    int32 type = 2;
}

message KafkaHeaderProto {
    string key = 1;
    bytes value = 2;                  // native bytes
}

No Host changes required

The Host simply copies ParameterBindingData.Content into ModelBindingData.content without interpreting it. Whether we send JSON or Protobuf, the Host code path is identical. No Host release dependency.

Only two components need changes (released together as extension packages):

• Kafka Extension (Host-side): serialize KafkaRecord to Protobuf
• Worker Extensions (per-language): deserialize Protobuf to native KafkaRecord

Performance comparison

Format	Key/Value encoding	Size overhead	CPU overhead	New dependency
JSON + Base64	String (base64)	+33%	encode + decode	None
Protobuf	Native bytes	None	Minimal	Google.Protobuf (already transitive via Confluent.SchemaRegistry.Serdes.Protobuf)

Cross-language support

Protobuf has mature, Google-maintained libraries for all Azure Functions languages:

• .NET: Google.Protobuf NuGet
• Node.js: protobufjs npm
• Java: protobuf-java Maven
• Python: protobuf pip

One .proto file generates type-safe code for all languages, ensuring serialization consistency.

Breaking changes

None. This uses a new source identifier (e.g. AzureKafkaRecord) and a new content_type (application/x-protobuf). Existing JSON-based paths for KafkaEventData remain completely untouched.

Addressing @daankets feedback on dependencies

Per your comment, helper methods will not pull in Protobuf/Avro/Schema Registry dependencies into the user's project. The Protobuf serialization is internal to the extension transport layer (Host Extension to Worker Extension). Users receive a plain KafkaRecord object with byte[] key/value. How they deserialize those bytes is entirely their choice.