diff --git a/compat-0.3/server-conversion/src/test/java/org/a2aproject/sdk/compat03/conversion/AbstractA2ARequestHandlerTest_v0_3.java b/compat-0.3/server-conversion/src/test/java/org/a2aproject/sdk/compat03/conversion/AbstractA2ARequestHandlerTest_v0_3.java index 01ea100fc..4c4e8320c 100644 --- a/compat-0.3/server-conversion/src/test/java/org/a2aproject/sdk/compat03/conversion/AbstractA2ARequestHandlerTest_v0_3.java +++ b/compat-0.3/server-conversion/src/test/java/org/a2aproject/sdk/compat03/conversion/AbstractA2ARequestHandlerTest_v0_3.java @@ -30,7 +30,6 @@ import org.a2aproject.sdk.server.tasks.InMemoryPushNotificationConfigStore; import org.a2aproject.sdk.server.tasks.InMemoryTaskStore; import org.a2aproject.sdk.server.tasks.PushNotificationConfigStore; -import org.a2aproject.sdk.server.tasks.PushNotificationSender; import org.a2aproject.sdk.server.tasks.PushNotificationUrlValidator; import org.a2aproject.sdk.server.tasks.TaskStore; import org.a2aproject.sdk.spec.A2AError; @@ -85,8 +84,6 @@ public abstract class AbstractA2ARequestHandlerTest_v0_3 { .parts(new TextPart_v0_3("test message")) .build(); - private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = (event, snapshot) -> {}; - // V1.0 backend infrastructure protected AgentExecutor agentExecutor; protected TaskStore taskStore; diff --git a/docs/content/dev/extra/http-clients.md b/docs/content/dev/extra/http-clients.md index 0c72f119d..e571376a4 100644 --- a/docs/content/dev/extra/http-clients.md +++ b/docs/content/dev/extra/http-clients.md @@ -266,3 +266,9 @@ public class MyHttpClientProducer { | 0 | JDK | Always (fallback) | If the CDI container is active but no `A2AHttpClient` bean is registered, this provider is skipped and the next available provider is used. + +### Compatibility with OpenTelemetry + +When the [OpenTelemetry server module](opentelemetry.md) is on the classpath alongside a custom `A2AHttpClient` CDI bean, the OpenTelemetry module wraps your bean with trace-context propagation rather than replacing it. Your custom configuration (mTLS, proxy, authentication, timeouts, etc.) is preserved for push notification delivery while `traceparent` and `tracestate` headers are injected automatically. + +This works through the `@Instrumented` CDI qualifier (defined in `server-common`): the OpenTelemetry module produces an `@Instrumented A2AHttpClient` that delegates to your unqualified bean. `BasePushNotificationSender` prefers the `@Instrumented` variant when available, so both your custom configuration and the trace-context headers are present on outbound requests. No changes to your producer code are required. diff --git a/docs/content/dev/extra/opentelemetry.md b/docs/content/dev/extra/opentelemetry.md index f084eeeed..e29676daa 100644 --- a/docs/content/dev/extra/opentelemetry.md +++ b/docs/content/dev/extra/opentelemetry.md @@ -177,6 +177,70 @@ The following attributes are automatically added to spans (defined in `A2AObserv | `gen_ai.agent.a2a.response` | Full response data (only if extraction enabled) | | `error.type` | Error message (on failures) | +### Push Notification Delivery Tracing + +When the server delivers a push notification to a client webhook, `OpenTelemetryPushNotificationSenderDecorator` creates a `CLIENT` span named `SendPushNotification`. This span is independent of the original request span but carries the `gen_ai.agent.a2a.task_id` attribute, allowing correlation by task ID in your observability tool. The span status reflects the actual delivery outcome: `OK` when all notifications succeed, `ERROR` when one or more deliveries fail (with `error.type` set to `delivery_failed`) or when an exception is thrown. + +The module also provides **W3C trace-context propagation** for outbound push notification HTTP requests via the `@Instrumented` CDI qualifier and `OpenTelemetryHttpClientProducer`. + +#### How it works + +The `OpenTelemetryHttpClientProducer` CDI bean produces an `@Instrumented A2AHttpClient` — a wrapper that injects `traceparent` and `tracestate` headers into every outbound HTTP request. `BasePushNotificationSender` resolves the HTTP client in this order: + +1. An `@Instrumented A2AHttpClient` bean (if available — e.g. the one produced by this module) +2. An unqualified `A2AHttpClient` bean (if the application provides one) +3. `A2AHttpClientFactory.create()` as last resort + +The `@Instrumented` qualifier (defined in `server-common`) marks HTTP client beans that have been wrapped with observability instrumentation. It ensures the instrumented client does not compete with the application's own `A2AHttpClient` bean in default CDI resolution — the two coexist, and the instrumented variant wraps the application's bean as its delegate. + +#### Custom HTTP client compatibility + +If your application provides a custom `A2AHttpClient` CDI bean (e.g. for mTLS, proxy, or custom timeouts), the OpenTelemetry wrapper **preserves that configuration** by delegating to your bean rather than replacing it. No additional annotations or configuration are needed — just produce your bean as usual: + +```java +@Produces +@ApplicationScoped +public A2AHttpClient myHttpClient() { + // your custom client (mTLS, proxy, auth, etc.) +} +``` + +The OpenTelemetry module automatically wraps it with trace-context injection. + +#### Context-propagating executor + +The producer also creates an `@Instrumented Executor` (via `Context.taskWrapping()`) that ensures the OTel trace context survives the async boundary when push notifications are dispatched to multiple recipients in parallel. This allows the receiving webhook to join the distributed trace. + +`BasePushNotificationSender` resolves the dispatch executor in this order: + +1. An `@Instrumented Executor` bean (if available — e.g. the one produced by this module) +2. A `@Named("a2a.pushNotificationDispatch") Executor` bean (if the application provides one) +3. `ForkJoinPool.commonPool()` as last resort + +#### Custom dispatch executor compatibility + +If your application provides a custom `@Named("a2a.pushNotificationDispatch") Executor` CDI bean (e.g. for custom thread pool sizing), the OpenTelemetry module **preserves that configuration** by wrapping it with `Context.taskWrapping()` rather than replacing it. Just produce your bean as usual: + +```java +@Produces +@ApplicationScoped +@Named("a2a.pushNotificationDispatch") +public Executor myDispatchExecutor() { + // your custom thread pool +} +``` + +The OpenTelemetry module automatically wraps it with context propagation. + +| Attribute | Description | +|-----------|-------------| +| `gen_ai.agent.a2a.operation.name` | `SendPushNotification` | +| `gen_ai.agent.a2a.task_id` | Task identifier | +| `gen_ai.agent.a2a.context_id` | Context identifier (when available) | +| `gen_ai.agent.a2a.push_notification.event_kind` | The type of streaming event delivered (e.g. `TaskStatusUpdateEvent`) | +| `gen_ai.agent.a2a.response` | Full event payload (only if extraction enabled) | +| `error.type` | `delivery_failed` when one or more deliveries fail, or the exception message when an exception is thrown | + ### Request/Response Extraction Enable request and response data extraction in spans using JVM system properties: @@ -195,7 +259,7 @@ Enable request and response data extraction in spans using JVM system properties ### Instrumentation -Adds OpenTelemetry spans to A2A client operations: +Adds OpenTelemetry spans and metrics to A2A client operations: ```xml @@ -206,11 +270,20 @@ Adds OpenTelemetry spans to A2A client operations: ### Metrics -The client transport records the following histogram: +The client transport wrapper is auto-discovered via ServiceLoader — just adding the dependency enables tracing. To also enable `gen_ai.client.operation.duration` metrics, pass a `Meter` via the transport config: + +```java +config.setParameters(Map.of( + OpenTelemetryClientTransportWrapper.OTEL_TRACER_KEY, openTelemetry.getTracer("my-service"), + OpenTelemetryClientTransportWrapper.OTEL_METER_KEY, openTelemetry.getMeter("my-service") +)); +``` + +If no `Meter` is provided, the histogram is silently skipped. | Instrument | Type | Unit | Description | |------------|------|------|-------------| -| `gen_ai.client.operation.duration` | Histogram | `s` | Duration of A2A client operations from invocation to completion or error | +| `gen_ai.client.operation.duration` | Histogram | `s` | Duration of each A2A client operation call | Attributes on `gen_ai.client.operation.duration`: @@ -220,14 +293,6 @@ Attributes on `gen_ai.client.operation.duration`: | `gen_ai.system` | `a2a` | | `error.type` | `error` (only on failure) | -Metrics require passing a `Meter` to the constructor: - -```java -OpenTelemetryClientTransport transport = new OpenTelemetryClientTransport(delegate, tracer, meter); -``` - -If `meter` is `null`, the histogram is silently skipped. - ### Context Propagation For trace context propagation across async client boundaries: diff --git a/extras/http-client-cdi/src/test/java/org/a2aproject/sdk/client/http/cdi/CdiA2AHttpClientFactoryTest.java b/extras/http-client-cdi/src/test/java/org/a2aproject/sdk/client/http/cdi/CdiA2AHttpClientFactoryTest.java index e3f7109a1..abac53ae3 100644 --- a/extras/http-client-cdi/src/test/java/org/a2aproject/sdk/client/http/cdi/CdiA2AHttpClientFactoryTest.java +++ b/extras/http-client-cdi/src/test/java/org/a2aproject/sdk/client/http/cdi/CdiA2AHttpClientFactoryTest.java @@ -81,6 +81,17 @@ public void testCreateDoesNotLogWarningWhenCdiSucceeds() { assertFalse(records.stream().anyMatch(r -> r.getLevel() == Level.WARNING), "No warning should be logged when CDI provider succeeds"); } + + @Test + public void testCreateExcludingCdiSkipsTheCdiBeanAndFallsBack() { + // Regression guard for the recursive CDI self-delegation bug: a CDI producer that + // produces an A2AHttpClient must call createExcluding("cdi") so the factory does not + // re-enter the container and return the proxy being constructed as its own delegate. + A2AHttpClient client = A2AHttpClientFactory.createExcluding("cdi"); + assertNotNull(client); + assertFalse(client instanceof DummyA2AHttpClient, + "createExcluding('cdi') must not return the CDI-registered bean"); + } } @Nested diff --git a/extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/A2AObservabilityNames.java b/extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/A2AObservabilityNames.java index 8ea1fa7c6..db94826af 100644 --- a/extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/A2AObservabilityNames.java +++ b/extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/A2AObservabilityNames.java @@ -42,4 +42,8 @@ public interface A2AObservabilityNames { String GENAI_STREAMING_DURATION = GENAI_PREFIX + ".streaming.duration"; String GENAI_SYSTEM = "gen_ai.system"; String GENAI_SYSTEM_VALUE = "a2a"; + + // Push notification + String PUSH_NOTIFICATION_EVENT_KIND = GENAI_PREFIX + ".push_notification.event_kind"; + String PUSH_NOTIFICATION_DELIVERY_FAILED = "delivery_failed"; } diff --git a/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutes.java b/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutes.java index 13ca3a44f..9774a73f8 100644 --- a/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutes.java +++ b/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutes.java @@ -1,8 +1,7 @@ package org.a2aproject.sdk.extras.opentelemetry.it; - - import static io.vertx.core.http.HttpHeaders.CONTENT_TYPE; +import static java.net.HttpURLConnection.HTTP_OK; import com.google.gson.Gson; import com.google.gson.GsonBuilder; @@ -12,6 +11,7 @@ import org.a2aproject.sdk.jsonrpc.common.json.JsonUtil; import org.a2aproject.sdk.spec.Task; import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent; +import org.a2aproject.sdk.spec.TaskPushNotificationConfig; import org.a2aproject.sdk.spec.TaskStatusUpdateEvent; import io.opentelemetry.api.OpenTelemetry; import io.opentelemetry.api.trace.Span; @@ -32,7 +32,10 @@ import jakarta.enterprise.inject.Produces; import jakarta.inject.Inject; import jakarta.inject.Singleton; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; /** * Test routes for OpenTelemetry integration testing. @@ -45,6 +48,11 @@ public class A2ATestRoutes { private static final String TEXT_PLAIN = "text/plain"; private static final Gson gson = new GsonBuilder().create(); + /** + * Captured webhook requests, each entry is a map of header name to value. + */ + private final List> webhookRequests = new CopyOnWriteArrayList<>(); + @Inject TestUtilsBean testUtilsBean; @Inject @@ -53,6 +61,8 @@ public class A2ATestRoutes { InMemoryMetricExporter inMemoryMetricExporter; @Inject OpenTelemetry openTelemetry; + @Inject + CustomDispatchExecutorProducer customDispatchExecutorProducer; @Inject Tracer tracer; @@ -85,6 +95,37 @@ void setupRoutes(@Observes Router router) { .produces(TEXT_PLAIN) .handler(ctx -> getChildQueueCount(ctx.pathParam("taskId"), ctx)); + router.post("/test/pushconfig/:taskId") + .consumes(APPLICATION_JSON) + .handler(BodyHandler.create()) + .blockingHandler(ctx -> savePushNotificationConfig(ctx.pathParam("taskId"), ctx.body().asString(), ctx)); + + router.delete("/test/pushconfig/:taskId/:configId") + .blockingHandler(ctx -> deletePushNotificationConfig( + ctx.pathParam("taskId"), ctx.pathParam("configId"), ctx)); + + router.post("/test/webhook") + .handler(BodyHandler.create()) + .handler(ctx -> captureWebhook(ctx)); + + router.get("/test/webhook/requests") + .produces(APPLICATION_JSON) + .handler(ctx -> getWebhookRequests(ctx)); + + router.delete("/test/webhook/requests") + .handler(ctx -> clearWebhookRequests(ctx)); + + router.get("/test/dispatch-executor/count") + .produces(TEXT_PLAIN) + .handler(ctx -> ctx.response().setStatusCode(HTTP_OK) + .end(String.valueOf(customDispatchExecutorProducer.getExecutionCount()))); + + router.delete("/test/dispatch-executor/count") + .handler(ctx -> { + customDispatchExecutorProducer.resetExecutionCount(); + ctx.response().setStatusCode(HTTP_OK).end(); + }); + router.get("/hello") .produces(TEXT_PLAIN) .handler(ctx -> hello(ctx)); @@ -197,6 +238,48 @@ public void getChildQueueCount(String taskId, RoutingContext rc) { .end(String.valueOf(count)); } + public void savePushNotificationConfig(String taskId, String body, RoutingContext rc) { + try { + TaskPushNotificationConfig config = JsonUtil.fromJson(body, TaskPushNotificationConfig.class); + testUtilsBean.savePushNotificationConfig(taskId, config); + rc.response() + .setStatusCode(200) + .end(); + } catch (Throwable t) { + errorResponse(t, rc); + } + } + + public void deletePushNotificationConfig(String taskId, String configId, RoutingContext rc) { + try { + testUtilsBean.deletePushNotificationConfig(taskId, configId); + rc.response() + .setStatusCode(200) + .end(); + } catch (Throwable t) { + errorResponse(t, rc); + } + } + + private void captureWebhook(RoutingContext rc) { + Map headers = new HashMap<>(); + rc.request().headers().forEach(entry -> headers.put(entry.getKey(), entry.getValue())); + webhookRequests.add(headers); + rc.response().setStatusCode(HTTP_OK).end(); + } + + private void getWebhookRequests(RoutingContext rc) { + rc.response() + .setStatusCode(HTTP_OK) + .putHeader(CONTENT_TYPE, APPLICATION_JSON) + .end(gson.toJson(webhookRequests)); + } + + private void clearWebhookRequests(RoutingContext rc) { + webhookRequests.clear(); + rc.response().setStatusCode(HTTP_OK).end(); + } + public void hello(RoutingContext rc) { Span span = tracer.spanBuilder("hello").startSpan(); try (Scope scope = span.makeCurrent()) { diff --git a/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/CustomDispatchExecutorProducer.java b/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/CustomDispatchExecutorProducer.java new file mode 100644 index 000000000..c3eda9fe8 --- /dev/null +++ b/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/CustomDispatchExecutorProducer.java @@ -0,0 +1,70 @@ +package org.a2aproject.sdk.extras.opentelemetry.it; + +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicInteger; + +import jakarta.annotation.PreDestroy; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.inject.Produces; +import jakarta.inject.Named; +import org.a2aproject.sdk.server.tasks.BasePushNotificationSender; + +/** + * Produces a custom {@link Named @Named} dispatch {@link Executor} CDI bean + * that uses a distinctive thread name prefix. Used by integration tests to + * verify that the OpenTelemetry module wraps the application's executor with + * context propagation instead of replacing it. + * + *

An {@link AtomicInteger} counter tracks how many tasks have been + * submitted, allowing tests to confirm the custom executor was actually used. + */ +@ApplicationScoped +public class CustomDispatchExecutorProducer { + + static final String CUSTOM_THREAD_PREFIX = "custom-dispatch-"; + + private final AtomicInteger executionCount = new AtomicInteger(); + private final ExecutorService pool; + + public CustomDispatchExecutorProducer() { + AtomicInteger threadCounter = new AtomicInteger(); + ThreadFactory threadFactory = r -> { + Thread t = new Thread(r, CUSTOM_THREAD_PREFIX + threadCounter.getAndIncrement()); + t.setDaemon(true); + return t; + }; + this.pool = Executors.newFixedThreadPool(2, threadFactory); + } + + @Produces + @ApplicationScoped + @Named(BasePushNotificationSender.DISPATCH_EXECUTOR_NAME) + Executor customDispatchExecutor() { + return command -> { + executionCount.incrementAndGet(); + pool.execute(command); + }; + } + + /** + * Returns the number of tasks submitted through this executor. + */ + public int getExecutionCount() { + return executionCount.get(); + } + + /** + * Resets the execution counter. + */ + public void resetExecutionCount() { + executionCount.set(0); + } + + @PreDestroy + void shutdown() { + pool.shutdownNow(); + } +} diff --git a/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/CustomHttpClientProducer.java b/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/CustomHttpClientProducer.java new file mode 100644 index 000000000..daf256c36 --- /dev/null +++ b/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/CustomHttpClientProducer.java @@ -0,0 +1,54 @@ +package org.a2aproject.sdk.extras.opentelemetry.it; + +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.inject.Produces; +import org.a2aproject.sdk.client.http.A2AHttpClient; +import org.a2aproject.sdk.client.http.A2AHttpClientFactory; + +/** + * Produces a custom {@link A2AHttpClient} CDI bean that adds a marker header to + * every outbound POST request. Used by integration tests to verify that the + * OpenTelemetry module wraps the application's client instead of replacing it. + */ +@ApplicationScoped +public class CustomHttpClientProducer { + + static final String CUSTOM_HEADER_NAME = "X-Custom-A2A-Client"; + static final String CUSTOM_HEADER_VALUE = "integration-test"; + + @Produces + @ApplicationScoped + A2AHttpClient customClient() { + return new MarkerHeaderA2AHttpClient(A2AHttpClientFactory.createExcluding("cdi")); + } + + /** + * Delegating {@link A2AHttpClient} that injects a marker header into every + * POST request for test observability. + */ + static class MarkerHeaderA2AHttpClient implements A2AHttpClient { + + private final A2AHttpClient delegate; + + MarkerHeaderA2AHttpClient(A2AHttpClient delegate) { + this.delegate = delegate; + } + + @Override + public GetBuilder createGet() { + return delegate.createGet(); + } + + @Override + public PostBuilder createPost() { + PostBuilder builder = delegate.createPost(); + builder.addHeader(CUSTOM_HEADER_NAME, CUSTOM_HEADER_VALUE); + return builder; + } + + @Override + public DeleteBuilder createDelete() { + return delegate.createDelete(); + } + } +} diff --git a/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/TestAgentCardProducer.java b/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/TestAgentCardProducer.java index a34d076f0..9ea7a3b9c 100644 --- a/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/TestAgentCardProducer.java +++ b/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/TestAgentCardProducer.java @@ -31,7 +31,7 @@ public AgentCard agentCard() { .documentationUrl("http://example.com/test") .capabilities(AgentCapabilities.builder() .streaming(true) - .pushNotifications(false) + .pushNotifications(true) .build()) .defaultInputModes(Collections.singletonList("text")) .defaultOutputModes(Collections.singletonList("text")) diff --git a/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/TestUtilsBean.java b/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/TestUtilsBean.java index 52a3ee7bb..4db0b2611 100644 --- a/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/TestUtilsBean.java +++ b/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/TestUtilsBean.java @@ -1,15 +1,17 @@ package org.a2aproject.sdk.extras.opentelemetry.it; import org.a2aproject.sdk.server.events.QueueManager; +import org.a2aproject.sdk.server.tasks.PushNotificationConfigStore; import org.a2aproject.sdk.server.tasks.TaskStore; import org.a2aproject.sdk.spec.Event; import org.a2aproject.sdk.spec.Task; +import org.a2aproject.sdk.spec.TaskPushNotificationConfig; import jakarta.enterprise.context.ApplicationScoped; import jakarta.inject.Inject; /** * Test utilities for OpenTelemetry integration tests. - * Allows direct manipulation of tasks and queues for testing. + * Allows direct manipulation of tasks, queues, and push notification configs for testing. */ @ApplicationScoped public class TestUtilsBean { @@ -20,6 +22,9 @@ public class TestUtilsBean { @Inject QueueManager queueManager; + @Inject + PushNotificationConfigStore pushNotificationConfigStore; + public void saveTask(Task task) { taskStore.save(task, false); } @@ -43,4 +48,14 @@ public void enqueueEvent(String taskId, Event event) { public int getChildQueueCount(String taskId) { return queueManager.getActiveChildQueueCount(taskId); } + + public void savePushNotificationConfig(String taskId, TaskPushNotificationConfig config) { + pushNotificationConfigStore.setInfo( + TaskPushNotificationConfig.builder(config).taskId(taskId).build(), + null); + } + + public void deletePushNotificationConfig(String taskId, String configId) { + pushNotificationConfigStore.deleteInfo(taskId, configId); + } } diff --git a/extras/opentelemetry/integration-tests/src/main/resources/application.properties b/extras/opentelemetry/integration-tests/src/main/resources/application.properties index aae4890aa..f4f1662c6 100644 --- a/extras/opentelemetry/integration-tests/src/main/resources/application.properties +++ b/extras/opentelemetry/integration-tests/src/main/resources/application.properties @@ -30,3 +30,7 @@ quarkus.log.category."io.opentelemetry".level=DEBUG # Authorization is not configured in this test application a2a.authorization.required=false + +# Allow HTTP and localhost for push notification testing +a2a.push-notification.url.allowed-schemes=http,https +a2a.push-notification.url.allow-private-network-targets=true diff --git a/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/OpenTelemetryA2ABaseTest.java b/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/OpenTelemetryA2ABaseTest.java index 93b9d91b7..6813976c7 100644 --- a/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/OpenTelemetryA2ABaseTest.java +++ b/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/OpenTelemetryA2ABaseTest.java @@ -9,17 +9,28 @@ import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.*; +import com.google.gson.Gson; +import com.google.gson.reflect.TypeToken; import org.a2aproject.sdk.A2A; import org.a2aproject.sdk.client.Client; import org.a2aproject.sdk.client.config.ClientConfig; import org.a2aproject.sdk.client.transport.jsonrpc.JSONRPCTransport; import org.a2aproject.sdk.client.transport.jsonrpc.JSONRPCTransportConfigBuilder; import org.a2aproject.sdk.jsonrpc.common.json.JsonUtil; -import org.a2aproject.sdk.spec.*; +import org.a2aproject.sdk.spec.A2AClientException; +import org.a2aproject.sdk.spec.CancelTaskParams; +import org.a2aproject.sdk.spec.ListTasksParams; +import org.a2aproject.sdk.spec.Task; +import org.a2aproject.sdk.spec.TaskPushNotificationConfig; +import org.a2aproject.sdk.spec.TaskQueryParams; +import org.a2aproject.sdk.spec.TaskState; +import org.a2aproject.sdk.spec.TaskStatus; import io.opentelemetry.api.trace.SpanKind; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.io.IOException; +import java.lang.reflect.Type; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; @@ -28,6 +39,10 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import org.a2aproject.sdk.spec.A2AMethods; +import org.a2aproject.sdk.spec.Message; +import org.a2aproject.sdk.spec.MessageSendParams; +import org.a2aproject.sdk.spec.TextPart; /** * Abstract base class for OpenTelemetry A2A integration tests. @@ -35,6 +50,14 @@ */ abstract class OpenTelemetryA2ABaseTest extends BaseTest { + private static final Gson GSON = new Gson(); + private static final HttpClient TEST_HTTP_CLIENT = HttpClient.newBuilder() + .version(HttpClient.Version.HTTP_2) + .build(); + + private static final String PUSH_TEST_TASK_ID = "push-propagation-task"; + private static final String PUSH_TEST_CONFIG_ID = "otel-push-config"; + protected Client client; protected final int serverPort = 8081; @@ -52,6 +75,17 @@ void setUp() throws A2AClientException { @BeforeEach void reset() { + given().delete("/test/webhook/requests").then().statusCode(HTTP_OK); + given().delete("/test/dispatch-executor/count").then().statusCode(HTTP_OK); + try { + // Pre-test cleanup: remove any push config left behind by a previously failed test + deletePushNotificationConfig(PUSH_TEST_TASK_ID, PUSH_TEST_CONFIG_ID); + } catch (Exception ignored) { + } + clearExportedSpans(); + } + + protected void clearExportedSpans() { await().atMost(5, SECONDS).until(() -> { List> spans = getSpans(); if (spans.isEmpty()) { @@ -77,7 +111,7 @@ void testGetTaskCreatesSpans() throws Exception { .build(); saveTaskInTaskStore(task); - reset(); + clearExportedSpans(); try { Task retrievedTask = client.getTask(new TaskQueryParams(taskId), null); @@ -85,8 +119,7 @@ void testGetTaskCreatesSpans() throws Exception { assertNotNull(retrievedTask); assertEquals(taskId, retrievedTask.id()); - Thread.sleep(5000); - + await().atMost(10, SECONDS).until(() -> !getSpans().isEmpty()); List> spans = getSpans(); System.out.println("We have created spans " + spans); assertFalse(spans.isEmpty(), "Should have created spans for getTask operation"); @@ -114,7 +147,7 @@ void testGetTaskCreatesSpans() throws Exception { @Test void testListTasksCreatesSpans() throws Exception { - reset(); + clearExportedSpans(); ListTasksParams params = new ListTasksParams( null, null, null, null, null, null, null, "" @@ -124,8 +157,7 @@ void testListTasksCreatesSpans() throws Exception { assertNotNull(result); - Thread.sleep(5000); - + await().atMost(10, SECONDS).until(() -> !getSpans().isEmpty()); List> spans = getSpans(); System.out.println("We have created spans " + spans); assertFalse(spans.isEmpty(), "Should have created spans for listTasks"); @@ -155,7 +187,7 @@ void testCancelTaskCreatesSpans() throws Exception { saveTaskInTaskStore(task); ensureQueueForTask(taskId); - reset(); + clearExportedSpans(); try { Task cancelledTask = client.cancelTask(new CancelTaskParams(taskId), null); @@ -163,8 +195,7 @@ void testCancelTaskCreatesSpans() throws Exception { assertNotNull(cancelledTask); assertEquals(TaskState.TASK_STATE_CANCELED, cancelledTask.status().state()); - Thread.sleep(5000); - + await().atMost(10, SECONDS).until(() -> !getSpans().isEmpty()); List> spans = getSpans(); System.out.println("We have created spans " + spans); assertFalse(spans.isEmpty(), "Should have created spans for cancelTask"); @@ -199,13 +230,12 @@ void testSpanAttributes() throws Exception { .build(); saveTaskInTaskStore(task); - reset(); + clearExportedSpans(); try { client.getTask(new TaskQueryParams(taskId), null); - Thread.sleep(5000); - + await().atMost(10, SECONDS).until(() -> !getSpans().isEmpty()); List> spans = getSpans(); System.out.println("We have created spans " + spans); assertFalse(spans.isEmpty()); @@ -325,67 +355,255 @@ void testSendMessageStreamRecordsStreamingDuration() throws Exception { "Histogram should have at least one data point"); } - protected void saveTaskInTaskStore(Task task) throws Exception { - HttpClient httpClient = HttpClient.newBuilder() - .version(HttpClient.Version.HTTP_2) + @Test + void testPushNotificationCarriesTraceContext() throws Exception { + String taskId = PUSH_TEST_TASK_ID; + String contextId = "push-propagation-ctx"; + + Task task = Task.builder() + .id(taskId) + .contextId(contextId) + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)) + .history(Collections.emptyList()) + .artifacts(Collections.emptyList()) .build(); + + saveTaskInTaskStore(task); + clearExportedSpans(); + savePushNotificationConfig(taskId, + TaskPushNotificationConfig.builder() + .id(PUSH_TEST_CONFIG_ID) + .url("http://localhost:" + serverPort + "/test/webhook") + .build()); + ensureQueueForTask(taskId); + + try { + client.cancelTask(new CancelTaskParams(taskId), null); + + await().atMost(10, SECONDS).until(() -> !fetchWebhookRequests().isEmpty()); + + Map webhookHeaders = fetchWebhookRequests().get(0); + String traceparent = assertTraceparentHeader(webhookHeaders); + + // Wait for the SendPushNotification CLIENT span to be exported + await().atMost(10, SECONDS).until(() -> getSpans().stream() + .anyMatch(span -> "SendPushNotification".equals(span.get("name")) + && SpanKind.CLIENT.name().equals(span.get("kind")))); + + List> spans = getSpans(); + Map pushSpan = spans.stream() + .filter(span -> "SendPushNotification".equals(span.get("name"))) + .filter(span -> SpanKind.CLIENT.name().equals(span.get("kind"))) + .findFirst() + .orElseThrow(() -> new AssertionError( + "No CLIENT span found for SendPushNotification. Spans: " + + spans.stream().map(s -> s.get("name")).toList())); + + assertNotNull(pushSpan.get("spanId")); + assertNotNull(pushSpan.get("traceId")); + + String traceIdFromHeader = traceparent.split("-")[1]; + assertEquals(traceIdFromHeader, pushSpan.get("traceId"), + "traceparent trace ID should match the SendPushNotification span's trace ID"); + } finally { + deletePushNotificationConfig(taskId, PUSH_TEST_CONFIG_ID); + deleteTaskInTaskStore(taskId); + } + } + + @Test + void testPushNotificationPreservesCustomHttpClient() throws Exception { + String taskId = "push-custom-client-task"; + String contextId = "push-custom-client-ctx"; + + Task task = Task.builder() + .id(taskId) + .contextId(contextId) + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)) + .history(Collections.emptyList()) + .artifacts(Collections.emptyList()) + .build(); + + saveTaskInTaskStore(task); + clearExportedSpans(); + savePushNotificationConfig(taskId, + TaskPushNotificationConfig.builder() + .id("custom-client-config") + .url("http://localhost:" + serverPort + "/test/webhook") + .build()); + ensureQueueForTask(taskId); + + try { + client.cancelTask(new CancelTaskParams(taskId), null); + + await().atMost(10, SECONDS).until(() -> !fetchWebhookRequests().isEmpty()); + + Map webhookHeaders = fetchWebhookRequests().get(0); + + String customHeader = getHeader(webhookHeaders, CustomHttpClientProducer.CUSTOM_HEADER_NAME); + assertNotNull(customHeader, + "Push notification must carry the custom client's marker header (" + + CustomHttpClientProducer.CUSTOM_HEADER_NAME + "), proving the OTel module " + + "wraps the application's HTTP client. Received headers: " + webhookHeaders.keySet()); + assertEquals(CustomHttpClientProducer.CUSTOM_HEADER_VALUE, customHeader); + + assertTraceparentHeader(webhookHeaders); + } finally { + deletePushNotificationConfig(taskId, "custom-client-config"); + deleteTaskInTaskStore(taskId); + } + } + + @Test + void testPushNotificationUsesCustomDispatchExecutorWithTraceContext() throws Exception { + String taskId = "push-custom-executor-task"; + String contextId = "push-custom-executor-ctx"; + + Task task = Task.builder() + .id(taskId) + .contextId(contextId) + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)) + .history(Collections.emptyList()) + .artifacts(Collections.emptyList()) + .build(); + + saveTaskInTaskStore(task); + clearExportedSpans(); + savePushNotificationConfig(taskId, + TaskPushNotificationConfig.builder() + .id("custom-executor-config") + .url("http://localhost:" + serverPort + "/test/webhook") + .build()); + ensureQueueForTask(taskId); + + try { + client.cancelTask(new CancelTaskParams(taskId), null); + + await().atMost(10, SECONDS).until(() -> !fetchWebhookRequests().isEmpty()); + + Map webhookHeaders = fetchWebhookRequests().get(0); + assertTraceparentHeader(webhookHeaders); + + int executionCount = Integer.parseInt( + given().get("/test/dispatch-executor/count") + .then().statusCode(HTTP_OK) + .extract().body().asString()); + assertTrue(executionCount > 0, + "Custom @Named dispatch executor must be used by the OTel-wrapped executor, " + + "but execution count was 0"); + } finally { + deletePushNotificationConfig(taskId, "custom-executor-config"); + deleteTaskInTaskStore(taskId); + } + } + + private static String getHeader(Map headers, String name) { + return headers.entrySet().stream() + .filter(e -> name.equalsIgnoreCase(e.getKey())) + .map(Map.Entry::getValue) + .findFirst() + .orElse(null); + } + + /** + * Asserts that the webhook headers contain a valid W3C {@code traceparent} header. + * + * @return the traceparent value for further assertions + */ + private static String assertTraceparentHeader(Map webhookHeaders) { + String traceparent = getHeader(webhookHeaders, "traceparent"); + assertNotNull(traceparent, + "Push notification should carry traceparent header. Received headers: " + webhookHeaders.keySet()); + assertTrue(traceparent.matches("00-[a-f0-9]{32}-[a-f0-9]{16}-[a-f0-9]{2}"), + "traceparent should be in W3C format, got: " + traceparent); + return traceparent; + } + + private URI testUri(String path) { + return URI.create("http://localhost:" + serverPort + "/test/" + path); + } + + private HttpResponse sendTestRequest(HttpRequest request) throws IOException, InterruptedException { + return TEST_HTTP_CLIENT.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); + } + + private void sendTestRequestExpecting200(HttpRequest request, String operation) throws IOException, InterruptedException { + HttpResponse response = sendTestRequest(request); + if (response.statusCode() != HTTP_OK) { + throw new RuntimeException(String.format("%s failed! Status: %d, Body: %s", + operation, response.statusCode(), response.body())); + } + } + + protected void saveTaskInTaskStore(Task task) throws Exception { HttpRequest request = HttpRequest.newBuilder() - .uri(URI.create("http://localhost:" + serverPort + "/test/task")) + .uri(testUri("task")) .POST(HttpRequest.BodyPublishers.ofString(JsonUtil.toJson(task))) .header("Content-Type", "application/json") .build(); - - HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); - if (response.statusCode() != 200) { - throw new RuntimeException(String.format("Saving task failed! Status: %d, Body: %s", response.statusCode(), response.body())); - } + sendTestRequestExpecting200(request, "Saving task"); } protected Task getTaskFromTaskStore(String taskId) throws Exception { - HttpClient httpClient = HttpClient.newBuilder() - .version(HttpClient.Version.HTTP_2) - .build(); HttpRequest request = HttpRequest.newBuilder() - .uri(URI.create("http://localhost:" + serverPort + "/test/task/" + taskId)) + .uri(testUri("task/" + taskId)) .GET() .build(); - - HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); + HttpResponse response = sendTestRequest(request); if (response.statusCode() == 404) { return null; } - if (response.statusCode() != 200) { - throw new RuntimeException(String.format("Getting task failed! Status: %d, Body: %s", response.statusCode(), response.body())); + if (response.statusCode() != HTTP_OK) { + throw new RuntimeException(String.format("Getting task failed! Status: %d, Body: %s", + response.statusCode(), response.body())); } return JsonUtil.fromJson(response.body(), Task.class); } protected void deleteTaskInTaskStore(String taskId) throws Exception { - HttpClient httpClient = HttpClient.newBuilder() - .version(HttpClient.Version.HTTP_2) + HttpRequest request = HttpRequest.newBuilder() + .uri(testUri("task/" + taskId)) + .DELETE() + .build(); + sendTestRequestExpecting200(request, "Deleting task"); + } + + protected void savePushNotificationConfig(String taskId, TaskPushNotificationConfig config) throws Exception { + HttpRequest request = HttpRequest.newBuilder() + .uri(testUri("pushconfig/" + taskId)) + .POST(HttpRequest.BodyPublishers.ofString(JsonUtil.toJson(config))) + .header("Content-Type", "application/json") .build(); + sendTestRequestExpecting200(request, "Saving push config"); + } + + protected void deletePushNotificationConfig(String taskId, String configId) throws Exception { HttpRequest request = HttpRequest.newBuilder() - .uri(URI.create(("http://localhost:" + serverPort + "/test/task/" + taskId))) + .uri(testUri("pushconfig/" + taskId + "/" + configId)) .DELETE() .build(); - HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); - if (response.statusCode() != 200) { - throw new RuntimeException(response.statusCode() + ": Deleting task failed!" + response.body()); - } + sendTestRequestExpecting200(request, "Deleting push config"); + } + + private static final Type WEBHOOK_LIST_TYPE = new TypeToken>>() {}.getType(); + + /** + * Fetches captured webhook requests via HTTP, so it works across JVM + * boundaries in {@code @QuarkusIntegrationTest} mode. + */ + protected List> fetchWebhookRequests() { + String json = given().get("/test/webhook/requests") + .then().statusCode(HTTP_OK) + .extract().body().asString(); + return GSON.fromJson(json, WEBHOOK_LIST_TYPE); } protected void ensureQueueForTask(String taskId) throws Exception { - HttpClient httpClient = HttpClient.newBuilder() - .version(HttpClient.Version.HTTP_2) - .build(); HttpRequest request = HttpRequest.newBuilder() - .uri(URI.create("http://localhost:" + serverPort + "/test/queue/ensure/" + taskId)) + .uri(testUri("queue/ensure/" + taskId)) .POST(HttpRequest.BodyPublishers.noBody()) .build(); - HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); - if (response.statusCode() != 200) { - throw new RuntimeException(String.format("Ensuring queue failed! Status: %d, Body: %s", response.statusCode(), response.body())); - } + sendTestRequestExpecting200(request, "Ensuring queue"); } } diff --git a/extras/opentelemetry/integration-tests/src/test/resources/application.properties b/extras/opentelemetry/integration-tests/src/test/resources/application.properties index b5eb37db9..2532a0050 100644 --- a/extras/opentelemetry/integration-tests/src/test/resources/application.properties +++ b/extras/opentelemetry/integration-tests/src/test/resources/application.properties @@ -27,3 +27,7 @@ quarkus.log.category."org.a2aproject.sdk".level=DEBUG # Authorization is not configured in test profiles a2a.authorization.required=false + +# Allow HTTP and localhost for push notification testing +a2a.push-notification.url.allowed-schemes=http,https +a2a.push-notification.url.allow-private-network-targets=true diff --git a/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryA2AHttpClient.java b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryA2AHttpClient.java new file mode 100644 index 000000000..02c8a1f6c --- /dev/null +++ b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryA2AHttpClient.java @@ -0,0 +1,183 @@ +package org.a2aproject.sdk.extras.opentelemetry.server; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.function.Consumer; + +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.context.Context; +import org.a2aproject.sdk.client.http.A2AHttpClient; +import org.a2aproject.sdk.client.http.A2AHttpResponse; +import org.a2aproject.sdk.client.http.ServerSentEvent; +import org.a2aproject.sdk.util.Assert; + +/** + * Delegating {@link A2AHttpClient} that injects W3C trace-context headers + * ({@code traceparent}, {@code tracestate}) into every outbound HTTP request. + * + *

The propagation uses the {@link io.opentelemetry.context.propagation.TextMapPropagator} + * registered with the supplied {@link OpenTelemetry} instance, reading the current + * {@link Context} at call time. This is the same mechanism the client-side + * {@code OpenTelemetryClientPropagatorTransport} uses, applied here to + * server-initiated outbound calls such as push-notification delivery. + * + *

SSE methods: {@code getAsyncSSE} and {@code postAsyncSSE} + * inject context into the initial HTTP request headers for interface completeness, + * but note that SSE callbacks execute on different threads where the originating + * span is typically already closed. Push notification delivery only uses + * synchronous {@code post()}. + * + * @see OpenTelemetryHttpClientProducer + */ +class OpenTelemetryA2AHttpClient implements A2AHttpClient { + + private final A2AHttpClient delegate; + private final OpenTelemetry openTelemetry; + + OpenTelemetryA2AHttpClient(A2AHttpClient delegate, OpenTelemetry openTelemetry) { + this.delegate = Assert.checkNotNullParam("delegate", delegate); + this.openTelemetry = Assert.checkNotNullParam("openTelemetry", openTelemetry); + } + + @Override + public GetBuilder createGet() { + return new PropagatingGetBuilder(delegate.createGet(), openTelemetry); + } + + @Override + public PostBuilder createPost() { + return new PropagatingPostBuilder(delegate.createPost(), openTelemetry); + } + + @Override + public DeleteBuilder createDelete() { + return new PropagatingDeleteBuilder(delegate.createDelete(), openTelemetry); + } + + /** + * Abstract base for propagating builder wrappers. Consolidates the common + * {@link Builder} delegation ({@code url}, {@code addHeaders}, {@code addHeader}) + * and the context injection logic into one place. + * + * @param the concrete propagating builder type (for method chaining) + * @param the delegate builder type + */ + private abstract static class AbstractPropagatingBuilder, D extends Builder> { + + final D delegate; + private final OpenTelemetry openTelemetry; + + AbstractPropagatingBuilder(D delegate, OpenTelemetry openTelemetry) { + this.delegate = delegate; + this.openTelemetry = openTelemetry; + } + + @SuppressWarnings("unchecked") + protected S self() { + return (S) this; + } + + public S url(String s) { + delegate.url(s); + return self(); + } + + public S addHeaders(Map headers) { + delegate.addHeaders(headers); + return self(); + } + + public S addHeader(String name, String value) { + delegate.addHeader(name, value); + return self(); + } + + protected void injectContextIntoDelegate() { + Map propagationHeaders = new HashMap<>(); + openTelemetry.getPropagators().getTextMapPropagator() + .inject(Context.current(), propagationHeaders, (carrier, key, value) -> { + if (carrier != null) { + carrier.put(key, value); + } + }); + delegate.addHeaders(propagationHeaders); + } + } + + private static class PropagatingGetBuilder + extends AbstractPropagatingBuilder + implements GetBuilder { + + PropagatingGetBuilder(GetBuilder delegate, OpenTelemetry openTelemetry) { + super(delegate, openTelemetry); + } + + @Override + public A2AHttpResponse get() throws IOException, InterruptedException { + injectContextIntoDelegate(); + return delegate.get(); + } + + @Override + public CompletableFuture getAsyncSSE( + Consumer messageConsumer, + Consumer errorConsumer, + Runnable completeRunnable) throws IOException, InterruptedException { + injectContextIntoDelegate(); + return delegate.getAsyncSSE(messageConsumer, errorConsumer, completeRunnable); + } + } + + private static class PropagatingPostBuilder + extends AbstractPropagatingBuilder + implements PostBuilder { + + PropagatingPostBuilder(PostBuilder delegate, OpenTelemetry openTelemetry) { + super(delegate, openTelemetry); + } + + @Override + public PostBuilder body(String body) { + delegate.body(body); + return this; + } + + @Override + public PostBuilder followRedirects(boolean follow) { + delegate.followRedirects(follow); + return this; + } + + @Override + public A2AHttpResponse post() throws IOException, InterruptedException { + injectContextIntoDelegate(); + return delegate.post(); + } + + @Override + public CompletableFuture postAsyncSSE( + Consumer messageConsumer, + Consumer errorConsumer, + Runnable completeRunnable) throws IOException, InterruptedException { + injectContextIntoDelegate(); + return delegate.postAsyncSSE(messageConsumer, errorConsumer, completeRunnable); + } + } + + private static class PropagatingDeleteBuilder + extends AbstractPropagatingBuilder + implements DeleteBuilder { + + PropagatingDeleteBuilder(DeleteBuilder delegate, OpenTelemetry openTelemetry) { + super(delegate, openTelemetry); + } + + @Override + public A2AHttpResponse delete() throws IOException, InterruptedException { + injectContextIntoDelegate(); + return delegate.delete(); + } + } +} diff --git a/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryHttpClientProducer.java b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryHttpClientProducer.java new file mode 100644 index 000000000..8defb3cdf --- /dev/null +++ b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryHttpClientProducer.java @@ -0,0 +1,157 @@ +package org.a2aproject.sdk.extras.opentelemetry.server; + +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.context.Context; +import jakarta.annotation.PreDestroy; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.inject.Instance; +import jakarta.enterprise.inject.Produces; +import jakarta.inject.Inject; +import jakarta.inject.Named; +import org.a2aproject.sdk.client.http.A2AHttpClient; +import org.a2aproject.sdk.client.http.A2AHttpClientFactory; +import org.a2aproject.sdk.server.Instrumented; +import org.a2aproject.sdk.server.tasks.BasePushNotificationSender; +import org.jspecify.annotations.Nullable; + +/** + * CDI producer that creates an {@link OpenTelemetryA2AHttpClient} wrapping + * the application's HTTP client delegate with W3C trace-context propagation. + * + *

When this module is on the classpath and CDI discovers this bean, + * {@link BasePushNotificationSender} will use the produced + * {@link Instrumented @Instrumented} client instead of calling + * {@link A2AHttpClientFactory#create()} directly, ensuring that outbound + * push-notification HTTP requests carry {@code traceparent} and + * {@code tracestate} headers. + * + *

The delegate is resolved from CDI first: if the application provides a + * custom unqualified {@link A2AHttpClient} bean (e.g. one configured with + * mTLS, a proxy, or custom timeouts), this producer wraps that bean so the + * application's HTTP configuration is preserved. When no CDI-provided bean + * exists, the delegate falls back to + * {@link A2AHttpClientFactory#createExcluding(String...)} with the + * {@code "cdi"} provider excluded (to avoid the self-delegation cycle + * that would otherwise occur when the CDI provider re-enters the container). + * + *

The produced client is qualified with {@link Instrumented} so it + * does not compete with the application's own {@link A2AHttpClient} bean in + * default CDI resolution. {@link BasePushNotificationSender} prefers a + * {@code @Instrumented} client when one is available. + */ +@ApplicationScoped +public class OpenTelemetryHttpClientProducer { + + /** + * Name of the CDI HTTP client provider to exclude when falling back to the factory, + * preventing the recursive self-delegation cycle described in the class Javadoc. + */ + static final String CDI_PROVIDER_NAME = "cdi"; + + private final OpenTelemetry openTelemetry; + private volatile @Nullable ExecutorService dispatchPool; + + @SuppressWarnings("NullAway") + protected OpenTelemetryHttpClientProducer() { + this.openTelemetry = null; + } + + @Inject + public OpenTelemetryHttpClientProducer(OpenTelemetry openTelemetry) { + this.openTelemetry = openTelemetry; + } + + /** + * Produces an {@link A2AHttpClient} that injects trace-context headers into + * every outbound request. + * + *

The delegate is resolved in order of preference: + *

    + *
  1. An application-provided unqualified {@link A2AHttpClient} CDI bean + * (custom mTLS, proxy, auth, etc.)
    2. + *
  2. A factory-created client via + * {@link A2AHttpClientFactory#createExcluding(String...)} with the + * {@code "cdi"} provider excluded to prevent recursive CDI + * self-delegation
    3. + *
+ * + * @param httpClient CDI lookup handle for the application's unqualified + * {@link A2AHttpClient} bean (injected by the container) + * @return a propagating HTTP client wrapping the resolved delegate + */ + @Produces + @ApplicationScoped + @Instrumented + A2AHttpClient createPropagatingClient(Instance httpClient) { + A2AHttpClient delegate = httpClient.isResolvable() + ? httpClient.get() + : A2AHttpClientFactory.createExcluding(CDI_PROVIDER_NAME); + return new OpenTelemetryA2AHttpClient(delegate, openTelemetry); + } + + /** + * Produces an {@link Executor} that propagates the current OpenTelemetry + * {@link Context} to submitted tasks, ensuring trace context survives + * {@code CompletableFuture.supplyAsync()} boundaries in push notification dispatch. + * + *

The delegate is resolved in order of preference: + *

    + *
  1. An application-provided + * {@link Named @Named}{@code ("a2a.pushNotificationDispatch")} + * {@link Executor} CDI bean (custom thread pool, sizing, etc.)
    2. + *
  2. A dedicated fixed thread pool sized at + * {@code max(4, 2 * availableProcessors)} to isolate I/O-bound webhook + * HTTP calls from {@code ForkJoinPool.commonPool()}
    3. + *
+ * + *

In both cases the delegate is wrapped with + * {@link Context#taskWrapping(Executor)} so that OpenTelemetry context + * propagation is transparent to the application. + * + * @param userExecutor CDI lookup handle for the application's + * {@code @Named("a2a.pushNotificationDispatch")} executor + * @return a context-propagating executor wrapping the resolved delegate + */ + @Produces + @ApplicationScoped + @Instrumented + Executor createContextPropagatingExecutor( + @Named(BasePushNotificationSender.DISPATCH_EXECUTOR_NAME) Instance userExecutor) { + if (userExecutor.isResolvable()) { + return Context.taskWrapping(userExecutor.get()); + } + int threads = Math.max(4, Runtime.getRuntime().availableProcessors() * 2); + AtomicInteger counter = new AtomicInteger(); + ThreadFactory threadFactory = r -> { + Thread t = new Thread(r, "a2a-push-dispatch-" + counter.getAndIncrement()); + t.setDaemon(true); + return t; + }; + ExecutorService pool = Executors.newFixedThreadPool(threads, threadFactory); + dispatchPool = pool; + return Context.taskWrapping(pool); + } + + @PreDestroy + void shutdownDispatchPool() { + ExecutorService pool = dispatchPool; + if (pool != null) { + pool.shutdown(); + try { + if (!pool.awaitTermination(30, TimeUnit.SECONDS)) { + pool.shutdownNow(); + } + } catch (InterruptedException e) { + pool.shutdownNow(); + Thread.currentThread().interrupt(); + } + } + } +} diff --git a/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryPushNotificationSenderDecorator.java b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryPushNotificationSenderDecorator.java new file mode 100644 index 000000000..44217ed8e --- /dev/null +++ b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryPushNotificationSenderDecorator.java @@ -0,0 +1,146 @@ +package org.a2aproject.sdk.extras.opentelemetry.server; + +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.ERROR_TYPE; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.EXTRACT_RESPONSE_SYS_PROPERTY; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_CONTEXT_ID; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_OPERATION_NAME; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_RESPONSE; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_TASK_ID; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.PUSH_NOTIFICATION_DELIVERY_FAILED; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.PUSH_NOTIFICATION_EVENT_KIND; + +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Scope; +import jakarta.annotation.Priority; +import jakarta.decorator.Decorator; +import jakarta.decorator.Delegate; +import jakarta.enterprise.inject.Any; +import jakarta.inject.Inject; +import org.a2aproject.sdk.server.tasks.PushNotificationSender; +import org.a2aproject.sdk.spec.Message; +import org.a2aproject.sdk.spec.StreamingEventKind; +import org.a2aproject.sdk.spec.Task; +import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent; +import org.a2aproject.sdk.spec.TaskStatusUpdateEvent; +import org.jspecify.annotations.Nullable; + +/** + * OpenTelemetry CDI Decorator for {@link PushNotificationSender}. + *

+ * Creates a {@code CLIENT} span named {@code SendPushNotification} for every push + * notification delivery attempt, capturing the task ID, context ID, and event kind + * as span attributes. The span status reflects the actual delivery outcome: + * {@code OK} when all notifications succeed, {@code ERROR} when one or more + * deliveries fail or an exception is thrown. + *

+ * This decorator is registered in the module's {@code META-INF/beans.xml} and is + * therefore active automatically when this module is on the classpath — no additional + * configuration is needed. + */ +@Decorator +@Priority(100) +public abstract class OpenTelemetryPushNotificationSenderDecorator implements PushNotificationSender { + + static final String SEND_PUSH_NOTIFICATION = "SendPushNotification"; + + @Inject + @Delegate + @Any + private PushNotificationSender delegate; + + @Inject + private Tracer tracer; + + private final boolean extractResponse; + + /** + * Default constructor for CDI. Reads {@code EXTRACT_RESPONSE_SYS_PROPERTY} once + * at bean initialisation so the value is not re-evaluated on every notification. + */ + public OpenTelemetryPushNotificationSenderDecorator() { + this.extractResponse = Boolean.getBoolean(EXTRACT_RESPONSE_SYS_PROPERTY); + } + + /** + * Constructor for unit testing. + * + * @param delegate the delegate push notification sender + * @param tracer the tracer to use + * @param extractResponse whether to capture the full event payload as a span attribute + */ + OpenTelemetryPushNotificationSenderDecorator(PushNotificationSender delegate, Tracer tracer, + boolean extractResponse) { + this.delegate = delegate; + this.tracer = tracer; + this.extractResponse = extractResponse; + } + + @Override + public boolean sendNotification(StreamingEventKind event, @Nullable Task taskSnapshot) { + var spanBuilder = tracer.spanBuilder(SEND_PUSH_NOTIFICATION) + .setSpanKind(SpanKind.CLIENT) + .setAttribute(GENAI_OPERATION_NAME, SEND_PUSH_NOTIFICATION) + .setAttribute(PUSH_NOTIFICATION_EVENT_KIND, event.getClass().getSimpleName()); + + EventIds ids = extractIds(event, taskSnapshot); + if (ids.taskId != null) { + spanBuilder.setAttribute(GENAI_TASK_ID, ids.taskId); + } + if (ids.contextId != null) { + spanBuilder.setAttribute(GENAI_CONTEXT_ID, ids.contextId); + } + if (extractResponse) { + spanBuilder.setAttribute(GENAI_RESPONSE, event.toString()); + } + + Span span = spanBuilder.startSpan(); + try (Scope scope = span.makeCurrent()) { + boolean result = delegate.sendNotification(event, taskSnapshot); + if (result) { + span.setStatus(StatusCode.OK); + } else { + span.setAttribute(ERROR_TYPE, PUSH_NOTIFICATION_DELIVERY_FAILED); + span.setStatus(StatusCode.ERROR, "One or more push notification deliveries failed"); + } + return result; + } catch (Error err) { + String errorDescription = err.getMessage() != null ? err.getMessage() : err.getClass().getName(); + span.setAttribute(ERROR_TYPE, errorDescription); + span.setStatus(StatusCode.ERROR, errorDescription); + throw err; + } catch (Exception ex) { + String errorDescription = ex.getMessage() != null ? ex.getMessage() : ex.getClass().getName(); + span.setAttribute(ERROR_TYPE, errorDescription); + span.setStatus(StatusCode.ERROR, errorDescription); + throw ex; + } finally { + span.end(); + } + } + + private record EventIds(@Nullable String taskId, @Nullable String contextId) { + } + + private static EventIds extractIds(StreamingEventKind event, @Nullable Task taskSnapshot) { + if (taskSnapshot != null) { + return new EventIds(taskSnapshot.id(), taskSnapshot.contextId()); + } + if (event instanceof Task task) { + return new EventIds(task.id(), task.contextId()); + } + if (event instanceof Message message) { + return new EventIds(message.taskId(), message.contextId()); + } + if (event instanceof TaskStatusUpdateEvent statusUpdate) { + return new EventIds(statusUpdate.taskId(), statusUpdate.contextId()); + } + if (event instanceof TaskArtifactUpdateEvent artifactUpdate) { + return new EventIds(artifactUpdate.taskId(), artifactUpdate.contextId()); + } + return new EventIds(null, null); + } + +} diff --git a/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml b/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml index a79cffb3c..f775403c9 100644 --- a/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml +++ b/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml @@ -5,5 +5,6 @@ bean-discovery-mode="annotated"> org.a2aproject.sdk.extras.opentelemetry.server.OpenTelemetryRequestHandlerDecorator + org.a2aproject.sdk.extras.opentelemetry.server.OpenTelemetryPushNotificationSenderDecorator diff --git a/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryA2AHttpClientTest.java b/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryA2AHttpClientTest.java new file mode 100644 index 000000000..0d697dae6 --- /dev/null +++ b/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryA2AHttpClientTest.java @@ -0,0 +1,187 @@ +package org.a2aproject.sdk.extras.opentelemetry.server; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.util.Map; + +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import io.opentelemetry.context.propagation.ContextPropagators; +import io.opentelemetry.context.propagation.TextMapPropagator; +import io.opentelemetry.context.propagation.TextMapSetter; +import org.a2aproject.sdk.client.http.A2AHttpClient; +import org.a2aproject.sdk.client.http.A2AHttpResponse; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InOrder; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.mockito.junit.jupiter.MockitoSettings; +import org.mockito.quality.Strictness; + +@ExtendWith(MockitoExtension.class) +@MockitoSettings(strictness = Strictness.LENIENT) +class OpenTelemetryA2AHttpClientTest { + + @Mock + private A2AHttpClient delegate; + + @Mock + private OpenTelemetry openTelemetry; + + @Mock + private ContextPropagators propagators; + + @Mock + private TextMapPropagator textMapPropagator; + + @Mock + private A2AHttpClient.PostBuilder delegatePostBuilder; + + @Mock + private A2AHttpClient.GetBuilder delegateGetBuilder; + + @Mock + private A2AHttpClient.DeleteBuilder delegateDeleteBuilder; + + @Mock + private A2AHttpResponse response; + + private OpenTelemetryA2AHttpClient client; + + @BeforeEach + void setUp() { + when(openTelemetry.getPropagators()).thenReturn(propagators); + when(propagators.getTextMapPropagator()).thenReturn(textMapPropagator); + + when(delegate.createPost()).thenReturn(delegatePostBuilder); + when(delegate.createGet()).thenReturn(delegateGetBuilder); + when(delegate.createDelete()).thenReturn(delegateDeleteBuilder); + + stubBuilderDefaults(delegatePostBuilder); + when(delegatePostBuilder.body(any())).thenReturn(delegatePostBuilder); + when(delegatePostBuilder.followRedirects(any(boolean.class))).thenReturn(delegatePostBuilder); + + stubBuilderDefaults(delegateGetBuilder); + stubBuilderDefaults(delegateDeleteBuilder); + + client = new OpenTelemetryA2AHttpClient(delegate, openTelemetry); + } + + private static > void stubBuilderDefaults(B builder) { + when(builder.url(any())).thenReturn(builder); + when(builder.addHeader(any(), any())).thenReturn(builder); + when(builder.addHeaders(any())).thenReturn(builder); + } + + @Test + @SuppressWarnings("unchecked") + void post_injectsContextBeforeDelegating() throws IOException, InterruptedException { + when(delegatePostBuilder.post()).thenReturn(response); + + A2AHttpClient.PostBuilder builder = client.createPost(); + A2AHttpResponse result = builder + .url("http://example.com/push") + .addHeader("Content-Type", "application/json") + .body("{}") + .post(); + + assertSame(response, result); + + InOrder order = inOrder(textMapPropagator, delegatePostBuilder); + order.verify(textMapPropagator).inject(any(Context.class), any(), any(TextMapSetter.class)); + order.verify(delegatePostBuilder).post(); + } + + @Test + @SuppressWarnings("unchecked") + void post_propagatesCurrentSpanContext() throws IOException, InterruptedException { + when(delegatePostBuilder.post()).thenReturn(response); + + Context testContext = Context.current().with(io.opentelemetry.context.ContextKey.named("test"), "value"); + try (Scope ignored = testContext.makeCurrent()) { + client.createPost() + .url("http://example.com/push") + .body("{}") + .post(); + } + + verify(textMapPropagator).inject(any(Context.class), any(), any(TextMapSetter.class)); + } + + @Test + @SuppressWarnings("unchecked") + void get_injectsContextBeforeDelegating() throws IOException, InterruptedException { + when(delegateGetBuilder.get()).thenReturn(response); + + A2AHttpClient.GetBuilder builder = client.createGet(); + A2AHttpResponse result = builder + .url("http://example.com/api") + .get(); + + assertSame(response, result); + + InOrder order = inOrder(textMapPropagator, delegateGetBuilder); + order.verify(textMapPropagator).inject(any(Context.class), any(), any(TextMapSetter.class)); + order.verify(delegateGetBuilder).get(); + } + + @Test + @SuppressWarnings("unchecked") + void delete_injectsContextBeforeDelegating() throws IOException, InterruptedException { + when(delegateDeleteBuilder.delete()).thenReturn(response); + + A2AHttpClient.DeleteBuilder builder = client.createDelete(); + A2AHttpResponse result = builder + .url("http://example.com/resource") + .delete(); + + assertSame(response, result); + + InOrder order = inOrder(textMapPropagator, delegateDeleteBuilder); + order.verify(textMapPropagator).inject(any(Context.class), any(), any(TextMapSetter.class)); + order.verify(delegateDeleteBuilder).delete(); + } + + @Test + void postBuilder_chainsCorrectly() { + A2AHttpClient.PostBuilder builder = client.createPost(); + + A2AHttpClient.PostBuilder chained = builder + .url("http://example.com") + .addHeader("X-Custom", "value") + .body("{}") + .followRedirects(false); + + assertNotNull(chained); + } + + @Test + @SuppressWarnings("unchecked") + void contextInjection_propagatesHeadersViaAddHeaders() throws IOException, InterruptedException { + when(delegatePostBuilder.post()).thenReturn(response); + + doAnswer(invocation -> { + Map carrier = invocation.getArgument(1); + TextMapSetter> setter = invocation.getArgument(2); + setter.set(carrier, "traceparent", "00-abc-def-01"); + return null; + }).when(textMapPropagator).inject(any(Context.class), any(), any(TextMapSetter.class)); + + client.createPost() + .url("http://example.com") + .body("{}") + .post(); + + verify(delegatePostBuilder).addHeaders(Map.of("traceparent", "00-abc-def-01")); + } +} diff --git a/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryHttpClientProducerTest.java b/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryHttpClientProducerTest.java new file mode 100644 index 000000000..304380012 --- /dev/null +++ b/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryHttpClientProducerTest.java @@ -0,0 +1,108 @@ +package org.a2aproject.sdk.extras.opentelemetry.server; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import io.opentelemetry.api.OpenTelemetry; +import jakarta.enterprise.inject.Instance; +import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicInteger; +import org.a2aproject.sdk.client.http.A2AHttpClient; +import org.junit.jupiter.api.Test; + +class OpenTelemetryHttpClientProducerTest { + + /** + * Verifies that when no application-provided {@link A2AHttpClient} bean + * exists, the producer falls back to a factory-created delegate. + * + *

Note: this test calls the real {@link org.a2aproject.sdk.client.http.A2AHttpClientFactory} + * via {@code ServiceLoader}. The JDK HTTP client provider ({@code http-client} module) must be + * on this module's test classpath for the factory to resolve a delegate. + */ + @Test + @SuppressWarnings("unchecked") + void createPropagatingClient_noUserBean_fallsBackToFactory() { + OpenTelemetry mockOTel = mock(OpenTelemetry.class); + OpenTelemetryHttpClientProducer producer = new OpenTelemetryHttpClientProducer(mockOTel); + + Instance emptyInstance = mock(Instance.class); + when(emptyInstance.isResolvable()).thenReturn(false); + + A2AHttpClient client = producer.createPropagatingClient(emptyInstance); + + assertNotNull(client); + assertInstanceOf(OpenTelemetryA2AHttpClient.class, client, + "Producer must wrap a factory-created delegate in OpenTelemetryA2AHttpClient"); + } + + /** + * Verifies that when an application-provided {@link A2AHttpClient} bean + * exists, the producer wraps it instead of creating a new one from the factory. + */ + @Test + @SuppressWarnings("unchecked") + void createPropagatingClient_withUserBean_wrapsUserClient() { + OpenTelemetry mockOTel = mock(OpenTelemetry.class); + OpenTelemetryHttpClientProducer producer = new OpenTelemetryHttpClientProducer(mockOTel); + + A2AHttpClient userClient = mock(A2AHttpClient.class); + Instance userInstance = mock(Instance.class); + when(userInstance.isResolvable()).thenReturn(true); + when(userInstance.get()).thenReturn(userClient); + + A2AHttpClient client = producer.createPropagatingClient(userInstance); + + assertNotNull(client); + assertInstanceOf(OpenTelemetryA2AHttpClient.class, client, + "Producer must wrap the user's client in OpenTelemetryA2AHttpClient"); + verify(userInstance).get(); + } + + @Test + @SuppressWarnings("unchecked") + void createContextPropagatingExecutor_noUserBean_createsDefaultPool() { + OpenTelemetry mockOTel = mock(OpenTelemetry.class); + OpenTelemetryHttpClientProducer producer = new OpenTelemetryHttpClientProducer(mockOTel); + + Instance emptyInstance = mock(Instance.class); + when(emptyInstance.isResolvable()).thenReturn(false); + + Executor executor = producer.createContextPropagatingExecutor(emptyInstance); + + try { + assertNotNull(executor, "Produced executor must not be null"); + } finally { + producer.shutdownDispatchPool(); + } + } + + @Test + @SuppressWarnings("unchecked") + void createContextPropagatingExecutor_withUserBean_wrapsUserExecutor() { + OpenTelemetry mockOTel = mock(OpenTelemetry.class); + OpenTelemetryHttpClientProducer producer = new OpenTelemetryHttpClientProducer(mockOTel); + + AtomicInteger executionCount = new AtomicInteger(); + Executor userExecutor = command -> { + executionCount.incrementAndGet(); + command.run(); + }; + Instance userInstance = mock(Instance.class); + when(userInstance.isResolvable()).thenReturn(true); + when(userInstance.get()).thenReturn(userExecutor); + + Executor wrappedExecutor = producer.createContextPropagatingExecutor(userInstance); + + assertNotNull(wrappedExecutor, "Produced executor must not be null"); + verify(userInstance).get(); + + wrappedExecutor.execute(() -> {}); + assertEquals(1, executionCount.get(), + "Wrapped executor must delegate to the user-provided executor"); + } +} diff --git a/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryPushNotificationSenderDecoratorTest.java b/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryPushNotificationSenderDecoratorTest.java new file mode 100644 index 000000000..1596c4882 --- /dev/null +++ b/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryPushNotificationSenderDecoratorTest.java @@ -0,0 +1,208 @@ +package org.a2aproject.sdk.extras.opentelemetry.server; + +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.ERROR_TYPE; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_CONTEXT_ID; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_OPERATION_NAME; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_RESPONSE; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_TASK_ID; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.PUSH_NOTIFICATION_DELIVERY_FAILED; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.PUSH_NOTIFICATION_EVENT_KIND; +import static org.a2aproject.sdk.extras.opentelemetry.server.OpenTelemetryPushNotificationSenderDecorator.SEND_PUSH_NOTIFICATION; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanBuilder; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Scope; +import java.util.Collections; +import java.util.List; +import org.a2aproject.sdk.server.tasks.PushNotificationSender; +import org.a2aproject.sdk.spec.Task; +import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent; +import org.a2aproject.sdk.spec.TaskState; +import org.a2aproject.sdk.spec.TaskStatus; +import org.a2aproject.sdk.spec.TaskStatusUpdateEvent; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class OpenTelemetryPushNotificationSenderDecoratorTest { + + @Mock + private PushNotificationSender delegate; + + @Mock + private Tracer tracer; + + @Mock + private SpanBuilder spanBuilder; + + @Mock + private Span span; + + @Mock + private Scope scope; + + private OpenTelemetryPushNotificationSenderDecorator decorator; + + @BeforeEach + void setUp() { + when(tracer.spanBuilder(anyString())).thenReturn(spanBuilder); + when(spanBuilder.setSpanKind(any(SpanKind.class))).thenReturn(spanBuilder); + when(spanBuilder.setAttribute(anyString(), anyString())).thenReturn(spanBuilder); + when(spanBuilder.startSpan()).thenReturn(span); + when(span.makeCurrent()).thenReturn(scope); + + decorator = new TestablePushNotificationSenderDecorator(delegate, tracer); + } + + @Test + void sendNotification_createsSpanWithTaskIdFromSnapshot() { + when(delegate.sendNotification(any(), any())).thenReturn(true); + Task taskSnapshot = Task.builder() + .id("task-1") + .contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)) + .history(Collections.emptyList()) + .artifacts(Collections.emptyList()) + .build(); + TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder() + .taskId("task-1") + .contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)) + .build(); + + decorator.sendNotification(event, taskSnapshot); + + verify(tracer).spanBuilder(SEND_PUSH_NOTIFICATION); + verify(spanBuilder).setSpanKind(SpanKind.CLIENT); + verify(spanBuilder).setAttribute(GENAI_OPERATION_NAME, SEND_PUSH_NOTIFICATION); + verify(spanBuilder).setAttribute(GENAI_TASK_ID, "task-1"); + verify(spanBuilder).setAttribute(GENAI_CONTEXT_ID, "ctx-1"); + verify(span).setStatus(StatusCode.OK); + verify(span).end(); + verify(delegate).sendNotification(event, taskSnapshot); + } + + @Test + void sendNotification_createsSpanWithTaskIdFromEvent() { + when(delegate.sendNotification(any(), any())).thenReturn(true); + org.a2aproject.sdk.spec.Artifact artifact = org.a2aproject.sdk.spec.Artifact.builder() + .artifactId("artifact-id") + .parts(List.of(new org.a2aproject.sdk.spec.TextPart("result"))) + .build(); + TaskArtifactUpdateEvent event = new TaskArtifactUpdateEvent( + "task-event-1", artifact, "ctx-event-1", null, null, null); + + decorator.sendNotification(event, null); + + verify(spanBuilder).setAttribute(GENAI_TASK_ID, "task-event-1"); + verify(spanBuilder).setAttribute(GENAI_CONTEXT_ID, "ctx-event-1"); + verify(spanBuilder, never()).setAttribute(eq(GENAI_TASK_ID), eq((String) null)); + verify(span).setStatus(StatusCode.OK); + verify(span).end(); + } + + @Test + void sendNotification_setsErrorStatusOnDeliveryFailure() { + TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder() + .taskId("task-fail") + .contextId("ctx-fail") + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)) + .build(); + when(delegate.sendNotification(any(), any())).thenReturn(false); + + boolean result = decorator.sendNotification(event, null); + + assertFalse(result); + verify(span).setAttribute(ERROR_TYPE, PUSH_NOTIFICATION_DELIVERY_FAILED); + verify(span).setStatus(StatusCode.ERROR, "One or more push notification deliveries failed"); + verify(span).end(); + } + + @Test + void sendNotification_setsErrorStatusOnException() { + TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder() + .taskId("task-err") + .contextId("ctx-err") + .status(new TaskStatus(TaskState.TASK_STATE_FAILED)) + .build(); + RuntimeException ex = new RuntimeException("webhook unreachable"); + doThrow(ex).when(delegate).sendNotification(any(), any()); + + assertThrows(RuntimeException.class, () -> decorator.sendNotification(event, null)); + + verify(span).setStatus(StatusCode.ERROR, "webhook unreachable"); + verify(span).setAttribute(ERROR_TYPE, "webhook unreachable"); + verify(span).end(); + } + + @Test + void sendNotification_setsEventKindAttribute() { + when(delegate.sendNotification(any(), any())).thenReturn(true); + TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder() + .taskId("task-kind") + .contextId("ctx-kind") + .status(new TaskStatus(TaskState.TASK_STATE_COMPLETED)) + .build(); + + decorator.sendNotification(event, null); + + verify(spanBuilder).setAttribute(PUSH_NOTIFICATION_EVENT_KIND, "TaskStatusUpdateEvent"); + } + + @Test + void sendNotification_extractsResponseWhenEnabled() { + when(delegate.sendNotification(any(), any())).thenReturn(true); + OpenTelemetryPushNotificationSenderDecorator decoratorWithExtract = + new TestablePushNotificationSenderDecorator(delegate, tracer, true); + TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder() + .taskId("task-resp") + .contextId("ctx-resp") + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)) + .build(); + + decoratorWithExtract.sendNotification(event, null); + + verify(spanBuilder).setAttribute(eq(GENAI_RESPONSE), anyString()); + } + + @Test + void sendNotification_noResponseAttributeWhenDisabled() { + when(delegate.sendNotification(any(), any())).thenReturn(true); + TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder() + .taskId("task-noresp") + .contextId("ctx-noresp") + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)) + .build(); + + // decorator is constructed with extractResponse=false (setUp default) + decorator.sendNotification(event, null); + + verify(spanBuilder, never()).setAttribute(eq(GENAI_RESPONSE), anyString()); + } + + static class TestablePushNotificationSenderDecorator extends OpenTelemetryPushNotificationSenderDecorator { + TestablePushNotificationSenderDecorator(PushNotificationSender delegate, Tracer tracer) { + super(delegate, tracer, false); + } + + TestablePushNotificationSenderDecorator(PushNotificationSender delegate, Tracer tracer, + boolean extractResponse) { + super(delegate, tracer, extractResponse); + } + } +} diff --git a/extras/push-notification-config-store-database-jpa/src/test/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/MockPushNotificationSender.java b/extras/push-notification-config-store-database-jpa/src/test/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/MockPushNotificationSender.java index d431f934c..2460f20c7 100644 --- a/extras/push-notification-config-store-database-jpa/src/test/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/MockPushNotificationSender.java +++ b/extras/push-notification-config-store-database-jpa/src/test/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/MockPushNotificationSender.java @@ -22,9 +22,15 @@ public class MockPushNotificationSender implements PushNotificationSender { private final Queue capturedEvents = new ConcurrentLinkedQueue<>(); + /** + * Captures the event for later assertion and always returns {@code true}. + * The return value is not meaningful in this test double — it exists only to satisfy + * the {@link org.a2aproject.sdk.server.tasks.PushNotificationSender} contract. + */ @Override - public void sendNotification(StreamingEventKind event, Task taskSnapshot) { + public boolean sendNotification(StreamingEventKind event, Task taskSnapshot) { capturedEvents.add(event); + return true; } public Queue getCapturedEvents() { diff --git a/extras/queue-manager-replicated/core/src/test/java/org/a2aproject/sdk/extras/queuemanager/replicated/core/ReplicatedQueueManagerTest.java b/extras/queue-manager-replicated/core/src/test/java/org/a2aproject/sdk/extras/queuemanager/replicated/core/ReplicatedQueueManagerTest.java index daced98d2..6c837ba4c 100644 --- a/extras/queue-manager-replicated/core/src/test/java/org/a2aproject/sdk/extras/queuemanager/replicated/core/ReplicatedQueueManagerTest.java +++ b/extras/queue-manager-replicated/core/src/test/java/org/a2aproject/sdk/extras/queuemanager/replicated/core/ReplicatedQueueManagerTest.java @@ -43,8 +43,6 @@ class ReplicatedQueueManagerTest { private StreamingEventKind testEvent; private MainEventBus mainEventBus; private MainEventBusProcessor mainEventBusProcessor; - private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = (event, snapshot) -> {}; - @BeforeEach void setUp() { // Create MainEventBus first @@ -59,7 +57,7 @@ void setUp() { ); // Create MainEventBusProcessor with QueueManager - mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, NOOP_PUSHNOTIFICATION_SENDER, queueManager); + mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(mainEventBusProcessor); testEvent = TaskStatusUpdateEvent.builder() diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientFactory.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientFactory.java index 95b45fbc6..8c4f2d363 100644 --- a/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientFactory.java +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientFactory.java @@ -3,6 +3,8 @@ import java.util.Comparator; import java.util.List; import java.util.ServiceLoader; +import java.util.Set; +import java.util.function.Predicate; import java.util.logging.Level; import java.util.logging.Logger; import java.util.stream.Stream; @@ -72,7 +74,32 @@ private A2AHttpClientFactory() { * @throws IllegalStateException if no provider found or all providers failed to instantiate */ public static A2AHttpClient create() { + return createInternal(p -> true); + } + + /** + * Creates a new A2AHttpClient instance, skipping any provider whose name is in + * {@code excludedProviderNames}. + * + *

This overload exists to break the CDI self-delegation cycle: a CDI producer method + * that itself produces an {@link A2AHttpClient} must not call the plain {@link #create()}, + * because the CDI provider (priority 200) would re-enter the container and return the + * proxy being constructed — causing infinite recursion. Calling + * {@code createExcluding("cdi")} bypasses the CDI provider and selects the next eligible + * implementation (e.g. Vert.x at priority 100, or JDK at priority 0). + * + * @param excludedProviderNames names of providers to skip (see {@link A2AHttpClientProvider#name()}) + * @return a new A2AHttpClient instance from the first non-excluded provider + * @throws IllegalStateException if no remaining provider could be instantiated + */ + public static A2AHttpClient createExcluding(String... excludedProviderNames) { + Set excluded = Set.of(excludedProviderNames); + return createInternal(p -> !excluded.contains(p.name())); + } + + private static A2AHttpClient createInternal(Predicate filter) { return PROVIDERS.stream() + .filter(filter) .flatMap(p -> { try { return Stream.of(p.create()); diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientFactoryTest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientFactoryTest.java index fdb34bd66..e26fea55c 100644 --- a/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientFactoryTest.java +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientFactoryTest.java @@ -102,6 +102,22 @@ public void testCreateWithEmptyProviderNameThrows() { ); } + @Test + public void testCreateExcludingUnknownNameIgnoresFilter() { + A2AHttpClient client = A2AHttpClientFactory.createExcluding("nonexistent-provider"); + assertNotNull(client); + assertInstanceOf(JdkA2AHttpClient.class, client, + "createExcluding an unknown name should still return the JDK client"); + } + + @Test + public void testCreateExcludingAllAvailableProvidersThrows() { + assertThrows( + IllegalStateException.class, + () -> A2AHttpClientFactory.createExcluding("jdk"), + "Factory should throw when every available provider is excluded"); + } + @Test public void testCreatedClientIsUsable() { A2AHttpClient client = A2AHttpClientFactory.create(); diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/Instrumented.java b/server-common/src/main/java/org/a2aproject/sdk/server/Instrumented.java new file mode 100644 index 000000000..6da45e4d6 --- /dev/null +++ b/server-common/src/main/java/org/a2aproject/sdk/server/Instrumented.java @@ -0,0 +1,52 @@ +package org.a2aproject.sdk.server; + +import static java.lang.annotation.ElementType.FIELD; +import static java.lang.annotation.ElementType.METHOD; +import static java.lang.annotation.ElementType.PARAMETER; +import static java.lang.annotation.ElementType.TYPE; +import static java.lang.annotation.RetentionPolicy.RUNTIME; + +import java.lang.annotation.Retention; +import java.lang.annotation.Target; + +import jakarta.enterprise.util.AnnotationLiteral; +import jakarta.inject.Qualifier; + +/** + * CDI qualifier for beans that have been wrapped with observability + * instrumentation (e.g. distributed tracing, context propagation, metrics). + * + *

When an instrumentation module (such as the OpenTelemetry server extra) is + * on the classpath, it may produce: + *

    + *
  • An {@code @Instrumented} {@link org.a2aproject.sdk.client.http.A2AHttpClient} + * that wraps the application's own HTTP client (or the default factory-created one) + * with trace-context header injection.
    • + *
  • An {@code @Instrumented} {@link java.util.concurrent.Executor} that wraps a + * user-provided {@link jakarta.inject.Named @Named} executor (or a default + * thread pool) with OpenTelemetry context propagation.
    • + *
+ * + *

{@link org.a2aproject.sdk.server.tasks.BasePushNotificationSender} prefers + * {@code @Instrumented} beans when available, falling back to unqualified / + * {@code @Named} beans, and finally to built-in defaults. + * + *

This qualifier lets instrumentation modules add behaviour transparently, + * without replacing the application's custom configuration (mTLS, proxy, + * authentication, custom thread pools, etc.). + * + * @see org.a2aproject.sdk.server.tasks.BasePushNotificationSender + */ +@Qualifier +@Retention(RUNTIME) +@Target({FIELD, TYPE, METHOD, PARAMETER}) +public @interface Instrumented { + + /** + * Supports programmatic CDI lookup of {@code @Instrumented} beans via + * {@code CDI.current().select(A2AHttpClient.class, Instrumented.Literal.INSTANCE)}. + */ + final class Literal extends AnnotationLiteral implements Instrumented { + public static final Literal INSTANCE = new Literal(); + } +} diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java index c0bf0e92b..5a8e32cc7 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java @@ -410,7 +410,10 @@ private void sendPushNotification(String taskId, StreamingEventKind event, @Null try { if (event != null) { LOGGER.debug("Sending push notification for task {}", taskId); - pushSender.sendNotification(event, taskSnapshot); + boolean sent = pushSender.sendNotification(event, taskSnapshot); + if (!sent) { + LOGGER.warn("One or more push notification deliveries failed for task {}", taskId); + } } else { LOGGER.debug("Skipping push notification - event is null for task {}", taskId); } diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/BasePushNotificationSender.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/BasePushNotificationSender.java index 9dff9da60..a20db3cef 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/BasePushNotificationSender.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/BasePushNotificationSender.java @@ -7,6 +7,7 @@ import jakarta.enterprise.context.ApplicationScoped; import jakarta.enterprise.inject.Instance; import jakarta.inject.Inject; +import jakarta.inject.Named; import java.io.IOException; import java.util.ArrayList; @@ -15,10 +16,14 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.ForkJoinPool; import org.a2aproject.sdk.client.http.A2AHttpClient; import org.a2aproject.sdk.client.http.A2AHttpClientFactory; +import org.a2aproject.sdk.client.http.A2AHttpResponse; import org.a2aproject.sdk.jsonrpc.common.json.JsonUtil; +import org.a2aproject.sdk.server.Instrumented; import org.a2aproject.sdk.spec.AuthenticationInfo; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult; @@ -38,6 +43,26 @@ public class BasePushNotificationSender implements PushNotificationSender { private static final Logger LOGGER = LoggerFactory.getLogger(BasePushNotificationSender.class); public static final int DEFAULT_PAGE_SIZE = 100; + /** + * CDI {@link Named @Named} qualifier ({@value}) for an optional application-provided + * {@link Executor} used to dispatch push notifications asynchronously. + * + *

Applications can produce a bean with this name to supply a custom thread pool + * (e.g. for sizing or monitoring). When an instrumentation module (such as the + * OpenTelemetry server extra) is present, it produces an + * {@link Instrumented @Instrumented} {@link Executor} that wraps the application's + * {@code @Named} executor (or a default pool) with context propagation. + * + *

Resolution order (see the {@link #BasePushNotificationSender CDI constructor}): + *

    + *
  1. {@link Instrumented @Instrumented} {@link Executor} (instrumented wrapper)
    2. + *
  2. {@code @Named("a2a.pushNotificationDispatch")} {@link Executor} + * (application-provided)
    3. + *
  3. {@link ForkJoinPool#commonPool()} as last resort
    4. + *
+ */ + public static final String DISPATCH_EXECUTOR_NAME = "a2a.pushNotificationDispatch"; + // Fields set by constructor injection cannot be final. We need a noargs constructor for // Jakarta compatibility, and it seems that making fields set by constructor injection // final, is not proxyable in all runtimes @@ -45,6 +70,7 @@ public class BasePushNotificationSender implements PushNotificationSender { private PushNotificationConfigStore configStore; private Map formattersByVersion; private PushNotificationUrlValidator urlValidator; + private Executor dispatchExecutor = ForkJoinPool.commonPool(); /** @@ -71,14 +97,51 @@ public BasePushNotificationSender(PushNotificationConfigStore configStore, this.urlValidator = urlValidator; } + /** + * CDI injection constructor. + * + *

The HTTP client is resolved in order of preference: + *

    + *
  1. A {@link Instrumented @Instrumented} bean (e.g. one + * produced by the OpenTelemetry server extra, which wraps the + * application's client with trace-context injection)
    2. + *
  2. An unqualified {@link A2AHttpClient} bean provided by the + * application (custom mTLS, proxy, authentication, etc.)
    3. + *
  3. {@link A2AHttpClientFactory#create()} as last resort
    4. + *
+ * + *

The dispatch executor is resolved in order of preference: + *

    + *
  1. An {@link Instrumented @Instrumented} {@link Executor} (e.g. one + * produced by the OpenTelemetry server extra, which wraps the + * application's executor with context propagation)
    2. + *
  2. A {@link Named @Named}{@code ("a2a.pushNotificationDispatch")} + * {@link Executor} provided by the application
    3. + *
  3. {@link ForkJoinPool#commonPool()} as last resort
    4. + *
+ * + * @param configStore the store for push notification configurations + * @param formatters optional formatters for protocol-version-specific payloads + * @param urlValidator validates push notification URLs to prevent SSRF + * @param httpClient optional CDI-provided HTTP client + * @param propagatingClient optional context-propagating HTTP client that wraps the + * application's client with observability instrumentation + * @param instrumentedExecutor optional instrumented executor with context propagation + * @param dispatchExecutor optional named executor for async dispatch + */ @Inject public BasePushNotificationSender(PushNotificationConfigStore configStore, Instance formatters, - PushNotificationUrlValidator urlValidator) { - this.httpClient = A2AHttpClientFactory.create(); + PushNotificationUrlValidator urlValidator, + Instance httpClient, + @Instrumented Instance propagatingClient, + @Instrumented Instance instrumentedExecutor, + @Named(DISPATCH_EXECUTOR_NAME) Instance dispatchExecutor) { + this.httpClient = resolveHttpClient(propagatingClient, httpClient); this.configStore = configStore; this.formattersByVersion = toFormatterMap(formatters); this.urlValidator = urlValidator; + this.dispatchExecutor = resolveExecutor(instrumentedExecutor, dispatchExecutor); } public BasePushNotificationSender(PushNotificationConfigStore configStore, A2AHttpClient httpClient, @@ -98,6 +161,28 @@ public BasePushNotificationSender(PushNotificationConfigStore configStore, A2AHt this.urlValidator = urlValidator; } + private static A2AHttpClient resolveHttpClient(Instance propagating, + Instance plain) { + if (propagating.isResolvable()) { + return propagating.get(); + } + if (plain.isResolvable()) { + return plain.get(); + } + return A2AHttpClientFactory.create(); + } + + private static Executor resolveExecutor(Instance instrumented, + Instance named) { + if (instrumented.isResolvable()) { + return instrumented.get(); + } + if (named.isResolvable()) { + return named.get(); + } + return ForkJoinPool.commonPool(); + } + private static Map toFormatterMap( Iterable formatters) { Map map = new HashMap<>(); @@ -108,11 +193,10 @@ private static Map toFormatterMap( } @Override - public void sendNotification(StreamingEventKind event, @Nullable Task taskSnapshot) { + public boolean sendNotification(StreamingEventKind event, @Nullable Task taskSnapshot) { String taskId = extractTaskId(event); if (taskId == null) { - LOGGER.warn("Cannot send push notification: event does not contain taskId"); - return; + throw new IllegalArgumentException("Cannot send push notification: event does not contain taskId"); } List configs = new ArrayList<>(); @@ -126,6 +210,11 @@ public void sendNotification(StreamingEventKind event, @Nullable Task taskSnapsh nextPageToken = pageResult.nextPageToken(); } while (nextPageToken != null); + if (configs.isEmpty()) { + LOGGER.debug("No push notification configs registered for taskId: {}", taskId); + return true; + } + Map versionsByConfigId = configStore.getProtocolVersions(taskId); List> dispatchResults = configs @@ -138,18 +227,27 @@ public void sendNotification(StreamingEventKind event, @Nullable Task taskSnapsh try { boolean allSent = dispatchResult.get(); if (!allSent) { - LOGGER.warn("Some push notifications failed to send for taskId: " + taskId); + LOGGER.warn("Some push notifications failed to send for taskId: {}", taskId); } - } catch (InterruptedException | ExecutionException e) { - LOGGER.warn("Some push notifications failed to send for taskId " + taskId + ": {}", e.getMessage(), e); + return allSent; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.warn("Some push notifications failed to send for taskId {}: {}", taskId, e.getMessage(), e); + return false; + } catch (ExecutionException e) { + LOGGER.warn("Some push notifications failed to send for taskId {}: {}", taskId, e.getMessage(), e); + return false; } } /** - * Extracts the task ID from a StreamingEventKind event. + * Extracts the task ID from a streaming event. * * @param event the streaming event - * @return the task ID, or null if not available + * @return the task ID, or {@code null} if the event type does not carry one + * (e.g. a {@link Message} without a task ID) + * @throws IllegalStateException if the event is not a recognized + * {@link StreamingEventKind} subtype */ protected @Nullable String extractTaskId(StreamingEventKind event) { if (event instanceof Task task) { @@ -171,7 +269,8 @@ private CompletableFuture dispatch(StreamingEventKind event, @Nullable Task taskSnapshot, TaskPushNotificationConfig pushInfo, Map versionsByConfigId) { - return CompletableFuture.supplyAsync(() -> dispatchNotification(event, taskSnapshot, pushInfo, versionsByConfigId)); + return CompletableFuture.supplyAsync( + () -> dispatchNotification(event, taskSnapshot, pushInfo, versionsByConfigId), dispatchExecutor); } private boolean dispatchNotification(StreamingEventKind event, @@ -243,13 +342,18 @@ private boolean dispatchNotification(StreamingEventKind event, } try { - postBuilder + A2AHttpResponse response = postBuilder .url(url) .addHeader(CONTENT_TYPE, APPLICATION_JSON) .body(body) .post(); + if (!response.success()) { + LOGGER.warn("Push notification [config={}] to {} failed with HTTP status {}", + pushInfo.id(), url, response.status()); + return false; + } } catch (IOException | InterruptedException e) { - LOGGER.debug("Error pushing data to " + url + ": {}", e.getMessage(), e); + LOGGER.warn("Push notification [config={}] to {} failed: {}", pushInfo.id(), url, e.getMessage(), e); return false; } return true; diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/PushNotificationSender.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/PushNotificationSender.java index 19d2886d0..301f3ab2e 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/PushNotificationSender.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/PushNotificationSender.java @@ -55,9 +55,10 @@ * KafkaProducer producer; * * @Override - * public void sendNotification(StreamingEventKind event, Task taskSnapshot) { + * public boolean sendNotification(StreamingEventKind event, Task taskSnapshot) { * String taskId = extractTaskId(event); * producer.send("task-updates", taskId, event); + * return true; * } * } * } @@ -65,12 +66,18 @@ *

Error Handling

* Implementations should handle errors gracefully: *
    - *
  • Log failures but don't throw exceptions (notifications are best-effort)
    • + *
  • Return {@code false} on delivery failures (network errors, HTTP + * errors, unreachable endpoints) rather than throwing exceptions — + * notifications are best-effort
    • + *
  • Throw {@link IllegalArgumentException} or {@link IllegalStateException} + * for precondition violations (e.g. an event without a task ID, or + * an unsupported event type) — these indicate caller bugs, not delivery + * problems
    • *
  • Don't block on network I/O - execute asynchronously if needed
    • *
  • Circuit breaker patterns for repeatedly failing endpoints
    • *
- * Throwing exceptions from this method will not fail the client request, but will - * be logged as errors. + * Exceptions thrown from this method will not fail the client request — callers + * catch and log them as errors. * *

Thread Safety

* May be called from multiple threads concurrently for different tasks. @@ -82,6 +89,12 @@ */ public interface PushNotificationSender { + /** + * A no-op sender that always reports success. Useful in tests and + * configurations where push notifications are not needed. + */ + PushNotificationSender NOOP = (event, taskSnapshot) -> true; + /** * Sends a push notification containing a streaming event. *

@@ -105,14 +118,20 @@ public interface PushNotificationSender { *

  • {@link org.a2aproject.sdk.spec.TaskArtifactUpdateEvent}
    • * *

    - * Error Handling: Log errors but don't throw exceptions. Notifications are - * best-effort and should not fail the primary request. + * Error Handling: Return {@code false} on delivery failures (network + * errors, HTTP errors). Throwing is reserved for precondition violations such + * as a missing task ID or an unsupported event type — these signal caller bugs. + * Callers catch all exceptions so neither case fails the client request. *

    * * @param event the streaming event to send. * @param taskSnapshot the current state of the task after the event has been applied. * Used by formatters for older protocol versions that require * the full task state in notifications. + * @return {@code true} if all notifications were delivered successfully, + * {@code false} if one or more deliveries failed + * @throws IllegalArgumentException if the event does not contain a task ID + * @throws IllegalStateException if the event type is not recognized */ - void sendNotification(StreamingEventKind event, @Nullable Task taskSnapshot); + boolean sendNotification(StreamingEventKind event, @Nullable Task taskSnapshot); } diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/events/EventConsumerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/events/EventConsumerTest.java index 54882a078..d290a2dcf 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/events/EventConsumerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/events/EventConsumerTest.java @@ -35,7 +35,6 @@ public class EventConsumerTest { - private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = (event, snapshot) -> {}; private static final String TASK_ID = "123"; // Must match MINIMAL_TASK id private EventQueue eventQueue; @@ -65,7 +64,7 @@ public void init() { InMemoryTaskStore taskStore = new InMemoryTaskStore(); mainEventBus = new MainEventBus(); InMemoryQueueManager queueManager = new InMemoryQueueManager(taskStore, mainEventBus); - mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, NOOP_PUSHNOTIFICATION_SENDER, queueManager); + mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(mainEventBusProcessor); eventQueue = EventQueueUtil.getEventQueueBuilder(mainEventBus) diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/events/EventQueueTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/events/EventQueueTest.java index 98953de05..22747108a 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/events/EventQueueTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/events/EventQueueTest.java @@ -67,15 +67,13 @@ public class EventQueueTest { } """; - private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = (event, snapshot) -> {}; - @BeforeEach public void init() { // Set up MainEventBus and processor for production-like test environment InMemoryTaskStore taskStore = new InMemoryTaskStore(); mainEventBus = new MainEventBus(); InMemoryQueueManager queueManager = new InMemoryQueueManager(taskStore, mainEventBus); - mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, NOOP_PUSHNOTIFICATION_SENDER, queueManager); + mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(mainEventBusProcessor); eventQueue = EventQueueUtil.getEventQueueBuilder(mainEventBus) diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/events/InMemoryQueueManagerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/events/InMemoryQueueManagerTest.java index 7a16c9074..ced36e695 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/events/InMemoryQueueManagerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/events/InMemoryQueueManagerTest.java @@ -29,15 +29,13 @@ public class InMemoryQueueManagerTest { private InMemoryTaskStore taskStore; private MainEventBus mainEventBus; private MainEventBusProcessor mainEventBusProcessor; - private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = (event, snapshot) -> {}; - @BeforeEach public void setUp() { taskStateProvider = new MockTaskStateProvider(); taskStore = new InMemoryTaskStore(); mainEventBus = new MainEventBus(); queueManager = new InMemoryQueueManager(taskStateProvider, mainEventBus); - mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, NOOP_PUSHNOTIFICATION_SENDER, queueManager); + mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(mainEventBusProcessor); } diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/events/MainEventBusProcessorExceptionTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/events/MainEventBusProcessorExceptionTest.java index d4e97fdc3..c2b45c3ef 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/events/MainEventBusProcessorExceptionTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/events/MainEventBusProcessorExceptionTest.java @@ -45,7 +45,6 @@ */ public class MainEventBusProcessorExceptionTest { - private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = (event, snapshot) -> {}; private static final String TASK_ID = "test-task-123"; private MainEventBus mainEventBus; @@ -64,7 +63,7 @@ public void setUp() { mainEventBus = new MainEventBus(); queueManager = new InMemoryQueueManager(null, mainEventBus); // null TaskStateProvider for tests mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, mockTaskStore, - NOOP_PUSHNOTIFICATION_SENDER, queueManager); + PushNotificationSender.NOOP, queueManager); // Set up log capture for verifying error messages Logger logger = (Logger) LoggerFactory.getLogger(MainEventBusProcessor.class); diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/events/MainEventBusProcessorPushNotificationOrderTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/events/MainEventBusProcessorPushNotificationOrderTest.java index 655973e85..01337f478 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/events/MainEventBusProcessorPushNotificationOrderTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/events/MainEventBusProcessorPushNotificationOrderTest.java @@ -67,6 +67,7 @@ public void testEventsForSameTaskDeliveredInOrder() throws InterruptedException } deliveryOrder.add(index); latch.countDown(); + return true; }; mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, sender, queueManager); @@ -115,6 +116,7 @@ public void testDifferentTasksAreIndependent() throws InterruptedException { fastDeliveredWhileSlowBlocked.set(slowLatch.getCount() > 0 ? 1 : 0); fastLatch.countDown(); } + return true; }; mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, sender, queueManager); @@ -157,7 +159,10 @@ public void testSynchronousExecutorDoesNotLeakChainEntries() throws InterruptedE // 3. Assert on the captured count, not a fresh read. int taskCount = 10; CountDownLatch latch = new CountDownLatch(taskCount * 2); - PushNotificationSender sender = (event, snapshot) -> latch.countDown(); + PushNotificationSender sender = (event, snapshot) -> { + latch.countDown(); + return true; + }; mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, sender, queueManager); mainEventBusProcessor.setPushNotificationExecutor(Runnable::run); @@ -193,6 +198,44 @@ public void testSynchronousExecutorDoesNotLeakChainEntries() throws InterruptedE "A completed task's chain entry must not be left in the map"); } + @Test + public void testDeliveryFailureDoesNotStopSubsequentNotifications() throws InterruptedException { + // When sendNotification() returns false, the processor must continue delivering + // subsequent events rather than stopping or throwing. + String taskId = "failure-resilience-task"; + int eventCount = 5; + CountDownLatch latch = new CountDownLatch(eventCount); + AtomicInteger attemptCount = new AtomicInteger(0); + + PushNotificationSender sender = (event, snapshot) -> { + int attempt = attemptCount.incrementAndGet(); + latch.countDown(); + // Alternate: fail on even-indexed attempts to prove the chain continues. + return attempt % 2 != 0; + }; + + mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, sender, queueManager); + EventQueueUtil.start(mainEventBusProcessor); + EventQueue eventQueue = EventQueueUtil.getEventQueueBuilder(mainEventBus) + .taskId(taskId) + .mainEventBus(mainEventBus) + .build().tap(); + + eventQueue.enqueueEvent(Task.builder() + .id(taskId) + .contextId(CONTEXT_ID) + .status(new TaskStatus(TaskState.TASK_STATE_SUBMITTED)) + .build()); + for (int i = 0; i < eventCount - 1; i++) { + eventQueue.enqueueEvent(statusEventWithSequence(taskId, i)); + } + + assertTrue(latch.await(10, TimeUnit.SECONDS), + "All notifications must be attempted even when some deliveries return false"); + assertEquals(eventCount, attemptCount.get(), + "Every event must be passed to the sender regardless of prior delivery failures"); + } + private TaskStatusUpdateEvent statusEventWithSequence(String taskId, int sequence) { return new TaskStatusUpdateEvent(taskId, new TaskStatus(TaskState.TASK_STATE_WORKING, null, null), CONTEXT_ID, java.util.Map.of("sequence", sequence)); diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/AbstractA2ARequestHandlerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/AbstractA2ARequestHandlerTest.java index 313e2d742..91e2a3865 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/AbstractA2ARequestHandlerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/AbstractA2ARequestHandlerTest.java @@ -34,7 +34,6 @@ import org.a2aproject.sdk.server.tasks.InMemoryPushNotificationConfigStore; import org.a2aproject.sdk.server.tasks.InMemoryTaskStore; import org.a2aproject.sdk.server.tasks.PushNotificationConfigStore; -import org.a2aproject.sdk.server.tasks.PushNotificationSender; import org.a2aproject.sdk.server.tasks.PushNotificationUrlValidator; import org.a2aproject.sdk.server.tasks.TaskStore; import org.a2aproject.sdk.spec.A2AError; @@ -71,8 +70,6 @@ public class AbstractA2ARequestHandlerTest { private static final String PREFERRED_TRANSPORT = "preferred-transport"; private static final String A2A_REQUESTHANDLER_TEST_PROPERTIES = "/a2a-requesthandler-test.properties"; - private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = (event, snapshot) -> {}; - protected AgentExecutor executor; protected TaskStore taskStore; protected RequestHandler requestHandler; diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerBuilderAuthorizationTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerBuilderAuthorizationTest.java index 172f13f94..a8e390722 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerBuilderAuthorizationTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerBuilderAuthorizationTest.java @@ -57,8 +57,6 @@ class DefaultRequestHandlerBuilderAuthorizationTest { .acceptedOutputModes(List.of()) .build(); - private static final PushNotificationSender NOOP_PUSH_SENDER = (event, snapshot) -> {}; - private InMemoryTaskStore taskStore; private InMemoryQueueManager queueManager; private MainEventBusProcessor mainEventBusProcessor; @@ -73,7 +71,7 @@ void setUp() { taskStore = new InMemoryTaskStore(); MainEventBus mainEventBus = new MainEventBus(); queueManager = new InMemoryQueueManager(taskStore, mainEventBus); - mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, NOOP_PUSH_SENDER, queueManager); + mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(mainEventBusProcessor); executor = Executors.newCachedThreadPool(); diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerReferenceTaskAuthorizationTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerReferenceTaskAuthorizationTest.java index 23e924c2a..d65248f02 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerReferenceTaskAuthorizationTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerReferenceTaskAuthorizationTest.java @@ -46,8 +46,6 @@ class DefaultRequestHandlerReferenceTaskAuthorizationTest { .acceptedOutputModes(List.of()) .build(); - private static final PushNotificationSender NOOP_PUSH_SENDER = (event, snapshot) -> {}; - private InMemoryTaskStore taskStore; private InMemoryQueueManager queueManager; private MainEventBusProcessor mainEventBusProcessor; @@ -66,7 +64,7 @@ void setUp() { taskStore = new InMemoryTaskStore(); MainEventBus mainEventBus = new MainEventBus(); queueManager = new InMemoryQueueManager(taskStore, mainEventBus); - mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, NOOP_PUSH_SENDER, queueManager); + mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(mainEventBusProcessor); executor = Executors.newCachedThreadPool(); diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java index 95ffd0f7c..7e3d21bde 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java @@ -96,8 +96,6 @@ public class DefaultRequestHandlerTest { .parts(new TextPart("test message")) .build(); - private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = (event, snapshot) -> {}; - // Test infrastructure components protected AgentExecutor executor; protected TaskStore taskStore; @@ -139,7 +137,7 @@ public void cancel(RequestContext context, AgentEmitter agentEmitter) throws A2A // Create MainEventBus and MainEventBusProcessor mainEventBus = new MainEventBus(); queueManager = new InMemoryQueueManager(inMemoryTaskStore, mainEventBus); - mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, NOOP_PUSHNOTIFICATION_SENDER, queueManager); + mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(mainEventBusProcessor); // Create DefaultRequestHandler @@ -197,7 +195,7 @@ void testBuilderStartsManuallyConstructedMainEventBusProcessor() throws Exceptio MainEventBus manualMainEventBus = new MainEventBus(); InMemoryQueueManager manualQueueManager = new InMemoryQueueManager(manualTaskStore, manualMainEventBus); MainEventBusProcessor manualProcessor = new MainEventBusProcessor( - manualMainEventBus, manualTaskStore, NOOP_PUSHNOTIFICATION_SENDER, manualQueueManager); + manualMainEventBus, manualTaskStore, PushNotificationSender.NOOP, manualQueueManager); try { AgentExecutor manualAgentExecutor = new AgentExecutor() { @@ -1735,7 +1733,7 @@ public EventQueue.EventQueueBuilder createBaseEventQueueBuilder(String taskId) { } }; mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, - NOOP_PUSHNOTIFICATION_SENDER, queueManager); + PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(mainEventBusProcessor); CountDownLatch workingEventProcessing = new CountDownLatch(1); diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/AgentEmitterTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/AgentEmitterTest.java index 233d49bad..6c8a8ae10 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/AgentEmitterTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/AgentEmitterTest.java @@ -43,7 +43,6 @@ public class AgentEmitterTest { private static final List> SAMPLE_PARTS = List.of(new TextPart("Test message")); - private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = (event, snapshot) -> {}; public static final int WAIT_MILLI_SECONDS = 5000; EventQueue eventQueue; @@ -59,7 +58,7 @@ public void init() { InMemoryTaskStore taskStore = new InMemoryTaskStore(); mainEventBus = new MainEventBus(); InMemoryQueueManager queueManager = new InMemoryQueueManager(taskStore, mainEventBus); - mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, NOOP_PUSHNOTIFICATION_SENDER, queueManager); + mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(mainEventBusProcessor); eventQueue = EventQueueUtil.getEventQueueBuilder(mainEventBus) diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/PushNotificationSenderTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/PushNotificationSenderTest.java index c581927ad..80304ff2e 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/PushNotificationSenderTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/PushNotificationSenderTest.java @@ -13,10 +13,13 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.stream.Stream; +import jakarta.enterprise.inject.Instance; import org.a2aproject.sdk.client.http.A2AHttpClient; import org.a2aproject.sdk.client.http.A2AHttpResponse; import org.a2aproject.sdk.client.http.ServerSentEvent; @@ -36,11 +39,15 @@ import org.a2aproject.sdk.spec.TextPart; import org.jspecify.annotations.Nullable; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + public class PushNotificationSenderTest { private TestHttpClient testHttpClient; @@ -58,6 +65,7 @@ private static class TestHttpClient implements A2AHttpClient { final List rawBodies = Collections.synchronizedList(new ArrayList<>()); volatile CountDownLatch latch; volatile boolean shouldThrowException = false; + volatile int responseStatusCode = 200; @Override public GetBuilder createGet() { @@ -102,15 +110,16 @@ public A2AHttpResponse post() throws IOException, InterruptedException { urls.add(url); headers.add(new java.util.HashMap<>(requestHeaders)); + int status = responseStatusCode; return new A2AHttpResponse() { @Override public int status() { - return 200; + return status; } @Override public boolean success() { - return true; + return status >= 200 && status < 300; } @Override @@ -590,4 +599,150 @@ public void testSendNotificationRejectsCrlfInAuthFields(String description, Stri assertTrue(testHttpClient.rawBodies.isEmpty(), "No body should have been sent"); } + static Stream nonSuccessHttpStatuses() { + return Stream.of( + Arguments.of("3xx redirect", 301), + Arguments.of("4xx client error", 404), + Arguments.of("5xx server error", 500)); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("nonSuccessHttpStatuses") + public void testSendNotificationReturnsFalseOnNon2xxResponse(String description, int statusCode) { + String taskId = "task_non2xx_" + statusCode; + Task taskData = createSampleTask(taskId, TaskState.TASK_STATE_COMPLETED); + TaskPushNotificationConfig config = createSamplePushConfig(taskId, "http://notify.me/here", "cfg1", null); + configStore.setInfo(config); + + testHttpClient.responseStatusCode = statusCode; + + boolean result = sender.sendNotification(taskData, null); + + assertFalse(result, description + " (HTTP " + statusCode + ") must cause sendNotification to return false"); + } + + @Nested + class ExecutorResolution { + + @SuppressWarnings("unchecked") + private Instance mockInstance(boolean resolvable, Executor executor) { + Instance instance = mock(Instance.class); + when(instance.isResolvable()).thenReturn(resolvable); + if (resolvable) { + when(instance.get()).thenReturn(executor); + } + return instance; + } + + @SuppressWarnings("unchecked") + private BasePushNotificationSender createWithExecutors(Instance instrumented, + Instance named) { + Instance formatters = mock(Instance.class); + when(formatters.iterator()).thenReturn(Collections.emptyIterator()); + + Instance httpClient = mock(Instance.class); + when(httpClient.isResolvable()).thenReturn(false); + + Instance propagatingClient = mock(Instance.class); + when(propagatingClient.isResolvable()).thenReturn(false); + + return new BasePushNotificationSender( + new InMemoryPushNotificationConfigStore(), + formatters, + PushNotificationUrlValidator.ALLOW_ALL, + httpClient, + propagatingClient, + instrumented, + named); + } + + private TestHttpClient prepareDispatch(BasePushNotificationSender sender, String taskId) { + InMemoryPushNotificationConfigStore store = new InMemoryPushNotificationConfigStore(); + store.setInfo(createSamplePushConfig(taskId, "http://notify.me/here", "cfg1", null)); + TestHttpClient http = new TestHttpClient(); + http.latch = new CountDownLatch(1); + try { + java.lang.reflect.Field storeField = BasePushNotificationSender.class.getDeclaredField("configStore"); + storeField.setAccessible(true); + storeField.set(sender, store); + java.lang.reflect.Field clientField = BasePushNotificationSender.class.getDeclaredField("httpClient"); + clientField.setAccessible(true); + clientField.set(sender, http); + } catch (ReflectiveOperationException e) { + throw new RuntimeException(e); + } + return http; + } + + @Test + void instrumentedExecutorIsPreferredOverNamed() throws InterruptedException { + AtomicInteger instrumentedCount = new AtomicInteger(); + AtomicInteger namedCount = new AtomicInteger(); + + Executor instrumentedExec = command -> { + instrumentedCount.incrementAndGet(); + command.run(); + }; + Executor namedExec = command -> { + namedCount.incrementAndGet(); + command.run(); + }; + + BasePushNotificationSender s = createWithExecutors( + mockInstance(true, instrumentedExec), + mockInstance(true, namedExec)); + + String taskId = "exec-resolution-instrumented"; + TestHttpClient httpClient = prepareDispatch(s, taskId); + + Task task = createSampleTask(taskId, TaskState.TASK_STATE_COMPLETED); + s.sendNotification(task, null); + + assertTrue(httpClient.latch.await(5, TimeUnit.SECONDS)); + assertTrue(instrumentedCount.get() > 0, + "@Instrumented executor must be used when available"); + assertEquals(0, namedCount.get(), + "@Named executor must NOT be used when @Instrumented is available"); + } + + @Test + void namedExecutorUsedWhenInstrumentedAbsent() throws InterruptedException { + AtomicInteger namedCount = new AtomicInteger(); + + Executor namedExec = command -> { + namedCount.incrementAndGet(); + command.run(); + }; + + BasePushNotificationSender s = createWithExecutors( + mockInstance(false, null), + mockInstance(true, namedExec)); + + String taskId = "exec-resolution-named"; + TestHttpClient httpClient = prepareDispatch(s, taskId); + + Task task = createSampleTask(taskId, TaskState.TASK_STATE_COMPLETED); + s.sendNotification(task, null); + + assertTrue(httpClient.latch.await(5, TimeUnit.SECONDS)); + assertTrue(namedCount.get() > 0, + "@Named executor must be used when @Instrumented is absent"); + } + + @Test + void commonPoolUsedWhenBothAbsent() throws InterruptedException { + BasePushNotificationSender s = createWithExecutors( + mockInstance(false, null), + mockInstance(false, null)); + + String taskId = "exec-resolution-fallback"; + TestHttpClient httpClient = prepareDispatch(s, taskId); + + Task task = createSampleTask(taskId, TaskState.TASK_STATE_COMPLETED); + s.sendNotification(task, null); + + assertTrue(httpClient.latch.await(5, TimeUnit.SECONDS), + "Notification must still be dispatched via the common pool fallback"); + } + } } diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/ResultAggregatorTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/ResultAggregatorTest.java index 04c6d582a..b80372d75 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/ResultAggregatorTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/ResultAggregatorTest.java @@ -254,7 +254,7 @@ void testConsumeAndBreakNonBlocking() throws Exception { InMemoryTaskStore taskStore = new InMemoryTaskStore(); InMemoryQueueManager queueManager = new InMemoryQueueManager(new MockTaskStateProvider(), mainEventBus); - MainEventBusProcessor processor = new MainEventBusProcessor(mainEventBus, taskStore, (event, snapshot) -> {}, queueManager); + MainEventBusProcessor processor = new MainEventBusProcessor(mainEventBus, taskStore, PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(processor); EventQueue queue = queueManager.getEventQueueBuilder(taskId).build().tap(); @@ -372,7 +372,7 @@ void testConsumeAndBreakOnAuthRequired_Blocking() throws Exception { InMemoryTaskStore taskStore = new InMemoryTaskStore(); InMemoryQueueManager queueManager = new InMemoryQueueManager(new MockTaskStateProvider(), mainEventBus); - MainEventBusProcessor processor = new MainEventBusProcessor(mainEventBus, taskStore, (event, snapshot) -> {}, queueManager); + MainEventBusProcessor processor = new MainEventBusProcessor(mainEventBus, taskStore, PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(processor); EventQueue queue = queueManager.getEventQueueBuilder(taskId).build().tap(); @@ -414,7 +414,7 @@ void testConsumeAndBreakOnAuthRequired_NonBlocking() throws Exception { InMemoryTaskStore taskStore = new InMemoryTaskStore(); InMemoryQueueManager queueManager = new InMemoryQueueManager(new MockTaskStateProvider(), mainEventBus); - MainEventBusProcessor processor = new MainEventBusProcessor(mainEventBus, taskStore, (event, snapshot) -> {}, queueManager); + MainEventBusProcessor processor = new MainEventBusProcessor(mainEventBus, taskStore, PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(processor); EventQueue queue = queueManager.getEventQueueBuilder(taskId).build().tap(); @@ -461,7 +461,7 @@ void testAuthRequiredWithTaskStatusUpdateEvent() throws Exception { InMemoryTaskStore taskStore = new InMemoryTaskStore(); InMemoryQueueManager queueManager = new InMemoryQueueManager(new MockTaskStateProvider(), mainEventBus); - MainEventBusProcessor processor = new MainEventBusProcessor(mainEventBus, taskStore, (event, snapshot) -> {}, queueManager); + MainEventBusProcessor processor = new MainEventBusProcessor(mainEventBus, taskStore, PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(processor); EventQueue queue = queueManager.getEventQueueBuilder(taskId).build().tap(); @@ -507,7 +507,7 @@ void testAuthRequiredWithTaskEvent() throws Exception { InMemoryTaskStore taskStore = new InMemoryTaskStore(); InMemoryQueueManager queueManager = new InMemoryQueueManager(new MockTaskStateProvider(), mainEventBus); - MainEventBusProcessor processor = new MainEventBusProcessor(mainEventBus, taskStore, (event, snapshot) -> {}, queueManager); + MainEventBusProcessor processor = new MainEventBusProcessor(mainEventBus, taskStore, PushNotificationSender.NOOP, queueManager); EventQueueUtil.start(processor); EventQueue queue = queueManager.getEventQueueBuilder(taskId).build().tap();