Support streaming execution in the HttpClient #1085

Randgalt · 2023-07-23T16:52:06Z

Support end-to-end streaming. This is useful for endpoints that return
a large list of entities. Instead of needing to queue that entire list
in memory in both the server and client the entities are streamed.
This has even greater impact when there is a client in the middle:
client -> server A -> server B -> back to client.

The flow is:

Client sends request to server
Server begins to stream entities
Client reads entities one by one

This necessitates giving the client ownership of the internal
input stream.

Additionally, adds JsonStreamingResponse to make it simpler/convenient to stream lists
of JSON entities.

Randgalt · 2023-07-23T16:52:32Z

cc @electrum @findepi @losipiuk

Praveen2112

LGTM. License header is missing

Praveen2112 · 2023-07-27T15:45:21Z

http-client/src/main/java/io/airlift/http/client/JsonStreamingResponseHandler.java

@@ -0,0 +1,184 @@
+package io.airlift.http.client;


Do we need license header

In my recent airlift PRs @electrum has said we don't want license headers. The old header was Proofpoint and that doesn't apply anymore.

Praveen2112 · 2023-07-27T16:13:37Z

http-client/src/main/java/io/airlift/http/client/jetty/JettyHttpClient.java

+        if (isStreaming) {
+            JettyResponse jettyResponse;
+            try {
+                jettyResponse = new JettyResponse(response, listener.getInputStream());


Can we extract this above the if statement ? Do we need to wrap between try

I don't think that would help. In the non-streaming case it gets the entire value after allocating the JettyResponse. The streaming case doesn't do that.

http-client/src/main/java/io/airlift/http/client/JsonStreamingResponseHandler.java

Praveen2112 · 2023-07-27T16:24:56Z

http-client/src/test/java/io/airlift/http/client/AbstractHttpClientTest.java

+    }
+
+    @Test(timeOut = 10000)
+    public void testStreamingBlockingJson()


Can we have an additional test case where the server returns JSONObject and an JSONArray with an invalid pattern and an empty JSONArray

oskar-szwajkowski · 2023-07-28T08:16:41Z

http-client/src/main/java/io/airlift/http/client/JsonStreamingResponseHandler.java

+
+        JsonParser parser;
+        try {
+            parser = objectMapper.createParser(new InputStreamReader(response.getInputStream(), StandardCharsets.UTF_8));


shouldn't

new InputStreamReader(response.getInputStream(), StandardCharsets.UTF_8)

be moved to previous line using try-with-resources?

No - the need here is that the parser is not closed. The caller has the responsibility of closing.

oskar-szwajkowski · 2023-07-28T08:24:16Z

http-client/src/test/java/io/airlift/http/client/AbstractHttpClientTest.java

+        JsonCodec<Thing> codec = jsonCodec(Thing.class);
+
+        servlet.setResponseBody(json);
+        servlet.addResponseHeader("Content-Type", "application/json");


I wonder if support for application/x-ndjson is something to consider

We can have a new StreamingResponseHandler for that, adding it when Airlift's other JSON hander gets support.

findepi · 2023-07-31T07:03:16Z

json/src/main/java/io/airlift/json/JsonCodec.java

@@ -215,6 +215,11 @@ public byte[] toJsonBytes(T instance)
        }
    }

+    public ObjectMapper mapper()
+    {
+        return mapper;


Mapper is mutable (and even potentially shared io.airlift.json.JsonCodec#OBJECT_MAPPER_SUPPLIER). This breaks encapsulation. This is a no-go from my perspective.

I'll find another solution.

@findepi I've come up with an alternate. PTAL.

also return mapper.copy(); would do, right?

Add new feature to create a streaming parser over an InputStream from a JsonCodec

losipiuk · 2023-08-15T09:51:00Z

json/src/main/java/io/airlift/json/JsonCodecParser.java

+        return parser.nextLongValue(defaultValue);
+    }
+
+    public Boolean nextBooleanValue()


why Boolean here (not bool) and int and long above?

That's how it is in JsonParser (Jackson). But, I'll improve this.

losipiuk · 2023-08-15T10:07:58Z

http-client/src/main/java/io/airlift/http/client/jetty/JettyHttpClient.java

+                value = responseHandler.handle(request, new JettyStreamingResponse(jettyResponse, span, finalizer));
+            }
+            catch (RuntimeException e) {
+                throw new StreamingException(e, finalizer);


why pass finalizer through StreamingException instead calling it immediately here before exception is thrown?

tbh - I don't know what I was thinking. That path is not needed and I've removed and simplified it.

losipiuk · 2023-08-15T10:10:31Z

http-client/src/main/java/io/airlift/http/client/jetty/JettyStreamingResponse.java

+    }
+
+    @Override
+    public void close()


nit: guard from calling more than once.

losipiuk · 2023-08-15T10:11:06Z

http-client/pom.xml

@@ -158,6 +153,12 @@
            <scope>runtime</scope>
        </dependency>

+        <dependency>


Merge Tests for streaming execution into commit which adds support for streaming responses.

Support end-to-end streaming. This is useful for endpoints that return a large list of entities. Instead of needing to queue that entire list in memory in both the server and client the entities are streamed. This has even greater impact when there is a client in the middle: client -> server A -> server B -> back to client. The flow is: - Client sends request to server - Server begins to stream entities - Client reads entities one by one This necessitates giving the client ownership of the internal input stream. Additionally, adds JsonStreamingResponse to make it simpler/convenient to stream lists of JSON entities.

wendigo · 2024-03-09T05:54:49Z

@Randgalt please rebase

Randgalt · 2024-05-26T13:34:24Z

Closing in favor of #1174

Randgalt force-pushed the jordanz/http-client-streaming branch 3 times, most recently from 5b9a923 to ff758e3 Compare July 25, 2023 12:43

Praveen2112 reviewed Jul 27, 2023

View reviewed changes

oskar-szwajkowski reviewed Jul 28, 2023

View reviewed changes

Randgalt force-pushed the jordanz/http-client-streaming branch from ff758e3 to 98fa6c4 Compare July 28, 2023 08:38

Randgalt requested review from Praveen2112 and oskar-szwajkowski July 28, 2023 08:38

Randgalt force-pushed the jordanz/http-client-streaming branch 2 times, most recently from 47c8118 to 8352452 Compare July 28, 2023 09:06

findepi reviewed Jul 31, 2023

View reviewed changes

Randgalt force-pushed the jordanz/http-client-streaming branch from 8352452 to b13232d Compare July 31, 2023 09:09

Randgalt requested a review from findepi July 31, 2023 09:09

Randgalt force-pushed the jordanz/http-client-streaming branch from b13232d to 2e768dd Compare July 31, 2023 09:11

Randgalt added 2 commits July 31, 2023 10:13

Introduce JsonCodecParser

5acba45

Add new feature to create a streaming parser over an InputStream from a JsonCodec

Extract response validation for use by other handlers

e1b42ec

Randgalt force-pushed the jordanz/http-client-streaming branch from 2e768dd to d3b3b61 Compare July 31, 2023 09:13

losipiuk reviewed Aug 15, 2023

View reviewed changes

losipiuk approved these changes Aug 15, 2023

View reviewed changes

Randgalt force-pushed the jordanz/http-client-streaming branch from d3b3b61 to 1643c30 Compare August 17, 2023 05:26

Randgalt force-pushed the jordanz/http-client-streaming branch from 1643c30 to c2d5ce3 Compare August 17, 2023 05:33

Randgalt closed this May 26, 2024

Randgalt deleted the jordanz/http-client-streaming branch May 26, 2024 13:34

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support streaming execution in the HttpClient #1085

Support streaming execution in the HttpClient #1085

Randgalt commented Jul 23, 2023

Randgalt commented Jul 23, 2023

Praveen2112 left a comment

Praveen2112 Jul 27, 2023

Randgalt Jul 27, 2023

Praveen2112 Jul 27, 2023

Randgalt Jul 28, 2023

Praveen2112 Jul 27, 2023

oskar-szwajkowski Jul 28, 2023

Randgalt Jul 28, 2023

oskar-szwajkowski Jul 28, 2023

Randgalt Jul 28, 2023

findepi Jul 31, 2023

Randgalt Jul 31, 2023

Randgalt Jul 31, 2023

oskar-szwajkowski Jul 31, 2023

losipiuk Aug 15, 2023 •

edited

Randgalt Aug 15, 2023

losipiuk Aug 15, 2023

Randgalt Aug 17, 2023 •

edited

losipiuk Aug 15, 2023

losipiuk Aug 15, 2023

wendigo commented Mar 9, 2024

Randgalt commented May 26, 2024

Support streaming execution in the HttpClient #1085

Support streaming execution in the HttpClient #1085

Conversation

Randgalt commented Jul 23, 2023

Randgalt commented Jul 23, 2023

Praveen2112 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

losipiuk Aug 15, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Randgalt Aug 17, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wendigo commented Mar 9, 2024

Randgalt commented May 26, 2024

losipiuk Aug 15, 2023 •

edited

Randgalt Aug 17, 2023 •

edited