feat(kafka): support Avro-encoded message key via schema registry by burnerlee · Pull Request #1201 · timeplus-io/proton

burnerlee · 2026-06-21T07:03:30Z

PR checklist:

Did you run ClangFormat? — new code follows surrounding style; all edited files are in Proton-specific paths
Did you separate headers to a different section in existing community code base? — new includes added in alphabetical order within existing include blocks
Did you surround proton: starts/ends for new code in existing community code base? — not applicable, all changes are in src/Storages/ExternalStream/Kafka/ which is Proton-specific code

Please write user-readable short description of the changes:

Adds a new message_key_schema_name setting to Kafka external streams. When set, Proton decodes the Kafka message key using the Confluent Avro wire format (5-byte wire prefix: magic byte + 4-byte schema ID) and emits the decoded value as a JSON string into _tp_message_key.

Motivation: Producers that use Confluent Schema Registry commonly Avro-encode message keys, not just values. Reading _tp_message_key on such topics previously returned raw bytes.

Behaviour:

message_key_schema_name names an Avro schema subject in the schema registry
Requires kafka_schema_registry_url (validated at CREATE EXTERNAL STREAM time)
_tp_message_key column must be String or Nullable(String) in Avro key mode (validated at CREATE time)
Schema lookup is cached via the existing KafkaSchemaRegistryForAvro mechanism
On read: Avro binary key is decoded into a GenericDatum, then re-encoded as Avro JSON (not plain JSON — union types use Avro union encoding, bytes fields are base64). The result is stored as a string in _tp_message_key
On write: throws NOT_IMPLEMENTED if message_key_schema_name is set — Avro-encoding keys on write is not yet supported. Plain string keys via _tp_message_key without message_key_schema_name continue to work
Empty key (key_len == 0) returns NULL for Nullable(String) columns, empty string for String columns

Example:

CREATE EXTERNAL STREAM orders (
    _tp_message_key String,
    payload         String
) SETTINGS
    type                       = 'kafka',
    brokers                    = 'localhost:9092',
    topic                      = 'orders',
    data_format                = 'JSONEachRow',
    kafka_schema_registry_url  = 'http://localhost:8081',
    message_key_schema_name    = 'orders-key';

SELECT _tp_message_key FROM orders;
-- Returns Avro JSON string, e.g. {"user_id": 42, "tenant": "acme"}

Closes #915

Added a new setting `message_key_schema_name` to Kafka external streams. When set, the Kafka message key is decoded using the Confluent Avro wire format (5-byte prefix: 0x00 magic + 4-byte schema ID) against the named subject in the configured schema registry, and the result is emitted as a JSON string into the `_tp_message_key` virtual column. Changes: - ExternalStreamSettings.h: add `message_key_schema_name` String setting - Kafka.cpp: validate setting requires `kafka_schema_registry_url`; restrict `_tp_message_key` column to String/FixedString when Avro key mode is active; build and store a `KafkaSchemaRegistryForAvro` for the key subject in the constructor; pass registry and subject name to each KafkaSource - Kafka.h: add `avro_key_schema_registry` member; include KafkaSchemaRegistryForAvro - KafkaSource.h: add `avro_key_schema_registry` and `avro_key_schema_subject` members; add constructor params; declare `decodeAvroKey()` private method - KafkaSource.cpp: implement `decodeAvroKey()` using AvroDeserializer against a single String column header; branch both the typed (String/FixedString switch case) and untyped (RESERVED_MESSAGE_KEY fallback) virtual column paths to call `decodeAvroKey()` when the registry is present, raw bytes otherwise

yuzifeng1984 · 2026-06-22T03:20:59Z

Need implement KafkaSink related things to encode message key or throw an exception on write if currently not supported.

- Return empty string (not decode) when key_len == 0 and column is non-nullable - Remove unused avro_key_schema_subject param/member — schema ID comes from key bytes - Use angle-bracket includes for Formats headers in Kafka.cpp

Replace broken AvroDeserializer approach (which mapped Avro fields by column name and always returned empty string) with GenericDatum round-trip: binary Avro -> GenericDatum -> Avro JSON string. Also fix include style in KafkaSource.h (angle brackets).

Writing _tp_message_key as Confluent Avro binary is not yet implemented — only decoding on read is supported. Throw a clear error on write with guidance to use a plain string key instead.

burnerlee · 2026-06-25T04:27:25Z

@yuzifeng1984 how can I test my changes in local - can you share your testing setup?

yuzifeng1984 · 2026-06-25T12:47:31Z

Hi @burnerlee, I use Redpanda for local testing. You may use its web UI to configure Kafka things including schema
https://docs.redpanda.com/labs/docker-compose/single-broker/

burnerlee · 2026-06-25T13:43:47Z

Hi @burnerlee, I use Redpanda for local testing. You may use its web UI to configure Kafka things including schema

https://docs.redpanda.com/labs/docker-compose/single-broker/

Thanks

yuzifeng1984 · 2026-06-30T06:48:22Z

@burnerlee The PR looks good.

Can you also update the the kafka_schema_registry_url setting? Just in case it is only used for key but not message body.

Blocker — Format validation conflict makes the documented use case impossible

File: src/Storages/ExternalStream/Kafka/Kafka.cpp:264-279 (verifySettings)

The pre-existing check unconditionally rejects any kafka_schema_registry_url value when data_format is not 'Avro' or 'ProtobufSingle':

if (!new_settings->kafka_schema_registry_url.value.empty())
{
const auto & format = new_settings->data_format.value;
const bool format_supported = format == "ProtobufSingle" || format == "Avro";
if (!format_supported)
throw Exception(...);
}

The PR's own example in the description uses data_format = 'JSONEachRow' with kafka_schema_registry_url set — exactly the case where a user wants Avro-encoded keys but JSON-encoded values. This CREATE EXTERNAL STREAM would throw at creation time, before reaching any of the new code.

Fix: Condition the format check on whether the registry is used for value decoding. When message_key_schema_name is non-empty and data_format is not itself Avro/Protobuf, the registry is used solely for key decoding and the format
constraint should not apply. For example:

const bool using_registry_for_value = new_settings->message_key_schema_name.value.empty();
if (!new_settings->kafka_schema_registry_url.value.empty() && using_registry_for_value)
{
// existing format check ...
}

burnerlee · 2026-06-30T06:52:47Z

@burnerlee The PR looks good.

Can you also update the the kafka_schema_registry_url setting? Just in case it is only used for key but not message body.

Blocker — Format validation conflict makes the documented use case impossible

File: src/Storages/ExternalStream/Kafka/Kafka.cpp:264-279 (verifySettings)

The pre-existing check unconditionally rejects any kafka_schema_registry_url value when data_format is not 'Avro' or 'ProtobufSingle':

if (!new_settings->kafka_schema_registry_url.value.empty())

{

const auto & format = new_settings->data_format.value;

const bool format_supported = format == "ProtobufSingle" || format == "Avro";

if (!format_supported)
  throw Exception(...);
}

The PR's own example in the description uses data_format = 'JSONEachRow' with kafka_schema_registry_url set — exactly the case where a user wants Avro-encoded keys but JSON-encoded values. This CREATE EXTERNAL STREAM would throw at creation time, before reaching any of the new code.

Fix: Condition the format check on whether the registry is used for value decoding. When message_key_schema_name is non-empty and data_format is not itself Avro/Protobuf, the registry is used solely for key decoding and the format

constraint should not apply. For example:

const bool using_registry_for_value = new_settings->message_key_schema_name.value.empty();

if (!new_settings->kafka_schema_registry_url.value.empty() && using_registry_for_value)

{
  // existing format check ...
}

Sure let me update

Previously `kafka_schema_registry_url` required the message body format to be Avro or ProtobufSingle. With Avro-encoded keys, users may want a non-Avro body format (e.g. JSONEachRow) but still need the schema registry for the key. Allow the registry URL when either the body format needs it or `message_key_schema_name` is set. Also fix a latent bug where the post-update validation used the old `settings` member instead of `new_settings` when checking `message_key_schema_name`.

burnerlee · 2026-06-30T08:44:31Z

@yuzifeng1984 updated. Could you check?

burnerlee added 2 commits June 21, 2026 12:31

Merge branch 'develop' into feat/kafka-avro-message-key

dc0e8d3

chenziliang requested a review from yuzifeng1984 June 21, 2026 17:10

yuzifeng1984 reviewed Jun 22, 2026

View reviewed changes

Comment thread src/Storages/ExternalStream/Kafka/KafkaSource.cpp Outdated

yuzifeng1984 reviewed Jun 22, 2026

View reviewed changes

Comment thread src/Storages/ExternalStream/Kafka/KafkaSource.h Outdated

yuzifeng1984 reviewed Jun 22, 2026

View reviewed changes

Comment thread src/Storages/ExternalStream/Kafka/KafkaSource.cpp Outdated

yuzifeng1984 reviewed Jun 22, 2026

View reviewed changes

Comment thread src/Storages/ExternalStream/Kafka/Kafka.cpp Outdated

burnerlee added 4 commits June 25, 2026 09:56

Merge branch 'develop' into feat/kafka-avro-message-key

ada9b7f

fix(kafka): address review nits for Avro key decoding

981846b

- Return empty string (not decode) when key_len == 0 and column is non-nullable - Remove unused avro_key_schema_subject param/member — schema ID comes from key bytes - Use angle-bracket includes for Formats headers in Kafka.cpp

fix(kafka): throw NOT_IMPLEMENTED when writing Avro-encoded message key

3e2b331

Writing _tp_message_key as Confluent Avro binary is not yet implemented — only decoding on read is supported. Throw a clear error on write with guidance to use a plain string key instead.

burnerlee force-pushed the feat/kafka-avro-message-key branch from 3c0661d to 6cde6ce Compare June 30, 2026 08:41

wrap avro parsing exception

f95ed9b

yuzifeng1984 merged commit b73dba7 into timeplus-io:develop Jul 1, 2026
10 checks passed

yokofly mentioned this pull request Jul 2, 2026

chore: sync internal changes (py314 + Kafka Avro message-key) #1205

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat(kafka): support Avro-encoded message key via schema registry#1201

feat(kafka): support Avro-encoded message key via schema registry#1201
yuzifeng1984 merged 8 commits into
timeplus-io:developfrom
burnerlee:feat/kafka-avro-message-key

burnerlee commented Jun 21, 2026 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

yuzifeng1984 commented Jun 22, 2026

Uh oh!

Uh oh!

burnerlee commented Jun 25, 2026

Uh oh!

yuzifeng1984 commented Jun 25, 2026 •

edited

Loading

Uh oh!

burnerlee commented Jun 25, 2026

Uh oh!

yuzifeng1984 commented Jun 30, 2026

Uh oh!

burnerlee commented Jun 30, 2026

Uh oh!

burnerlee commented Jun 30, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

burnerlee commented Jun 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

yuzifeng1984 commented Jun 22, 2026

Uh oh!

Uh oh!

burnerlee commented Jun 25, 2026

Uh oh!

yuzifeng1984 commented Jun 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

burnerlee commented Jun 25, 2026

Uh oh!

yuzifeng1984 commented Jun 30, 2026

Uh oh!

burnerlee commented Jun 30, 2026

Uh oh!

burnerlee commented Jun 30, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

burnerlee commented Jun 21, 2026 •

edited

Loading

yuzifeng1984 commented Jun 25, 2026 •

edited

Loading