Extend Expression Language to Access HTTP Headers and to Use Multiple Parameters #1512

Yoni-Weisberg · 2025-06-30T14:59:34Z

Description

This is an improvement including features #1508 #1507 #1506

…ders

…ders (SSE support)

…ders (Fix for missing header)

jfallows

@Yoni-Weisberg Thank you for this PR. I've left a few initial comments for you.

Please update the examples/http.kafka.crud/etc/zilla.yaml, examples/http.kafka.avro.json/etc/zilla.yaml and examples/sse.kafka.crud/etc/zilla.yaml configuration to demonstrate the changes and we'll go from there.

jfallows · 2025-07-04T06:31:23Z

...rc/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/cache/KafkaCachePartition.java

-                KafkaCachePaddedKeyFW.Builder paddedKeyBuilder = paddedKeyRW;
-                final int keySize = paddedKeyBuilder.key(k -> k.length(length).value(buffer, index, length)).sizeof();
-                paddedKeyBuilder.padding(logFile.buffer(), 0, paddedKeySize - keySize - SIZE_OF_INT);
-                KafkaCachePaddedKeyFW newPaddedKey = paddedKeyBuilder.build();
+                paddedKeyRW.rewrap();
+                KafkaKeyFW keyFW = keyRW.rewrap().length(length).value(buffer, index, length).build();
+                paddedKeyRW.key(keyFW);
+                final int keyAndMetadataSize = keyFW.sizeof() + KafkaCachePaddedKeyFW.FIELD_OFFSET_PADDING;
+                paddedKeyRW.padding(logFile.buffer(), 0, paddedKeySize - keyAndMetadataSize);
+                KafkaCachePaddedKeyFW newPaddedKey = paddedKeyRW.build();


IIUC, the intent of this change is to fix an issue with paddedKeyRW reuse, which can be simplified to the following:

Suggested change

KafkaCachePaddedKeyFW.Builder paddedKeyBuilder = paddedKeyRW;

final int keySize = paddedKeyBuilder.key(k -> k.length(length).value(buffer, index, length)).sizeof();

paddedKeyBuilder.padding(logFile.buffer(), 0, paddedKeySize - keySize - SIZE_OF_INT);

KafkaCachePaddedKeyFW newPaddedKey = paddedKeyBuilder.build();

paddedKeyRW.rewrap();

KafkaKeyFW keyFW = keyRW.rewrap().length(length).value(buffer, index, length).build();

paddedKeyRW.key(keyFW);

final int keyAndMetadataSize = keyFW.sizeof() + KafkaCachePaddedKeyFW.FIELD_OFFSET_PADDING;

paddedKeyRW.padding(logFile.buffer(), 0, paddedKeySize - keyAndMetadataSize);

KafkaCachePaddedKeyFW newPaddedKey = paddedKeyRW.build();

KafkaCachePaddedKeyFW.Builder paddedKeyBuilder = paddedKeyRW.rewrap();

final int keySize = paddedKeyBuilder.key(k -> k.length(length).value(buffer, index, length)).sizeof();

paddedKeyBuilder.padding(logFile.buffer(), 0, paddedKeySize - keySize - SIZE_OF_INT);

KafkaCachePaddedKeyFW newPaddedKey = paddedKeyBuilder.build();

logFile.writeBytes(position, newPaddedKey.buffer(), newPaddedKey.offset(), newPaddedKey.sizeof());

Agree?

Agree, I was gradually understanding the issue, so I guess that was a leftover from a previous attempt to solve it another way.
Anyway, I've fixed it.

jfallows · 2025-07-04T06:38:04Z

...n/java/io/aklivity/zilla/runtime/binding/sse/kafka/internal/config/SseKafkaWithResolver.java

 public final class SseKafkaWithResolver
 {
    private static final Pattern PARAMS_PATTERN = Pattern.compile("\\$\\{params\\.([a-zA-Z_]+)\\}");
+    private static final Pattern HEADERS_PATTERN = Pattern.compile("\\$\\{headers\\.([a-zA-Z_\\-]+)\\}");


What use case did you have in mind here for headers in sse?

Note: sse abstracts away http headers, leaving only schema, authority, path (and the lastId for reconnect).

Hi. This is correct if you're using a standard SSE client. However, if you want to consume messages as a stream in an application, you can use a simple HTTP client instead.
This is the use case we're working with, and we pass a custom header to filter the messages.
Anyway, it will not affect users who use standard SSE client.

I see, so you are implementing the SSE protocol from a simple HTTP client to give you more control over headers so you can apply HTTP request header values as filter criteria?

Do you have any specific need for the SSE protocol, or your use case more generally requires a streaming HTTP response with filter criteria coming from HTTP header values?

I don't have specific need for the SSE protocol. if it was possible I would rather use simple HTTP, but as far as I see there is no option for streaming over HTTP with filters

Okay, that makes sense thanks.

Suggest we remove the proposed changes to sse from this PR, since they break the abstraction, plus you don't need sse protocol per se. Instead let's consider adding support for application/json-lines in the http-kafka binding which is something we've been thinking to add anyway.

https://jsonlines.org/
Azure-Samples/azure-search-openai-demo#968 (review)

All data inside Zilla flows in streams, so even the HTTP GET many is currently streaming the Kafka messages until end of (Kafka) stream. It's just that the end of stream is currently triggered by catching up to live messages in Kafka, rather than continuing to send further live messages.

HttpKafkaProxyFactory.HttpFetchManyProxy.onKafkaData(...)
HttpKafkaProxyFactory.HttpFetchManyProxy.onKafkaEnd(...)

Note: unlike the application/json content type for merging a stream of Kafka messages in an HTTP response, the application/json-lines content type doesn't need any preamble or postamble around all the messages, just a newline separator between messages.

HTTP GET many also supports with filters to filter for matching Kafka messages in the response so this should work without requiring any further changes to filter the streamed response.

HttpKafkaProxyFactory.HttpFetchManyProxy.onHttpBegin(...)

Suggest we keep this PR scoped to just the expression language enhancements, and handle the http-kafka application/json-lines changes in a separate PR.

Hey,
In "streaming" I meant the response is blocking until client aborts the request (or until a given timeout). and also have an event-id to resume after disconnection. So how do you suggest to implement it with json-lines?
Also, I guess I don't need to revert the entire change in SSE (including the ability to append text to parameters, and to use identity), but only the header part, am I right?

@Yoni-Weisberg

In "streaming" I meant the response is blocking until client aborts the request (or until a given timeout). and also have an event-id to resume after disconnection. So how do you suggest to implement it with json-lines?

Agreed, event-id per message and last-event-id on recovering request are important concepts to retain in any streaming HTTP response format that wants to have a delivery guarantee.

There is some overlap with the concept of pagination across a larger data set, with a cursor position to advance to the next page, whereas streaming needs a cursor position to advance to the next message.

https://swagger.io/docs/specification/v3_0/links/#when-to-use-links

GET /items?cursor=Q1MjAwNz&limit=100

We can apply a similar approach to streaming. For example:

{ "metadata": { "next": "<cursor>" }, "data": {"name": "Alice", "age": 30} } { "metadata": { "next": "<cursor>" }, "data": {"name": "Bob", "age": 25} } { "metadata": { "next": "<cursor>" }, "data": {"name": "Charlie", "age": 35} }

Then on recovery, the next request would include a cursor query parameter with the opaque cursor value to pick up from where it left off.

This approach feels reasonably consistent with REST API pagination strategies, and therefore more easily consumed by HTTP clients.

Feedback welcome. 😄

Also, I guess I don't need to revert the entire change in SSE (including the ability to append text to parameters, and to use identity), but only the header part, am I right?

Yep, makes sense to keep the new work to support identity and a broader text pattern for sse parameters resolved into with, we just don't need the headers part as you say. 👍

Sounds good. So lets close this PR with the current content. And continue the discussion on another one. I made all the necessary fixes.

…ders (Adding license to test)

…e for headers substitution

…ders (Adding examples)

Yoni-Weisberg · 2025-07-15T06:43:45Z

Hey @jfallows, I updated the configuration of examples/http.kafka.crud/etc/zilla.yaml and examples/http.kafka.avro.json/etc/zilla.yaml.
However, there is no examples/sse.kafka.crud. If you want, I can add an example for demonstrating the use-case of using SSE via curl, but this is not a standard use-case as we discussed above.
Let me know what you want...

Yoni-Weisberg · 2025-07-29T09:57:41Z

@jfallows, I fixed the issue with HeaderFW in the test. Is there anything left that's preventing this from being merged?

# Conflicts: # runtime/binding-http-kafka/src/main/java/io/aklivity/zilla/runtime/binding/http/kafka/internal/config/HttpKafkaWithResolver.java # runtime/binding-sse-kafka/src/main/java/io/aklivity/zilla/runtime/binding/sse/kafka/internal/config/SseKafkaWithResolver.java

Yoni-Weisberg · 2025-08-10T14:02:52Z

Hi, @jfallows, Is there anything left that's preventing this from being merged?

jfallows · 2025-08-10T23:45:47Z

Hi, @jfallows, Is there anything left that's preventing this from being merged?

@Yoni-Weisberg
The build is passing but the http.kafka.avro.json example is no longer passing the automated test.
See https://github.com/Yoni-Weisberg/zilla/blob/develop/examples/http.kafka.avro.json/.github/test.sh for details.

Yoni-Weisberg · 2025-08-11T10:32:20Z

Fixed the test

Yoni-Weisberg added 2 commits June 30, 2025 17:54

Adding option to substitute multiple parameters and also use HTTP hea…

ec3c063

…ders

Merge remote-tracking branch 'origin/develop' into develop

92a469f

ankitk-me requested a review from jfallows July 1, 2025 11:45

Yoni-Weisberg added 2 commits July 2, 2025 11:58

Adding option to substitute multiple parameters and also use HTTP hea…

2b8ffaa

…ders (SSE support)

Adding option to substitute multiple parameters and also use HTTP hea…

4f24ab0

…ders (Fix for missing header)

jfallows reviewed Jul 4, 2025

View reviewed changes

Yoni-Weisberg added 3 commits July 6, 2025 10:10

Adding option to substitute multiple parameters and also use HTTP hea…

47d3d24

…ders (Adding license to test)

Code review note (simplifying transformation) + using equalsIgnoreCas…

dd3c0ad

…e for headers substitution

Adding option to substitute multiple parameters and also use HTTP hea…

c6a08e2

…ders (Adding examples)

Reverting headers replacement in SSE

d275f31

Yoni-Weisberg added 3 commits July 29, 2025 12:59

Merge remote-tracking branch 'origin/develop' into develop

99f6ab2

Fixing test HeaderFW issue

7cbb8bb

jfallows previously approved these changes Aug 10, 2025

View reviewed changes

Fix of http.kafka.avro.json docker-compose

629b6ae

Yoni-Weisberg dismissed jfallows’s stale review via 629b6ae August 11, 2025 08:57

Extend Expression Language to Access HTTP Headers and to Use Multiple Parameters #1512

Are you sure you want to change the base?

Extend Expression Language to Access HTTP Headers and to Use Multiple Parameters #1512

Uh oh!

Conversation

Yoni-Weisberg commented Jun 30, 2025

Description

Uh oh!

jfallows left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jfallows Jul 4, 2025

Choose a reason for hiding this comment

Uh oh!

Yoni-Weisberg Jul 9, 2025

Choose a reason for hiding this comment

Uh oh!

jfallows Jul 4, 2025

Choose a reason for hiding this comment

Uh oh!

Yoni-Weisberg Jul 9, 2025

Choose a reason for hiding this comment

Uh oh!

jfallows Jul 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Yoni-Weisberg Jul 13, 2025

Choose a reason for hiding this comment

Uh oh!

jfallows Jul 21, 2025

Choose a reason for hiding this comment

Uh oh!

Yoni-Weisberg Jul 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jfallows Jul 24, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Yoni-Weisberg Jul 27, 2025

Choose a reason for hiding this comment

Uh oh!

Yoni-Weisberg commented Jul 15, 2025

Uh oh!

Yoni-Weisberg commented Jul 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Yoni-Weisberg commented Aug 10, 2025

Uh oh!

jfallows commented Aug 10, 2025

Uh oh!

Yoni-Weisberg commented Aug 11, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

jfallows left a comment •

edited

Loading

jfallows Jul 11, 2025 •

edited

Loading

Yoni-Weisberg Jul 22, 2025 •

edited

Loading

jfallows Jul 24, 2025 •

edited

Loading

Yoni-Weisberg commented Jul 29, 2025 •

edited

Loading