Repository navigation
Add opt-in native HTTP streaming support - #904
Draft
Ahmed Muhsin (ahmedmuhsin) wants to merge 8 commits into
Draft
Ahmed Muhsin (ahmedmuhsin) wants to merge 8 commits into
Ahmed Muhsin (ahmedmuhsin) wants to merge 8 commits into
Conversation
Introduce a new com.microsoft.azure.functions.worker.http package that lays the foundation for receiving HTTP-triggered invocations directly from the Functions host via the HttpUri capability (parity with the Go, Python, and .NET isolated workers). * HttpProxyServer wraps the JDK built-in com.sun.net.httpserver.HttpServer. Binds to 127.0.0.1 on an ephemeral port, dispatches requests on a cached executor mirroring the existing gRPC dispatch pool (unbounded growth, daemon threads, 15s shutdown drain). No new third-party dependencies. * HttpInvocationCoordinator synchronizes the two halves of each invocation - the HTTP request and the gRPC InvocationRequest - via a ConcurrentMap of per-invocation slots keyed by invocationId. Each slot exposes CompletableFutures for HTTP arrival, gRPC arrival, and completion. * ProxyConfig captures bind address/port only. No body-size or timeout caps; matches Go/Python/.NET behavior of relying on the platform for those guards. Pure infrastructure commit: nothing is wired into the worker lifecycle yet. 15 new unit tests cover both rendezvous orderings, slot release/failure semantics, ephemeral port binding, request routing, and close-after-start behavior.
Adds HttpProxyHandler and HttpBodyBridge to connect the embedded HTTP proxy server to the existing gRPC invocation dispatch path. HttpProxyHandler parks each forwarded request on HttpInvocationCoordinator (keyed by x-ms-invocation-id), then blocks on the slot's completion future so the connection stays open until the gRPC side has fully written the response. Missing header / duplicate arrival / invocation failure paths return appropriate HTTP error responses. HttpBodyBridge handles the request body <-> protobuf body conversion and writes RpcHttp responses back to HttpExchange. Body classification mirrors the host's PopulateBody behavior (application/json -> JSON TypedData, text/form-encoded/xml/js -> string, otherwise bytes) so downstream RpcHttpRequestDataSource logic continues to work unchanged. API refactor: HttpInvocationCoordinator.register*Arrival() now returns the HttpInvocationSlot itself instead of the opposite side's future, because the HTTP handler needs to await completion() (not grpcArrival). HttpInvocationSlot is now public to expose this surface. Tests: - HttpBodyBridgeTest: 15 tests covering content-type classification, charset handling, body enrichment, response writing, and status parsing. - HttpProxyHandlerTest: 4 tests covering missing-header, successful-completion, invocation-failure, and duplicate-arrival paths. - HttpInvocationCoordinatorTest: updated for the new slot-returning API; added completionFutureResolvesOnRelease. No worker wiring yet -- the proxy server is not started and no capability is advertised. This change just stages the parts that the next commit will glue into JavaWorkerClient + the request handlers.
Wires HttpProxyServer and HttpInvocationCoordinator into the worker lifecycle. When the FUNCTIONS_JAVA_DISABLE_HTTP_PROXY environment variable is not set to true: - JavaWorkerClient constructs a single HttpProxyServer and a single HttpInvocationCoordinator, shared across all handler instances. - WorkerInitRequestHandler starts the proxy server during worker init, advertises HttpUri = "http://127.0.0.1:<port>" and RequiresRouteParameters = "true" so the host forwards HTTP traffic for HTTP-triggered functions directly to the worker. - InvocationRequestHandler detects requests that contain an HTTP input binding, rendezvous-es with the matching HttpExchange via the coordinator, folds the request body into the gRPC InvocationRequest, invokes the user function, writes the RpcHttp response back to the exchange, and releases the slot. Failures fail the slot so the HTTP handler returns 500. - Non-HTTP invocations (queue, timer, etc.) bypass the coordinator and execute on the existing gRPC-only path. - JavaWorkerClient.close() stops the proxy server before tearing down the gRPC peer so in-flight HTTP handlers drain cleanly. Fixes #781: requests sent with Transfer-Encoding: chunked previously arrived at the worker with an empty body because of a null-comparison bug in the host's RpcHttp content-length handling. Moving body delivery to HTTP (matching Go / Python / .NET isolated workers) sidesteps that host bug entirely. The escape hatch FUNCTIONS_JAVA_DISABLE_HTTP_PROXY=true restores the old behavior if anything regresses. Test verification: full test suite (mvn -pl . test) passes 111/111 tests including 35 new tests across HttpProxyServer, coordinator, handler, and body bridge.
…am() When a function returns an HttpResponseMessage whose body is an InputStream or HttpResponseMessage.IOConsumer<OutputStream>, the HTTP proxy dispatch path now writes the body directly to the underlying HttpExchange response stream using chunked transfer-encoding instead of buffering the entire payload through a protobuf TypedData. This enables Server-Sent Events, large file downloads, and other long-lived streaming responses without first materializing the whole body in memory. Design highlights: - RpcHttpDataTarget.toRpcHttpData detects streaming bodies and skips the RpcUnspecifiedDataTarget serialization step, leaving the RpcHttp body field unset so the status + headers envelope still flows through the existing pipeline unchanged. - BindingDataStore exposes getHttpResponseRawBody() so the dispatch layer can recover the raw (unserialized) body without ripping apart the protobuf reply. - JavaFunctionBroker.invokeMethodForHttpProxy is a new method overload (not a modification of invokeMethod) that returns an HttpInvocationOutcome containing both the protobuf reply and the raw body. The original invokeMethod is left untouched for backward compatibility with the existing gRPC dispatch path. - HttpBodyBridge.writeStreamingResponse(InputStream) and writeStreamingResponse(IOConsumer) handle the actual streaming write; both use sendResponseHeaders(status, 0) to select chunked transfer-encoding (or close-delimited for HTTP/1.0 clients) and close/flush the streams reliably in try-with-resources. - InvocationRequestHandler.executeProxiedHttp dispatches based on the raw body type, falling back to the existing buffered writer when the body is not a streaming type. The pom.xml is bumped to consume azure-functions-java-core-library 1.4.0-SNAPSHOT, which adds the bodyStream() overloads and the IOConsumer functional interface. Tests added: 5 new HttpBodyBridge streaming tests, 8 new RpcHttpDataTarget tests, 6 new BindingDataStore tests. Total worker test suite: 130 passing (was 111).
When a function declares its trigger parameter as HttpRequestMessage<InputStream> (or any InputStream subtype), the HTTP proxy dispatch path now hands the function the live request body stream from the HttpExchange instead of eagerly buffering the entire payload into a protobuf TypedData. This enables streaming uploads, multi-gigabyte request bodies, and proxy-style scenarios where the function consumes the body incrementally without first materializing it in memory. Design highlights: - RpcHttpRequestDataSource captures a per-thread HttpExchange via a ThreadLocal at construction time; its HTTP_DATA_OPERATIONS resolver detects InputStream-assignable type arguments and returns exchange.getRequestBody() directly when the exchange is present. Otherwise the existing buffered bodyDataSource path is preserved unchanged. - JavaFunctionBroker.methodHasStreamingHttpBody(id) introspects the function signature so the dispatch layer can decide upfront whether to skip the eager body read. - InvocationRequestHandler.executeProxiedHttp consults the pre-check; when true it skips HttpBodyBridge.enrichRequestWithBody (which would consume the stream), installs the exchange on the ThreadLocal across the broker invocation, and clears it in a finally block so the thread can be safely reused. The existing buffered path is the default for all other signatures (HttpRequestMessage<String>, HttpRequestMessage<byte[]>, raw HttpRequestMessage, JSON-bound POJOs) and behaves identically to before. Non-proxy gRPC dispatch is unaffected. Tests added: - RpcHttpRequestDataSourceTest: 3 streaming-input cases (live stream handed back when exchange captured; ClassCastException on fallthrough when no exchange; ThreadLocal cleared correctly). - JavaFunctionBrokerStreamingTest: 6 signature-introspection cases (InputStream, InputStream subtype, String, raw, non-HTTP, unknown id).
Wires HttpProxyServer + HttpProxyHandler + HttpInvocationCoordinator +
InvocationRequestHandler together with a mocked JavaFunctionBroker and
exercises seven scenarios end-to-end:
1. buffered request / buffered response (sanity for the existing path)
2. buffered request / streaming InputStream response
3. buffered request / streaming IOConsumer (SSE) response
4. streaming request / buffered response (validates the live exchange
is exposed via the per-thread side channel and the body is NOT
folded into the protobuf envelope)
5. full streaming (request + response) (echoes uploaded bytes back)
6. coordinator slot is released after a successful invocation
7. broker exception surfaces as an HTTP 500 to the client
The tests fire real HTTP requests against an embedded HttpServer on an
ephemeral port and drive the simulated gRPC arrival on the test thread,
exercising the rendezvous, body propagation, and response writeback in
their integrated form.
Supporting library-side changes (no behavior impact on the production
path):
- JavaFunctionBroker.HttpInvocationOutcome: constructor is now public
so tests outside the broker package can construct the DTO without
reflection. It was already a plain DTO with public getters.
- RpcHttpRequestDataSource.currentExchange(): public accessor for the
per-thread HttpExchange installed by setCurrentExchange(). Used by
the integration tests to assert that the dispatch layer correctly
installs and clears the side channel around the broker invocation.
- HttpInvocationCoordinator.activeInvocationCount(): widened from
package-private to public so the integration tests (in the handler
package) can assert slot release.
…ng when proxy is unavailable Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Issue #781 | Companion API PR #55
Issue describing the changes in this PR
Related to #781. Replaces the implementation in #877 with native HTTP streaming work rebuilt on current
dev.Pull request checklist
release_notes.mdAdditional information
Why the change
Native HTTP transport lets Java functions read request bodies and stream responses without buffering them through gRPC, while preserving the legacy path for applications that do not opt in.
Special things to note
FUNCTIONS_JAVA_ENABLE_HTTP_STREAMING=true;FUNCTIONS_JAVA_DISABLE_HTTP_PROXY=truetakes precedence.1.4.0-SNAPSHOTis a local integration dependency. The host's HTTP/1.1 upload behavior requires staging unread request bytes to temporary disk, so this is not a full-duplex guarantee.mvn -o packagepassed with 178 tests, and the packaged worker passed a local Functions host smoke run covering large uploads, progressive response delivery, metadata, failure handling and concurrent correlations. JDK 8 runtime and deployed/cloud E2E coverage have not been run.Change outline
The HTTP request and gRPC invocation arrive independently. The coordinator matches them by invocation ID, then the worker invokes the function and completes HTTP output on the native exchange.