Describe the desired outcome from the user's perspective
As a binding author, I want a ${...} expression to be able to reference the Kafka record currently being produced — its key, or a named header — so that a produced key or header value can be derived from another part of the same in-flight message, the same way existing expression contexts already let a produced value be derived from route params, auth context, or an echoed identifier.
Naming collision to resolve before implementing. KafkaTopicTransformsConfig's extractKey/extractHeaders already accept literal strings shaped like ${message.key.id} / ${message.value.correlationId} — but as a plain literal match against a source path (KafkaExtractTransform compares by path.equals(source.getPath())), not a parsed ${} expression: there is no ${}-aware evaluator, no startup validation of the contents, and no header resolution behind that syntax today. This issue's message.key/message.headers. vocabulary would collide visually with that existing, differently-behaved usage. Resolve the collision (rename this issue's expression root, or reconcile the two mechanisms into one) before implementing, rather than shipping two different things that parse the same-looking string differently.
Acceptance criteria
- A new expression root, evaluated against the record being produced, exposing at minimum:
message.key — the Kafka record keymessage.headers.<name> — a named header's value
- Usable anywhere a binding's config currently accepts a plain string for a produced key or header value, alongside the existing
${params.X} (HTTP path-parameter substitution, HttpKafkaWithResolver), ${guarded['...'].identity}/${guarded['...'].attributes} (auth context), ${correlationId}, and ${idempotencyKey} contexts already resolved there — this is a peer of those, not a replacement
- Resolved once per message, at produce time, against whichever record is currently being produced — never cached or reused across messages, and never re-evaluated after produce; whatever consumes the resolved value at produce time is responsible for persisting it if a later reader needs the same resolved string
- A configured value with no
${...} in it is passed through unresolved — this must not become mandatory templating machinery for the common static-value case
- Startup validation, not just runtime resolution. A configured string containing
${ must be syntactically well-formed and reference only a supported path — message.key or message.headers.<name> — checked once at startup, independent of any particular message. Reject anything else with a diagnostic naming the field and the invalid expression, the same validation posture the existing ${params.X} context already applies
- This is narrower than, and distinct from, any resolvability question that must wait until a specific message arrives (e.g. a referenced header turning out to be absent on one particular message) — syntax and referenced-path validity are startup concerns; whether a given message actually carries that header is not
- Verify against
HttpKafkaWithResolver's existing pattern-matched expression resolution (runtime/binding-http-kafka/.../internal/config/HttpKafkaWithResolver.java) for whatever shared expression-evaluation infrastructure already exists, and extend or mirror it rather than building a second, parallel evaluator
- Unit coverage:
message.key resolution, message.headers.<name> resolution for a present and an absent header, a plain string with no ${...} passed through unchanged, a malformed expression rejected at startup, a syntactically valid expression referencing an unsupported path rejected at startup, and two messages with different message.key values on the same schema/field producing different resolved values.
Additional context
Where this most likely lives. message.key/message.headers are Kafka-specific concepts, and the existing ${params.X}/${guarded[...]}/${correlationId}/${idempotencyKey} contexts are each hand-implemented per binding as hardcoded pattern/matcher pairs rather than registered against any shared pluggable "expression context" abstraction (no such abstraction currently exists in the engine) — so a message context most likely follows the same shape: a new pattern/matcher pair added to whichever binding's with-resolver needs it, mirroring HttpKafkaWithResolver, not a new engine-level registration point. Confirm this against the binding actually consuming it before assuming a location, and against the naming collision noted above.
Depends on nothing new architecturally — this is an expression-evaluation addition only, independent of any particular consumer's config shape.
Describe the desired outcome from the user's perspective
As a binding author, I want a
${...}expression to be able to reference the Kafka record currently being produced — its key, or a named header — so that a produced key or header value can be derived from another part of the same in-flight message, the same way existing expression contexts already let a produced value be derived from route params, auth context, or an echoed identifier.Naming collision to resolve before implementing.
KafkaTopicTransformsConfig'sextractKey/extractHeadersalready accept literal strings shaped like${message.key.id}/${message.value.correlationId}— but as a plain literal match against a source path (KafkaExtractTransformcompares bypath.equals(source.getPath())), not a parsed${}expression: there is no${}-aware evaluator, no startup validation of the contents, and no header resolution behind that syntax today. This issue'smessage.key/message.headers.vocabulary would collide visually with that existing, differently-behaved usage. Resolve the collision (rename this issue's expression root, or reconcile the two mechanisms into one) before implementing, rather than shipping two different things that parse the same-looking string differently.Acceptance criteria
message.key— the Kafka record keymessage.headers.<name>— a named header's value${params.X}(HTTP path-parameter substitution,HttpKafkaWithResolver),${guarded['...'].identity}/${guarded['...'].attributes}(auth context),${correlationId}, and${idempotencyKey}contexts already resolved there — this is a peer of those, not a replacement${...}in it is passed through unresolved — this must not become mandatory templating machinery for the common static-value case${must be syntactically well-formed and reference only a supported path —message.keyormessage.headers.<name>— checked once at startup, independent of any particular message. Reject anything else with a diagnostic naming the field and the invalid expression, the same validation posture the existing${params.X}context already appliesHttpKafkaWithResolver's existing pattern-matched expression resolution (runtime/binding-http-kafka/.../internal/config/HttpKafkaWithResolver.java) for whatever shared expression-evaluation infrastructure already exists, and extend or mirror it rather than building a second, parallel evaluatormessage.keyresolution,message.headers.<name>resolution for a present and an absent header, a plain string with no${...}passed through unchanged, a malformed expression rejected at startup, a syntactically valid expression referencing an unsupported path rejected at startup, and two messages with differentmessage.keyvalues on the same schema/field producing different resolved values.Additional context
Where this most likely lives.
message.key/message.headersare Kafka-specific concepts, and the existing${params.X}/${guarded[...]}/${correlationId}/${idempotencyKey}contexts are each hand-implemented per binding as hardcoded pattern/matcher pairs rather than registered against any shared pluggable "expression context" abstraction (no such abstraction currently exists in the engine) — so amessagecontext most likely follows the same shape: a new pattern/matcher pair added to whichever binding'swith-resolver needs it, mirroringHttpKafkaWithResolver, not a new engine-level registration point. Confirm this against the binding actually consuming it before assuming a location, and against the naming collision noted above.Depends on nothing new architecturally — this is an expression-evaluation addition only, independent of any particular consumer's config shape.