From c420781ba8f114f9d8ba82fba9245becd8336044 Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Wed, 9 Sep 2026 23:39:04 +0000 Subject: [PATCH 1/2] feat(insight): add HTTP workflow-insight exporter Port the JS workflow-insight HttpExporter to Java: POST/PUT each record to any HTTP(S) endpoint or webhook as application/json. - Builder config mirrors the JS contract: url (required, validated absolute http/https), method (POST default / PUT), custom headers, timeoutMs (default 10000), operationsFormat (ARRAY/BY_NAME/BOTH), optional maxRecordSizeBytes (no default). - Transport uses the JDK's own java.net.http.HttpClient, so no new dependency is added. An injectable HttpSender seam (default JdkHttpSender) makes the exporter mockable; render()/export() share one shaping path per the InsightExporter size-limiter contract. - Non-2xx responses throw and are contained by the plugin's existing per-exporter isolation, so a bad endpoint never disrupts execution. - Deterministic unit tests cover the mock seam and a real in-process com.sun.net.httpserver server; README documents the exporter. No changes to sdk core or the WorkflowInsight plugin wiring. --- insight-plugin/README.md | 49 ++- .../insight/exporters/HttpExporter.java | 218 +++++++++++++ .../durable/insight/exporters/HttpSender.java | 44 +++ .../insight/exporters/JdkHttpSender.java | 71 +++++ .../durable/insight/HttpExporterTest.java | 291 ++++++++++++++++++ 5 files changed, 667 insertions(+), 6 deletions(-) create mode 100644 insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpExporter.java create mode 100644 insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpSender.java create mode 100644 insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/JdkHttpSender.java create mode 100644 insight-plugin/src/test/java/software/amazon/lambda/durable/insight/HttpExporterTest.java diff --git a/insight-plugin/README.md b/insight-plugin/README.md index f626ad830..4481f8143 100644 --- a/insight-plugin/README.md +++ b/insight-plugin/README.md @@ -28,13 +28,14 @@ DurableConfig config = DurableConfig.builder() Exporters: `LambdaLogExporter` (default; writes the `operationsByName` map to stdout → CloudWatch), `S3Exporter` (canonical `operations` array, one object per execution), -`CloudWatchLogsExporter` (PutLogEvents to a specific log group, `operationsByName` map). Implement -`InsightExporter` for custom sinks. +`CloudWatchLogsExporter` (PutLogEvents to a specific log group, `operationsByName` map), +`HttpExporter` (POST/PUT each record to any HTTP(S) endpoint or webhook, choice of operations shape). +Implement `InsightExporter` for custom sinks. -`LambdaLogExporter` needs no extra dependency. The AWS SDK service modules used by the remote -exporters are optional so applications that use only Lambda logs do not package them. Add the module -for each remote exporter you configure, using the AWS SDK for Java 2.x version managed by your -application: +`LambdaLogExporter` and `HttpExporter` need no extra dependency — the HTTP exporter uses the JDK's own +HTTP client. The AWS SDK service modules used by the remote exporters are optional so applications +that use only Lambda logs or HTTP do not package them. Add the module for each remote exporter you +configure, using the AWS SDK for Java 2.x version managed by your application: ```xml @@ -52,6 +53,42 @@ application: ``` +### HttpExporter + +`HttpExporter` POSTs (or PUTs) each record to any HTTP(S) endpoint or webhook as a JSON body with +`Content-Type: application/json`. It uses the JDK's built-in HTTP client, so it adds **no dependency**. +It mirrors the JS `HttpExporter`. + +```java +.addExporter(HttpExporter.builder() + .url("https://collector.example.com/workflow-insight") // required, absolute http/https URL + .method(HttpExporter.Method.POST) // POST (default) or PUT + .addHeader("Authorization", "Bearer " + token) // optional auth / API-key headers + .timeoutMs(10_000) // per-request timeout, default 10s + .operationsFormat(HttpExporter.OperationsFormat.ARRAY) // ARRAY (default) | BY_NAME | BOTH + .build()) +``` + +Behavior: + +- **Endpoint / method / headers.** Records are sent to `url` with `method` (POST or PUT — use PUT for + endpoints that upsert by URL path). Your headers are merged on top of the fixed + `Content-Type: application/json`. +- **Auth.** There is no built-in auth scheme; pass whatever your endpoint expects as a header + (`Authorization: Bearer …`, `x-api-key: …`, etc.) via `addHeader`/`headers`. +- **Timeout.** `timeoutMs` (default 10000) bounds the whole request; a slow or unreachable endpoint + fails fast instead of stalling the invocation. As with every exporter, a failure here is logged and + isolated — it never disrupts the durable execution or the other exporters. +- **Success / failure.** A 2xx response is success; any other status throws (and is contained by the + plugin), so a misconfigured endpoint is visible in logs. +- **Operations shape.** `operationsFormat` selects what the body carries: `ARRAY` (the canonical + `operations` array, default), `BY_NAME` (the name-keyed `operationsByName` map), or `BOTH`. +- **Size cap.** `maxRecordSizeBytes` has no default — a generic HTTP endpoint has no known limit. Set + it only if your endpoint caps request size; truncation is then measured against the shape actually + sent. +- **Testing seam.** For unit tests you can inject an `HttpSender` via `.sender(...)` instead of making + a real network call, or point `url` at a local in-process server. + ## Design - **Snapshot-based, not accumulated.** Each record is built directly from the current-invocation diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpExporter.java new file mode 100644 index 000000000..39b6e6b07 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpExporter.java @@ -0,0 +1,218 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import java.net.URI; +import java.net.URISyntaxException; +import java.time.Duration; +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; +import software.amazon.lambda.durable.annotations.Experimental; +import software.amazon.lambda.durable.insight.InsightExporter; +import software.amazon.lambda.durable.insight.Json; +import software.amazon.lambda.durable.insight.OperationSummary; +import software.amazon.lambda.durable.insight.OperationsIndex; +import software.amazon.lambda.durable.insight.WorkflowInsightRecord; + +/** + * Exports workflow insight records to any HTTP(S) endpoint via {@code POST} (or {@code PUT}). Each record is sent as a + * JSON body with {@code Content-Type: application/json}; the endpoint must return a 2xx status or the export throws. + * Mirrors the JS {@code HttpExporter} contract (URL, method, custom headers, request timeout, operations format, and + * optional size cap). Uses the JDK's own HTTP client, so it needs no extra dependency. + */ +@Experimental +public final class HttpExporter implements InsightExporter { + + /** HTTP method used to deliver the record. */ + @Experimental + public enum Method { + POST, + PUT + } + + /** + * How the record's operations are shaped in the posted body: the canonical {@code operations} array, the name-keyed + * {@code operationsByName} map, or both. Ports the JS {@code OperationsFormat}. + */ + @Experimental + public enum OperationsFormat { + ARRAY, + BY_NAME, + BOTH + } + + private final String url; + private final Method method; + private final Map headers; + private final Duration timeout; + private final OperationsFormat operationsFormat; + private final Integer maxRecordSizeBytes; + private final HttpSender sender; + + private HttpExporter(Builder b) { + this.url = b.url; + this.method = b.method != null ? b.method : Method.POST; + // Defensive copy so a caller mutating their builder map after build() cannot change this exporter's headers. + this.headers = b.headers != null ? new LinkedHashMap<>(b.headers) : new LinkedHashMap<>(); + this.timeout = b.timeoutMs != null ? Duration.ofMillis(b.timeoutMs) : Duration.ofMillis(10_000); + this.operationsFormat = b.operationsFormat != null ? b.operationsFormat : OperationsFormat.ARRAY; + this.maxRecordSizeBytes = b.maxRecordSizeBytes; + this.sender = b.sender != null ? b.sender : new JdkHttpSender(); + } + + public static Builder builder() { + return new Builder(); + } + + @Override + public Integer maxRecordSizeBytes() { + // No default: a generic HTTP endpoint has no known size limit (matches JS). Truncation is off unless set. + return maxRecordSizeBytes; + } + + @Override + public Object render(WorkflowInsightRecord record) { + switch (operationsFormat) { + case BY_NAME: + return record.toByNameWireMap(); + case BOTH: + return toBothWireMap(record); + case ARRAY: + default: + return record.toWireMap(); + } + } + + /** + * The {@code both} shape: the canonical array wire map with an added {@code operationsByName} map. Composed here + * from the record's public renderings so the SDK core needs no new method. The {@code operationsByName} entry is + * placed before the trailing truncation markers to keep a stable, readable field order. + */ + private Map toBothWireMap(WorkflowInsightRecord record) { + Map data = new LinkedHashMap<>(record.toWireMap()); + Map byName = new LinkedHashMap<>(); + for (Map.Entry e : + OperationsIndex.buildOperationsByName(record.operations()).entrySet()) { + byName.put(e.getKey(), e.getValue().toWireMap()); + } + Map merged = new LinkedHashMap<>(); + for (Map.Entry e : data.entrySet()) { + // Insert operationsByName immediately after the operations array, before truncation markers. + if ("truncated".equals(e.getKey()) && !merged.containsKey("operationsByName")) { + merged.put("operationsByName", byName); + } + merged.put(e.getKey(), e.getValue()); + } + merged.putIfAbsent("operationsByName", byName); + return merged; + } + + @Override + public void export(WorkflowInsightRecord record) { + String body = Json.stringify(render(record)); + Map requestHeaders = new LinkedHashMap<>(); + requestHeaders.put("Content-Type", "application/json"); + requestHeaders.putAll(headers); + HttpSender.Response response = sender.send(url, method.name(), requestHeaders, body, timeout); + int status = response.statusCode(); + if (status < 200 || status >= 300) { + throw new IllegalStateException( + "HttpExporter: endpoint returned " + status + " " + response.reasonPhrase()); + } + } + + /** Builder for {@link HttpExporter}. */ + public static final class Builder { + private String url; + private Method method; + private Map headers; + private Integer timeoutMs; + private OperationsFormat operationsFormat; + private Integer maxRecordSizeBytes; + private HttpSender sender; + + /** Endpoint to POST/PUT records to (required). Must be an absolute {@code http} or {@code https} URL. */ + public Builder url(String url) { + this.url = url; + return this; + } + + /** HTTP method. Default {@link Method#POST}. Use {@link Method#PUT} for endpoints that upsert by URL path. */ + public Builder method(Method method) { + this.method = method; + return this; + } + + /** Additional request headers (for example an {@code Authorization} token or API key). Copied defensively. */ + public Builder headers(Map headers) { + this.headers = headers; + return this; + } + + /** Adds a single request header; convenient for one auth header without building a map. */ + public Builder addHeader(String name, String value) { + if (this.headers == null) { + this.headers = new LinkedHashMap<>(); + } + this.headers.put(name, value); + return this; + } + + /** Request timeout in milliseconds. Default 10000 (10s). Must be positive. */ + public Builder timeoutMs(Integer timeoutMs) { + this.timeoutMs = timeoutMs; + return this; + } + + /** How operations are rendered in the posted body. Default {@link OperationsFormat#ARRAY}. */ + public Builder operationsFormat(OperationsFormat operationsFormat) { + this.operationsFormat = operationsFormat; + return this; + } + + /** Max serialized record size before truncation. No default; set it if your endpoint caps request size. */ + public Builder maxRecordSizeBytes(Integer maxRecordSizeBytes) { + this.maxRecordSizeBytes = maxRecordSizeBytes; + return this; + } + + /** Test seam: inject an {@link HttpSender} instead of the default JDK HTTP client. */ + public Builder sender(HttpSender sender) { + this.sender = sender; + return this; + } + + public HttpExporter build() { + validate(); + return new HttpExporter(this); + } + + private void validate() { + if (url == null || url.isBlank()) { + throw new IllegalArgumentException("HttpExporter requires a url"); + } + URI uri; + try { + uri = new URI(url); + } catch (URISyntaxException e) { + throw new IllegalArgumentException("HttpExporter url is not a valid URI: " + url, e); + } + String scheme = uri.getScheme(); + if (scheme == null || uri.getHost() == null) { + throw new IllegalArgumentException("HttpExporter url must be absolute with a host: " + url); + } + String lower = scheme.toLowerCase(Locale.ROOT); + if (!"http".equals(lower) && !"https".equals(lower)) { + throw new IllegalArgumentException("HttpExporter url scheme must be http or https: " + url); + } + if (timeoutMs != null && timeoutMs <= 0) { + throw new IllegalArgumentException("HttpExporter timeoutMs must be positive: " + timeoutMs); + } + if (maxRecordSizeBytes != null && maxRecordSizeBytes <= 0) { + throw new IllegalArgumentException( + "HttpExporter maxRecordSizeBytes must be positive: " + maxRecordSizeBytes); + } + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpSender.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpSender.java new file mode 100644 index 000000000..9755bd651 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpSender.java @@ -0,0 +1,44 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import java.time.Duration; +import java.util.Map; +import software.amazon.lambda.durable.annotations.Experimental; + +/** + * Test/injection seam for {@link HttpExporter}'s HTTP transport. The default implementation ({@link JdkHttpSender}) is + * backed by {@link java.net.http.HttpClient}; tests inject a stub or point the exporter at a local in-process server. + * Mirrors the {@code client(...)} builder seam on the AWS SDK exporters, but avoids adding any dependency by staying on + * the JDK's own HTTP client. + */ +@Experimental +public interface HttpSender { + + /** + * Sends one request and returns the response status. Implementations MUST apply {@code timeout} to the request and + * MUST throw on transport failure (I/O error, timeout, interruption) so the exporter surfaces it like the JS + * reference (whose {@code fetch} rejects). {@code headers} already includes {@code Content-Type: application/json}. + */ + Response send(String url, String method, Map headers, String body, Duration timeout); + + /** The parts of an HTTP response the exporter needs to decide success/failure. */ + @Experimental + final class Response { + private final int statusCode; + private final String reasonPhrase; + + public Response(int statusCode, String reasonPhrase) { + this.statusCode = statusCode; + this.reasonPhrase = reasonPhrase != null ? reasonPhrase : ""; + } + + public int statusCode() { + return statusCode; + } + + public String reasonPhrase() { + return reasonPhrase; + } + } +} diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/JdkHttpSender.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/JdkHttpSender.java new file mode 100644 index 000000000..e8e140a36 --- /dev/null +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/JdkHttpSender.java @@ -0,0 +1,71 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight.exporters; + +import java.io.IOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpRequest.BodyPublishers; +import java.net.http.HttpResponse; +import java.net.http.HttpResponse.BodyHandlers; +import java.time.Duration; +import java.util.Map; +import software.amazon.lambda.durable.annotations.Experimental; + +/** + * Default {@link HttpSender} backed by the JDK's {@link HttpClient} (Java 11+); adds no third-party dependency. The + * {@link HttpClient} is created once and reused across exports. The per-request timeout bounds the whole exchange, + * mirroring the JS reference's {@code AbortController}. The reason phrase is taken from the HTTP/1.1 status line when + * present; HTTP/2 has no reason phrase, so it is derived from the status code. + */ +@Experimental +final class JdkHttpSender implements HttpSender { + + private final HttpClient client; + + JdkHttpSender() { + this(HttpClient.newBuilder().build()); + } + + JdkHttpSender(HttpClient client) { + this.client = client; + } + + @Override + public Response send(String url, String method, Map headers, String body, Duration timeout) { + HttpRequest.Builder builder = HttpRequest.newBuilder() + .uri(URI.create(url)) + .timeout(timeout) + .method(method, BodyPublishers.ofString(body)); + for (Map.Entry h : headers.entrySet()) { + builder.header(h.getKey(), h.getValue()); + } + try { + HttpResponse response = client.send(builder.build(), BodyHandlers.discarding()); + return new Response(response.statusCode(), reasonFor(response.statusCode())); + } catch (IOException e) { + throw new IllegalStateException("HttpExporter: request to " + url + " failed", e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("HttpExporter: request to " + url + " was interrupted", e); + } + } + + private static String reasonFor(int status) { + // HttpClient does not expose the HTTP/1.1 reason phrase and HTTP/2 has none; derive a stable label from status. + if (status >= 200 && status < 300) { + return "OK"; + } + if (status >= 500) { + return "Server Error"; + } + if (status >= 400) { + return "Client Error"; + } + if (status >= 300) { + return "Redirect"; + } + return "Informational"; + } +} diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/HttpExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/HttpExporterTest.java new file mode 100644 index 000000000..27bbe5ccb --- /dev/null +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/HttpExporterTest.java @@ -0,0 +1,291 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.insight; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sun.net.httpserver.HttpServer; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; +import software.amazon.lambda.durable.insight.exporters.HttpExporter; +import software.amazon.lambda.durable.insight.exporters.HttpSender; + +class HttpExporterTest { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private WorkflowInsightRecord sampleRecord() { + WorkflowInsightRecord r = new WorkflowInsightRecord(); + r.executionArn = "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST"; + r.executionName = "exec-1"; + r.functionName = "fn"; + r.status = "SUCCEEDED"; + r.startTime = "2026-08-05T00:00:00Z"; + r.addOperation(new OperationRecord() + .id("op-1") + .name("greet") + .type("STEP") + .subType("Step") + .status("SUCCEEDED")); + return r; + } + + /** Records the arguments of the last send; returns a configurable status. */ + private static final class RecordingSender implements HttpSender { + String url; + String method; + Map headers; + String body; + Duration timeout; + int status = 200; + + @Override + public Response send(String url, String method, Map headers, String body, Duration timeout) { + this.url = url; + this.method = method; + this.headers = headers; + this.body = body; + this.timeout = timeout; + return new Response(status, status == 200 ? "OK" : "Server Error"); + } + } + + @Test + void postsRecordAsJsonWithContentTypeByDefault() throws Exception { + RecordingSender sender = new RecordingSender(); + HttpExporter exporter = HttpExporter.builder() + .url("https://hook.example/insight") + .sender(sender) + .build(); + + exporter.export(sampleRecord()); + + assertEquals("https://hook.example/insight", sender.url); + assertEquals("POST", sender.method); + assertEquals("application/json", sender.headers.get("Content-Type")); + assertEquals(Duration.ofMillis(10_000), sender.timeout); + JsonNode body = MAPPER.readTree(sender.body); + assertEquals( + "arn:aws:lambda:us-east-1:123456789012:function:fn:$LATEST", + body.get("executionArn").asText()); + assertTrue(body.get("operations").isArray(), "array format emits the canonical operations array"); + assertFalse(body.has("operationsByName"), "array format must not emit the by-name map"); + } + + @Test + void usesPutAndMergesCustomHeadersWhenConfigured() { + RecordingSender sender = new RecordingSender(); + HttpExporter exporter = HttpExporter.builder() + .url("https://hook.example/insight") + .method(HttpExporter.Method.PUT) + .addHeader("Authorization", "Bearer token123") + .sender(sender) + .build(); + + exporter.export(sampleRecord()); + + assertEquals("PUT", sender.method); + assertEquals("Bearer token123", sender.headers.get("Authorization")); + assertEquals("application/json", sender.headers.get("Content-Type")); + } + + @Test + void throwsWhenEndpointReturnsNon2xx() { + RecordingSender sender = new RecordingSender(); + sender.status = 500; + HttpExporter exporter = HttpExporter.builder() + .url("https://hook.example/insight") + .sender(sender) + .build(); + + IllegalStateException ex = assertThrows(IllegalStateException.class, () -> exporter.export(sampleRecord())); + assertTrue(ex.getMessage().contains("500"), ex.getMessage()); + } + + @Test + void byNameFormatEmitsOperationsByNameMap() throws Exception { + RecordingSender sender = new RecordingSender(); + HttpExporter exporter = HttpExporter.builder() + .url("https://hook.example/insight") + .operationsFormat(HttpExporter.OperationsFormat.BY_NAME) + .sender(sender) + .build(); + + exporter.export(sampleRecord()); + + JsonNode body = MAPPER.readTree(sender.body); + assertTrue(body.has("operationsByName"), "by-name format emits the operationsByName map"); + assertTrue(body.get("operationsByName").has("greet")); + assertFalse(body.has("operations"), "by-name format must not emit the operations array"); + } + + @Test + void bothFormatEmitsArrayAndByNameMap() throws Exception { + RecordingSender sender = new RecordingSender(); + HttpExporter exporter = HttpExporter.builder() + .url("https://hook.example/insight") + .operationsFormat(HttpExporter.OperationsFormat.BOTH) + .sender(sender) + .build(); + + exporter.export(sampleRecord()); + + JsonNode body = MAPPER.readTree(sender.body); + assertTrue(body.get("operations").isArray(), "both format keeps the operations array"); + assertTrue(body.get("operationsByName").has("greet"), "both format adds the operationsByName map"); + } + + @Test + void appliesConfiguredTimeout() { + RecordingSender sender = new RecordingSender(); + HttpExporter exporter = HttpExporter.builder() + .url("https://hook.example/insight") + .timeoutMs(2_500) + .sender(sender) + .build(); + + exporter.export(sampleRecord()); + + assertEquals(Duration.ofMillis(2_500), sender.timeout); + } + + @Test + void builderRejectsMissingUrl() { + assertThrows( + IllegalArgumentException.class, () -> HttpExporter.builder().build()); + } + + @Test + void builderRejectsNonHttpScheme() { + assertThrows( + IllegalArgumentException.class, + () -> HttpExporter.builder().url("ftp://host/path").build()); + } + + @Test + void builderRejectsNonAbsoluteUrl() { + assertThrows( + IllegalArgumentException.class, + () -> HttpExporter.builder().url("/relative/path").build()); + } + + @Test + void builderRejectsNonPositiveTimeout() { + assertThrows(IllegalArgumentException.class, () -> HttpExporter.builder() + .url("https://hook.example/insight") + .timeoutMs(0) + .build()); + } + + // --- Real in-process HTTP server tests (exercise the default JDK sender end to end) --- + + @Test + void postsToRealLocalServerWithDefaultSender() throws Exception { + AtomicReference receivedMethod = new AtomicReference<>(); + AtomicReference receivedContentType = new AtomicReference<>(); + AtomicReference receivedBody = new AtomicReference<>(); + HttpServer server = startServer(exchange -> { + receivedMethod.set(exchange.getRequestMethod()); + receivedContentType.set(exchange.getRequestHeaders().getFirst("Content-Type")); + receivedBody.set(new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8)); + exchange.sendResponseHeaders(200, -1); + exchange.close(); + }); + try { + HttpExporter exporter = HttpExporter.builder() + .url("http://" + authority(server) + "/insight") + .build(); + + exporter.export(sampleRecord()); + + assertEquals("POST", receivedMethod.get()); + assertEquals("application/json", receivedContentType.get()); + JsonNode body = MAPPER.readTree(receivedBody.get()); + assertEquals("exec-1", body.get("executionName").asText()); + assertTrue(body.get("operations").isArray()); + } finally { + server.stop(0); + } + } + + @Test + void realLocalServer500Throws() throws Exception { + HttpServer server = startServer(exchange -> { + exchange.sendResponseHeaders(500, -1); + exchange.close(); + }); + try { + HttpExporter exporter = HttpExporter.builder() + .url("http://" + authority(server) + "/insight") + .build(); + + IllegalStateException ex = assertThrows(IllegalStateException.class, () -> exporter.export(sampleRecord())); + assertTrue(ex.getMessage().contains("500"), ex.getMessage()); + } finally { + server.stop(0); + } + } + + @Test + void realLocalServerReceivesCustomAuthHeaderAndPutMethod() throws Exception { + List auth = new CopyOnWriteArrayList<>(); + AtomicReference method = new AtomicReference<>(); + HttpServer server = startServer(exchange -> { + String header = exchange.getRequestHeaders().getFirst("Authorization"); + if (header != null) { + auth.add(header); + } + method.set(exchange.getRequestMethod()); + exchange.sendResponseHeaders(204, -1); + exchange.close(); + }); + try { + HttpExporter exporter = HttpExporter.builder() + .url("http://" + authority(server) + "/insight") + .method(HttpExporter.Method.PUT) + .addHeader("Authorization", "Bearer abc") + .build(); + + exporter.export(sampleRecord()); + + assertEquals("PUT", method.get()); + assertEquals(List.of("Bearer abc"), auth); + } finally { + server.stop(0); + } + } + + private interface Handler { + void handle(com.sun.net.httpserver.HttpExchange exchange) throws IOException; + } + + private static HttpServer startServer(Handler handler) throws IOException { + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/insight", exchange -> { + try { + handler.handle(exchange); + } catch (RuntimeException e) { + exchange.sendResponseHeaders(500, -1); + exchange.close(); + } + }); + server.start(); + return server; + } + + private static String authority(HttpServer server) { + return "127.0.0.1:" + server.getAddress().getPort(); + } +} From 29339283f3902904e12049c2cc010cea14edc198 Mon Sep 17 00:00:00 2001 From: Alex Wang Date: Thu, 10 Sep 2026 00:03:36 +0000 Subject: [PATCH 2/2] fix(insight): case-insensitive HTTP header merge Merge request headers case-insensitively so a caller-supplied Content-Type of any casing overrides the default application/json and yields exactly one header with no duplicate on the wire. Add default JDK-sender transport-failure coverage: a connection refusal (reserved-then-released loopback port, bounded timeout) and thread interruption, both proving IllegalStateException; the interruption test also asserts the interrupt flag is preserved. Align HttpExporter Javadoc and README wording: Content-Type is a default a caller may override. No response-body logging added. --- insight-plugin/README.md | 11 +- .../insight/exporters/HttpExporter.java | 40 ++++-- .../durable/insight/HttpExporterTest.java | 119 ++++++++++++++++++ 3 files changed, 156 insertions(+), 14 deletions(-) diff --git a/insight-plugin/README.md b/insight-plugin/README.md index 4481f8143..f49cc9f2e 100644 --- a/insight-plugin/README.md +++ b/insight-plugin/README.md @@ -55,9 +55,9 @@ configure, using the AWS SDK for Java 2.x version managed by your application: ### HttpExporter -`HttpExporter` POSTs (or PUTs) each record to any HTTP(S) endpoint or webhook as a JSON body with -`Content-Type: application/json`. It uses the JDK's built-in HTTP client, so it adds **no dependency**. -It mirrors the JS `HttpExporter`. +`HttpExporter` POSTs (or PUTs) each record to any HTTP(S) endpoint or webhook as a JSON body with a +default `Content-Type: application/json` header that a caller may override. It uses the JDK's built-in +HTTP client, so it adds **no dependency**. It mirrors the JS `HttpExporter`. ```java .addExporter(HttpExporter.builder() @@ -72,8 +72,9 @@ It mirrors the JS `HttpExporter`. Behavior: - **Endpoint / method / headers.** Records are sent to `url` with `method` (POST or PUT — use PUT for - endpoints that upsert by URL path). Your headers are merged on top of the fixed - `Content-Type: application/json`. + endpoints that upsert by URL path). Your headers are layered on top of the default + `Content-Type: application/json`. Header names are matched case-insensitively, so a caller-supplied + `Content-Type` (any casing) overrides the default and never produces a duplicate header. - **Auth.** There is no built-in auth scheme; pass whatever your endpoint expects as a header (`Authorization: Bearer …`, `x-api-key: …`, etc.) via `addHeader`/`headers`. - **Timeout.** `timeoutMs` (default 10000) bounds the whole request; a slow or unreachable endpoint diff --git a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpExporter.java b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpExporter.java index 39b6e6b07..7868b069f 100644 --- a/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpExporter.java +++ b/insight-plugin/src/main/java/software/amazon/lambda/durable/insight/exporters/HttpExporter.java @@ -17,9 +17,10 @@ /** * Exports workflow insight records to any HTTP(S) endpoint via {@code POST} (or {@code PUT}). Each record is sent as a - * JSON body with {@code Content-Type: application/json}; the endpoint must return a 2xx status or the export throws. - * Mirrors the JS {@code HttpExporter} contract (URL, method, custom headers, request timeout, operations format, and - * optional size cap). Uses the JDK's own HTTP client, so it needs no extra dependency. + * JSON body with a default {@code Content-Type: application/json} header that a caller may override; the endpoint must + * return a 2xx status or the export throws. Mirrors the JS {@code HttpExporter} contract (URL, method, custom headers, + * request timeout, operations format, and optional size cap). Uses the JDK's own HTTP client, so it needs no extra + * dependency. */ @Experimental public final class HttpExporter implements InsightExporter { @@ -111,10 +112,7 @@ private Map toBothWireMap(WorkflowInsightRecord record) { @Override public void export(WorkflowInsightRecord record) { String body = Json.stringify(render(record)); - Map requestHeaders = new LinkedHashMap<>(); - requestHeaders.put("Content-Type", "application/json"); - requestHeaders.putAll(headers); - HttpSender.Response response = sender.send(url, method.name(), requestHeaders, body, timeout); + HttpSender.Response response = sender.send(url, method.name(), mergeHeaders(), body, timeout); int status = response.statusCode(); if (status < 200 || status >= 300) { throw new IllegalStateException( @@ -122,6 +120,23 @@ public void export(WorkflowInsightRecord record) { } } + /** + * Builds the request headers: the default {@code Content-Type: application/json} with the caller's headers layered + * on top. Header names are matched case-insensitively (HTTP header names are case-insensitive), so a caller header + * of any casing — {@code content-type}, {@code CONTENT-TYPE}, {@code Content-Type} — replaces an earlier entry + * instead of adding a duplicate. The result carries exactly one header per name, and any caller-supplied + * {@code Content-Type} overrides the default. + */ + private Map mergeHeaders() { + Map merged = new LinkedHashMap<>(); + merged.put("Content-Type", "application/json"); + for (Map.Entry e : headers.entrySet()) { + merged.keySet().removeIf(existing -> existing.equalsIgnoreCase(e.getKey())); + merged.put(e.getKey(), e.getValue()); + } + return merged; + } + /** Builder for {@link HttpExporter}. */ public static final class Builder { private String url; @@ -144,13 +159,20 @@ public Builder method(Method method) { return this; } - /** Additional request headers (for example an {@code Authorization} token or API key). Copied defensively. */ + /** + * Additional request headers (for example an {@code Authorization} token or API key). Copied defensively. + * Header names are matched case-insensitively, and a header named {@code Content-Type} (any casing) overrides + * the exporter's default {@code application/json}. + */ public Builder headers(Map headers) { this.headers = headers; return this; } - /** Adds a single request header; convenient for one auth header without building a map. */ + /** + * Adds a single request header; convenient for one auth header without building a map. Follows the same + * case-insensitive, default-overriding rules as {@link #headers(Map)}. + */ public Builder addHeader(String name, String value) { if (this.headers == null) { this.headers = new LinkedHashMap<>(); diff --git a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/HttpExporterTest.java b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/HttpExporterTest.java index 27bbe5ccb..740c3a5ba 100644 --- a/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/HttpExporterTest.java +++ b/insight-plugin/src/test/java/software/amazon/lambda/durable/insight/HttpExporterTest.java @@ -11,12 +11,17 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.sun.net.httpserver.HttpServer; import java.io.IOException; +import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.ServerSocket; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.List; import java.util.Map; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; import software.amazon.lambda.durable.insight.exporters.HttpExporter; @@ -101,6 +106,29 @@ void usesPutAndMergesCustomHeadersWhenConfigured() { assertEquals("application/json", sender.headers.get("Content-Type")); } + @Test + void callerContentTypeOverridesDefaultCaseInsensitivelyWithNoDuplicate() { + RecordingSender sender = new RecordingSender(); + HttpExporter exporter = HttpExporter.builder() + .url("https://hook.example/insight") + .addHeader("content-type", "application/json; charset=utf-8") + .sender(sender) + .build(); + + exporter.export(sampleRecord()); + + int contentTypeCount = 0; + String contentTypeValue = null; + for (Map.Entry h : sender.headers.entrySet()) { + if (h.getKey().equalsIgnoreCase("content-type")) { + contentTypeCount++; + contentTypeValue = h.getValue(); + } + } + assertEquals(1, contentTypeCount, "exactly one content-type header regardless of casing: " + sender.headers); + assertEquals("application/json; charset=utf-8", contentTypeValue, "caller value overrides the default"); + } + @Test void throwsWhenEndpointReturnsNon2xx() { RecordingSender sender = new RecordingSender(); @@ -267,6 +295,97 @@ void realLocalServerReceivesCustomAuthHeaderAndPutMethod() throws Exception { } } + @Test + void realLocalServerReceivesSingleContentTypeWhenCallerOverridesCasing() throws Exception { + List contentTypes = new CopyOnWriteArrayList<>(); + HttpServer server = startServer(exchange -> { + List received = exchange.getRequestHeaders().get("Content-Type"); + if (received != null) { + contentTypes.addAll(received); + } + exchange.sendResponseHeaders(204, -1); + exchange.close(); + }); + try { + HttpExporter exporter = HttpExporter.builder() + .url("http://" + authority(server) + "/insight") + .addHeader("content-type", "application/json") + .build(); + + exporter.export(sampleRecord()); + + assertEquals(1, contentTypes.size(), "exactly one Content-Type header on the wire: " + contentTypes); + assertEquals("application/json", contentTypes.get(0)); + } finally { + server.stop(0); + } + } + + @Test + void defaultSenderThrowsIllegalStateExceptionWhenConnectionRefused() throws Exception { + // Reserve then release a loopback port so nothing is listening on it: connecting is refused deterministically. + int deadPort; + try (ServerSocket socket = new ServerSocket(0, 0, InetAddress.getByName("127.0.0.1"))) { + deadPort = socket.getLocalPort(); + } + HttpExporter exporter = HttpExporter.builder() + .url("http://127.0.0.1:" + deadPort + "/insight") + .timeoutMs(2_000) // bounded so a stray listener could never hang the test + .build(); + + IllegalStateException ex = assertThrows(IllegalStateException.class, () -> exporter.export(sampleRecord())); + assertTrue(ex.getMessage().contains("failed"), ex.getMessage()); + assertTrue(ex.getMessage().contains("127.0.0.1:" + deadPort), ex.getMessage()); + } + + @Test + void defaultSenderThrowsAndPreservesInterruptWhenExportingThreadInterrupted() throws Exception { + CountDownLatch requestReceived = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + // Handler blocks without responding, so the client stays parked waiting for the response headers. + HttpServer server = startServer(exchange -> { + requestReceived.countDown(); + try { + release.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + exchange.sendResponseHeaders(204, -1); + exchange.close(); + }); + try { + HttpExporter exporter = HttpExporter.builder() + .url("http://" + authority(server) + "/insight") + .timeoutMs(30_000) + .build(); + + AtomicReference thrown = new AtomicReference<>(); + AtomicBoolean interruptPreserved = new AtomicBoolean(); + Thread worker = new Thread(() -> { + try { + exporter.export(sampleRecord()); + } catch (Throwable t) { + thrown.set(t); + interruptPreserved.set(Thread.currentThread().isInterrupted()); + } + }); + worker.start(); + + assertTrue(requestReceived.await(10, TimeUnit.SECONDS), "server should receive the request first"); + worker.interrupt(); + worker.join(TimeUnit.SECONDS.toMillis(10)); + + assertFalse(worker.isAlive(), "worker should return after interruption"); + Throwable t = thrown.get(); + assertTrue(t instanceof IllegalStateException, "expected IllegalStateException, got " + t); + assertTrue(t.getMessage().contains("interrupted"), t.getMessage()); + assertTrue(interruptPreserved.get(), "interrupt flag must be preserved after interruption"); + } finally { + release.countDown(); + server.stop(0); + } + } + private interface Handler { void handle(com.sun.net.httpserver.HttpExchange exchange) throws IOException; }