From 56a4b4ba319d6a3fecbdd269599153a1187f9480 Mon Sep 17 00:00:00 2001 From: Kabir Khan Date: Tue, 24 Feb 2026 15:11:05 +0000 Subject: [PATCH 1/5] feat: improve TaskStore error reporting with domain-specific exceptions (#516) Replace generic RuntimeException with TaskSerializationException and TaskPersistenceException throughout the TaskStore persistence layer. Wire execptions into MainEventQueueBusProcessor. --- .gitignore | 1 + .../database/jpa/JpaDatabaseTaskStore.java | 181 ++++++-- .../server/events/MainEventBusProcessor.java | 39 ++ .../a2a/server/tasks/InMemoryTaskStore.java | 57 +++ .../tasks/TaskPersistenceException.java | 174 ++++++++ .../tasks/TaskSerializationException.java | 97 ++++ .../java/io/a2a/server/tasks/TaskStore.java | 129 ++++++ .../a2a/server/tasks/TaskStoreException.java | 113 +++++ .../MainEventBusProcessorExceptionTest.java | 312 +++++++++++++ .../tasks/AbstractTaskStoreExceptionTest.java | 168 +++++++ .../tasks/TaskPersistenceExceptionTest.java | 419 ++++++++++++++++++ .../tasks/TaskSerializationExceptionTest.java | 272 ++++++++++++ .../server/tasks/TaskStoreExceptionTest.java | 191 ++++++++ 13 files changed, 2112 insertions(+), 41 deletions(-) create mode 100644 server-common/src/main/java/io/a2a/server/tasks/TaskPersistenceException.java create mode 100644 server-common/src/main/java/io/a2a/server/tasks/TaskSerializationException.java create mode 100644 server-common/src/main/java/io/a2a/server/tasks/TaskStoreException.java create mode 100644 server-common/src/test/java/io/a2a/server/events/MainEventBusProcessorExceptionTest.java create mode 100644 server-common/src/test/java/io/a2a/server/tasks/AbstractTaskStoreExceptionTest.java create mode 100644 server-common/src/test/java/io/a2a/server/tasks/TaskPersistenceExceptionTest.java create mode 100644 server-common/src/test/java/io/a2a/server/tasks/TaskSerializationExceptionTest.java create mode 100644 server-common/src/test/java/io/a2a/server/tasks/TaskStoreExceptionTest.java diff --git a/.gitignore b/.gitignore index 3679c8f0c..441ef20e1 100644 --- a/.gitignore +++ b/.gitignore @@ -55,3 +55,4 @@ nbproject/ .serena/ .bob/ claudedocs +backlog/ diff --git a/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java b/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java index 086e640e3..6693ab733 100644 --- a/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java +++ b/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java @@ -13,6 +13,7 @@ import jakarta.inject.Inject; import jakarta.persistence.EntityManager; import jakarta.persistence.PersistenceContext; +import jakarta.persistence.PersistenceException; import jakarta.persistence.TypedQuery; import jakarta.transaction.Transactional; @@ -22,6 +23,8 @@ import io.a2a.server.config.A2AConfigProvider; import io.a2a.server.tasks.TaskStateProvider; import io.a2a.server.tasks.TaskStore; +import io.a2a.server.tasks.TaskSerializationException; +import io.a2a.server.tasks.TaskPersistenceException; import io.a2a.spec.Artifact; import io.a2a.spec.ListTasksParams; import io.a2a.spec.Message; @@ -63,6 +66,67 @@ void initConfig() { gracePeriodSeconds = Long.parseLong(configProvider.getValue(A2A_REPLICATION_GRACE_PERIOD_SECONDS)); } + /** + * Determines if a database exception represents a transient failure that may succeed on retry. + * + *

Transient Errors (return true):

+ * + * + *

Non-Transient Errors (return false):

+ * + * + * @param e the persistence exception to analyze + * @return true if retry may succeed, false if manual intervention required + */ + private static boolean isTransientDatabaseError(PersistenceException e) { + // Check exception class hierarchy for transient indicators + Throwable cause = e; + while (cause != null) { + String className = cause.getClass().getName(); + String message = cause.getMessage() != null ? cause.getMessage().toLowerCase() : ""; + + // Transient exception types + if (className.contains("Timeout") || + className.contains("LockTimeout") || + className.contains("QueryTimeout")) { + return true; + } + + // Transient message patterns + if (message.contains("timeout") || + message.contains("deadlock") || + message.contains("connection") && (message.contains("refused") || message.contains("closed")) || + message.contains("pool") && message.contains("exhausted")) { + return true; + } + + // Non-transient indicators + if (className.contains("ConstraintViolation") || + className.contains("IntegrityConstraint") || + message.contains("unique constraint") || + message.contains("foreign key") || + message.contains("disk full") || + message.contains("permission denied")) { + return false; + } + + cause = cause.getCause(); + } + + // Default to non-transient for safety (don't retry unless confident) + return false; + } + @Transactional @Override public void save(Task task, boolean isReplicated) { @@ -85,7 +149,13 @@ public void save(Task task, boolean isReplicated) { } } catch (JsonProcessingException e) { LOGGER.error("Failed to serialize task with ID: {}", task.id(), e); - throw new RuntimeException("Failed to serialize task with ID: " + task.id(), e); + throw new TaskSerializationException(task.id(), + "Failed to serialize task for persistence", e); + } catch (PersistenceException e) { + boolean isTransient = isTransientDatabaseError(e); + LOGGER.error("Database save failed for task with ID: {} (transient: {})", task.id(), isTransient, e); + throw new TaskPersistenceException(task.id(), + "Database save failed for task", e, isTransient); } } @@ -93,19 +163,28 @@ public void save(Task task, boolean isReplicated) { @Override public Task get(String taskId) { LOGGER.debug("Retrieving task with ID: {}", taskId); - JpaTask jpaTask = em.find(JpaTask.class, taskId); - if (jpaTask == null) { - LOGGER.debug("Task not found with ID: {}", taskId); - return null; - } - try { - Task task = jpaTask.getTask(); - LOGGER.debug("Successfully retrieved task with ID: {}", taskId); - return task; - } catch (JsonProcessingException e) { - LOGGER.error("Failed to deserialize task with ID: {}", taskId, e); - throw new RuntimeException("Failed to deserialize task with ID: " + taskId, e); + JpaTask jpaTask = em.find(JpaTask.class, taskId); + if (jpaTask == null) { + LOGGER.debug("Task not found with ID: {}", taskId); + return null; + } + + try { + Task task = jpaTask.getTask(); + LOGGER.debug("Successfully retrieved task with ID: {}", taskId); + return task; + } catch (JsonProcessingException e) { + LOGGER.error("Failed to deserialize task with ID: {}", taskId, e); + throw new TaskSerializationException(taskId, + "Failed to deserialize task from database", e); + } + + } catch (PersistenceException e) { + boolean isTransient = isTransientDatabaseError(e); + LOGGER.error("Database retrieval failed for task with ID: {} (transient: {})", taskId, isTransient, e); + throw new TaskPersistenceException(taskId, + "Database retrieval failed for task", e, isTransient); } } @@ -113,12 +192,19 @@ public Task get(String taskId) { @Override public void delete(String taskId) { LOGGER.debug("Deleting task with ID: {}", taskId); - JpaTask jpaTask = em.find(JpaTask.class, taskId); - if (jpaTask != null) { - em.remove(jpaTask); - LOGGER.debug("Successfully deleted task with ID: {}", taskId); - } else { - LOGGER.debug("Task not found for deletion with ID: {}", taskId); + try { + JpaTask jpaTask = em.find(JpaTask.class, taskId); + if (jpaTask != null) { + em.remove(jpaTask); + LOGGER.debug("Successfully deleted task with ID: {}", taskId); + } else { + LOGGER.debug("Task not found for deletion with ID: {}", taskId); + } + } catch (PersistenceException e) { + boolean isTransient = isTransientDatabaseError(e); + LOGGER.error("Database deletion failed for task with ID: {} (transient: {})", taskId, isTransient, e); + throw new TaskPersistenceException(taskId, + "Database deletion failed for task", e, isTransient); } } @@ -231,14 +317,15 @@ public ListTasksResult list(ListTasksParams params) { LOGGER.debug("Listing tasks with params: contextId={}, status={}, pageSize={}, pageToken={}", params.contextId(), params.status(), params.pageSize(), params.pageToken()); - // Parse pageToken once at the beginning - PageToken pageToken = PageToken.fromString(params.pageToken()); - Instant tokenTimestamp = pageToken != null ? pageToken.timestamp() : null; - String tokenId = pageToken != null ? pageToken.id() : null; + try { + // Parse pageToken once at the beginning + PageToken pageToken = PageToken.fromString(params.pageToken()); + Instant tokenTimestamp = pageToken != null ? pageToken.timestamp() : null; + String tokenId = pageToken != null ? pageToken.id() : null; - // Build dynamic JPQL query with WHERE clauses for filtering - StringBuilder queryBuilder = new StringBuilder("SELECT t FROM JpaTask t WHERE 1=1"); - StringBuilder countQueryBuilder = new StringBuilder("SELECT COUNT(t) FROM JpaTask t WHERE 1=1"); + // Build dynamic JPQL query with WHERE clauses for filtering + StringBuilder queryBuilder = new StringBuilder("SELECT t FROM JpaTask t WHERE 1=1"); + StringBuilder countQueryBuilder = new StringBuilder("SELECT COUNT(t) FROM JpaTask t WHERE 1=1"); // Apply contextId filter using denormalized column if (params.contextId() != null) { @@ -311,16 +398,17 @@ public ListTasksResult list(ListTasksParams params) { } int totalSize = countQuery.getSingleResult().intValue(); - // Deserialize tasks from JSON - List tasks = new ArrayList<>(); - for (JpaTask jpaTask : jpaTasksPage) { - try { - tasks.add(jpaTask.getTask()); - } catch (JsonProcessingException e) { - LOGGER.error("Failed to deserialize task with ID: {}", jpaTask.getId(), e); - throw new RuntimeException("Failed to deserialize task with ID: " + jpaTask.getId(), e); + // Deserialize tasks from JSON + List tasks = new ArrayList<>(); + for (JpaTask jpaTask : jpaTasksPage) { + try { + tasks.add(jpaTask.getTask()); + } catch (JsonProcessingException e) { + LOGGER.error("Failed to deserialize task with ID: {}", jpaTask.getId(), e); + throw new TaskSerializationException(jpaTask.getId(), + "Failed to deserialize task during list operation", e); + } } - } // Determine next page token (timestamp:ID of last task if there are more results) // Format: "timestamp_millis:taskId" for keyset pagination @@ -336,12 +424,23 @@ public ListTasksResult list(ListTasksParams params) { int historyLength = params.getEffectiveHistoryLength(); boolean includeArtifacts = params.shouldIncludeArtifacts(); - List transformedTasks = tasks.stream() - .map(task -> transformTask(task, historyLength, includeArtifacts)) - .toList(); - - LOGGER.debug("Returning {} tasks out of {} total", transformedTasks.size(), totalSize); - return new ListTasksResult(transformedTasks, totalSize, transformedTasks.size(), nextPageToken); + List transformedTasks = tasks.stream() + .map(task -> transformTask(task, historyLength, includeArtifacts)) + .toList(); + + LOGGER.debug("Returning {} tasks out of {} total", transformedTasks.size(), totalSize); + return new ListTasksResult(transformedTasks, totalSize, transformedTasks.size(), nextPageToken); + + } catch (TaskSerializationException e) { + // Re-throw TaskSerializationException from inner loop + throw e; + } catch (PersistenceException e) { + // Database errors from query creation, execution, or count + boolean isTransient = isTransientDatabaseError(e); + LOGGER.error("Database query failed during list operation (transient: {})", isTransient, e); + throw new TaskPersistenceException(null, // No single taskId for list operation + "Database query failed during list operation", e, isTransient); + } } private Task transformTask(Task task, int historyLength, boolean includeArtifacts) { diff --git a/server-common/src/main/java/io/a2a/server/events/MainEventBusProcessor.java b/server-common/src/main/java/io/a2a/server/events/MainEventBusProcessor.java index 575078679..c8f7db106 100644 --- a/server-common/src/main/java/io/a2a/server/events/MainEventBusProcessor.java +++ b/server-common/src/main/java/io/a2a/server/events/MainEventBusProcessor.java @@ -10,6 +10,8 @@ import io.a2a.server.tasks.PushNotificationSender; import io.a2a.server.tasks.TaskManager; +import io.a2a.server.tasks.TaskPersistenceException; +import io.a2a.server.tasks.TaskSerializationException; import io.a2a.server.tasks.TaskStore; import io.a2a.spec.A2AError; import io.a2a.spec.A2AServerException; @@ -41,6 +43,19 @@ * Note: This bean is eagerly initialized by {@link MainEventBusProcessorInitializer} * to ensure the background thread starts automatically when the application starts. *

+ * + *

Exception Handling

+ * TaskStore persistence failures are caught and handled gracefully: + *
    + *
  • {@link TaskSerializationException} - Data corruption or schema mismatch (permanent). + * Logged at ERROR level, distributed as {@link InternalError} to clients.
    • + *
  • {@link TaskPersistenceException} - Database failure (transient or permanent). + * Transient failures logged at WARN level, permanent at ERROR level. + * InternalError message indicates retry viability.
    • + *
+ * + *

Processing continues after errors - the failed event is distributed as InternalError + * to all ChildQueues, and the MainEventBusProcessor continues consuming subsequent events.

*/ @ApplicationScoped public class MainEventBusProcessor implements Runnable { @@ -293,11 +308,35 @@ private boolean updateTaskStore(String taskId, Event event, boolean isReplicated LOGGER.debug("TaskStore updated via TaskManager.process() for task {}: {} (final: {}, replicated: {})", taskId, event.getClass().getSimpleName(), isFinal, isReplicated); return isFinal; + + } catch (TaskSerializationException e) { + // Data corruption or schema mismatch - ALWAYS permanent + LOGGER.error("Task {} event serialization failed - data corruption detected: {}", + taskId, e.getMessage(), e); + throw new InternalError("Failed to serialize task " + taskId + ": " + e.getMessage()); + + } catch (TaskPersistenceException e) { + // Database failure - check if transient for retry guidance + if (e.isTransient()) { + LOGGER.warn("Task {} event persistence failed (TRANSIENT) - retry may succeed: {}", + taskId, e.getMessage()); + throw new InternalError("Temporary storage failure for task " + taskId + + " (retry may succeed): " + e.getMessage()); + } else { + LOGGER.error("Task {} event persistence failed (PERMANENT) - manual intervention required: {}", + taskId, e.getMessage(), e); + throw new InternalError("Permanent storage failure for task " + taskId + + " (requires intervention): " + e.getMessage()); + } + } catch (InternalError e) { + // Already an InternalError from TaskManager validation - pass through LOGGER.error("Error updating TaskStore via TaskManager for task {}", taskId, e); // Rethrow to prevent distributing unpersisted event to clients throw e; + } catch (Exception e) { + // Unexpected exception type - treat as permanent failure LOGGER.error("Unexpected error updating TaskStore for task {}", taskId, e); // Rethrow to prevent distributing unpersisted event to clients throw new InternalError("TaskStore persistence failed: " + e.getMessage()); diff --git a/server-common/src/main/java/io/a2a/server/tasks/InMemoryTaskStore.java b/server-common/src/main/java/io/a2a/server/tasks/InMemoryTaskStore.java index 15f94d7e8..de8774eb8 100644 --- a/server-common/src/main/java/io/a2a/server/tasks/InMemoryTaskStore.java +++ b/server-common/src/main/java/io/a2a/server/tasks/InMemoryTaskStore.java @@ -25,6 +25,63 @@ *

* This is the default TaskStore used when no other implementation is provided. *

+ * + *

Exception Behavior

+ * InMemoryTaskStore has minimal exception scenarios compared to database-backed implementations: + *
    + *
  • No TaskSerializationException: Task objects are stored directly in memory without + * serialization. No JSON parsing or schema compatibility issues can occur.
    • + *
  • No TaskPersistenceException: ConcurrentHashMap operations do not involve I/O, + * network, or transactional concerns. Standard put/get/remove operations are guaranteed + * to succeed under normal JVM operation.
    • + *
  • OutOfMemoryError (potential): The only failure scenario is JVM heap exhaustion if + * too many tasks are stored. This is an {@link Error} (not Exception) and indicates a fatal + * system condition requiring JVM restart and capacity planning.
    • + *
+ * + *

Design Rationale

+ * This implementation intentionally does NOT throw {@link TaskStoreException} or its subclasses + * because: + *
    + *
  • No serialization step exists - tasks stored as Java objects
    • + *
  • No I/O or network operations that can fail
    • + *
  • ConcurrentHashMap guarantees thread-safe operations without checked exceptions
    • + *
  • Memory exhaustion (OutOfMemoryError) is an unrecoverable system failure, not a transient + * storage error
    • + *
+ * + *

Comparison to Database Implementations

+ * Database-backed implementations (e.g., JpaDatabaseTaskStore) throw exceptions for: + *
    + *
  • Serialization errors (JSON parsing, schema mismatches)
    • + *
  • Connection failures (network, timeouts)
    • + *
  • Transaction failures (deadlocks, constraint violations)
    • + *
  • Capacity issues (disk full, quota exceeded)
    • + *
+ * InMemoryTaskStore avoids all of these by operating entirely in-process. + * + *

Memory Management Considerations

+ * Callers should monitor memory usage and implement task cleanup policies: + * 
{@code
+ * // Example: Delete finalized tasks older than 48 hours
+ * ListTasksParams params = new ListTasksParams.Builder()
+ *     .statusTimestampBefore(Instant.now().minus(Duration.ofHours(48)))
+ *     .build();
+ *
+ * List oldTasks = taskStore.list(params).tasks();
+ * oldTasks.stream()
+ *     .filter(task -> task.status().state().isFinal())
+ *     .forEach(task -> taskStore.delete(task.id()));
+ * }
+ * + *

Thread Safety

+ * All operations are thread-safe via {@link ConcurrentHashMap}. Multiple threads can + * concurrently save, get, list, and delete tasks without synchronization. Last-write-wins + * semantics apply for concurrent {@code save()} calls to the same task ID. + * + * @see TaskStore for interface contract and exception documentation + * @see TaskStoreException for exception hierarchy (not thrown by this implementation) + * @see TaskStateProvider for queue lifecycle integration */ @ApplicationScoped public class InMemoryTaskStore implements TaskStore, TaskStateProvider { diff --git a/server-common/src/main/java/io/a2a/server/tasks/TaskPersistenceException.java b/server-common/src/main/java/io/a2a/server/tasks/TaskPersistenceException.java new file mode 100644 index 000000000..0b40b4310 --- /dev/null +++ b/server-common/src/main/java/io/a2a/server/tasks/TaskPersistenceException.java @@ -0,0 +1,174 @@ +package io.a2a.server.tasks; + +import org.jspecify.annotations.Nullable; + +/** + * Exception for database/storage system failures during task persistence operations. + *

+ * Indicates failures in the underlying storage system (database, filesystem, etc.) rather + * than data format issues. Includes a {@link #isTransient()} flag to distinguish between + * temporary failures (retry recommended) and permanent failures (manual intervention required). + * + *

Transient Failures (isTransient = true)

+ * Temporary issues that may resolve with retry: + *
    + *
  • Database connection timeout or network partition
    • + *
  • Transaction deadlock or lock wait timeout
    • + *
  • Connection pool exhausted
    • + *
  • Temporary storage system overload
    • + *
+ * Recovery: Exponential backoff retry, circuit breaker patterns + * + *

Non-Transient Failures (isTransient = false)

+ * Persistent issues requiring intervention: + *
    + *
  • Disk full / storage quota exceeded
    • + *
  • Database constraint violations (unique key, foreign key)
    • + *
  • Insufficient permissions or authentication failures
    • + *
  • Database schema incompatibilities
    • + *
+ * Recovery: Manual intervention, capacity planning, configuration fixes + * + *

Usage Example

+ * 
{@code
+ * try {
+ *     em.merge(jpaTask);
+ * } catch (PersistenceException e) {
+ *     boolean transient = isTransientDatabaseError(e);
+ *     throw new TaskPersistenceException(taskId, "Database save failed", e, transient);
+ * }
+ * }
+ * + * @see TaskStoreException + * @see TaskSerializationException for data format errors + */ +public class TaskPersistenceException extends TaskStoreException { + + private final boolean isTransientFailure; + + /** + * Creates a new TaskPersistenceException with no message or cause. + * The failure is assumed to be non-transient. + */ + public TaskPersistenceException() { + super(); + this.isTransientFailure = false; + } + + /** + * Creates a new TaskPersistenceException with the specified message. + * The failure is assumed to be non-transient. + * + * @param msg the exception message + */ + public TaskPersistenceException(final String msg) { + super(msg); + this.isTransientFailure = false; + } + + /** + * Creates a new TaskPersistenceException with the specified cause. + * The failure is assumed to be non-transient. + * + * @param cause the underlying cause + */ + public TaskPersistenceException(final Throwable cause) { + super(cause); + this.isTransientFailure = false; + } + + /** + * Creates a new TaskPersistenceException with the specified message and cause. + * The failure is assumed to be non-transient. + * + * @param msg the exception message + * @param cause the underlying cause + */ + public TaskPersistenceException(final String msg, final Throwable cause) { + super(msg, cause); + this.isTransientFailure = false; + } + + /** + * Creates a new TaskPersistenceException with the specified task ID and message. + * The failure is assumed to be non-transient. + * + * @param taskId the task identifier (may be null for operations not tied to a specific task) + * @param msg the exception message + */ + public TaskPersistenceException(@Nullable final String taskId, final String msg) { + super(taskId, msg); + this.isTransientFailure = false; + } + + /** + * Creates a new TaskPersistenceException with the specified task ID, message, and cause. + * The failure is assumed to be non-transient. + * + * @param taskId the task identifier (may be null for operations not tied to a specific task) + * @param msg the exception message + * @param cause the underlying cause + */ + public TaskPersistenceException(@Nullable final String taskId, final String msg, final Throwable cause) { + super(taskId, msg, cause); + this.isTransientFailure = false; + } + + /** + * Creates a new TaskPersistenceException with the specified message and transient flag. + * + * @param msg the exception message + * @param isTransient true if the failure is transient and may resolve with retry, false otherwise + */ + public TaskPersistenceException(final String msg, final boolean isTransient) { + super(msg); + this.isTransientFailure = isTransient; + } + + /** + * Creates a new TaskPersistenceException with the specified message, cause, and transient flag. + * + * @param msg the exception message + * @param cause the underlying cause + * @param isTransient true if the failure is transient and may resolve with retry, false otherwise + */ + public TaskPersistenceException(final String msg, final Throwable cause, final boolean isTransient) { + super(msg, cause); + this.isTransientFailure = isTransient; + } + + /** + * Creates a new TaskPersistenceException with the specified task ID, message, and transient flag. + * + * @param taskId the task identifier (may be null for operations not tied to a specific task) + * @param msg the exception message + * @param isTransient true if the failure is transient and may resolve with retry, false otherwise + */ + public TaskPersistenceException(@Nullable final String taskId, final String msg, final boolean isTransient) { + super(taskId, msg); + this.isTransientFailure = isTransient; + } + + /** + * Creates a new TaskPersistenceException with the specified task ID, message, cause, and transient flag. + * + * @param taskId the task identifier (may be null for operations not tied to a specific task) + * @param msg the exception message + * @param cause the underlying cause + * @param isTransient true if the failure is transient and may resolve with retry, false otherwise + */ + public TaskPersistenceException(@Nullable final String taskId, final String msg, final Throwable cause, + final boolean isTransient) { + super(taskId, msg, cause); + this.isTransientFailure = isTransient; + } + + /** + * Indicates whether this failure is transient (retry may help). + * + * @return true if transient (network, timeout, deadlock), false if permanent (disk full, constraint) + */ + public boolean isTransient() { + return isTransientFailure; + } +} diff --git a/server-common/src/main/java/io/a2a/server/tasks/TaskSerializationException.java b/server-common/src/main/java/io/a2a/server/tasks/TaskSerializationException.java new file mode 100644 index 000000000..ac960f91f --- /dev/null +++ b/server-common/src/main/java/io/a2a/server/tasks/TaskSerializationException.java @@ -0,0 +1,97 @@ +package io.a2a.server.tasks; + +import org.jspecify.annotations.Nullable; + +/** + * Exception for task serialization/deserialization failures. + *

+ * Indicates failures converting between Task domain objects and persistent storage format (JSON). + * These failures are typically non-transient - they indicate data corruption, schema + * mismatches, or invalid field values that require manual intervention. + * + *

Common Scenarios

+ *
    + *
  • JSON parsing errors during {@code get()} operations
    • + *
  • JSON serialization errors during {@code save()} operations
    • + *
  • Invalid enum values or missing required fields
    • + *
  • Data format version mismatches after upgrades
    • + *
+ * + *

Recovery Strategy

+ * Non-Transient: Retry will not help. Requires: + *
    + *
  • Data migration or repair for corrupted tasks
    • + *
  • Schema updates for version mismatches
    • + *
  • Manual intervention to fix invalid data
    • + *
+ * + *

Usage Example

+ * 
{@code
+ * try {
+ *     Task task = jsonMapper.readValue(json, Task.class);
+ * } catch (JsonProcessingException e) {
+ *     throw new TaskSerializationException(taskId, "Failed to deserialize task", e);
+ * }
+ * }
+ * + * @see TaskStoreException + * @see TaskPersistenceException for database failures + */ +public class TaskSerializationException extends TaskStoreException { + + /** + * Creates a new TaskSerializationException with no message or cause. + */ + public TaskSerializationException() { + super(); + } + + /** + * Creates a new TaskSerializationException with the specified message. + * + * @param msg the exception message + */ + public TaskSerializationException(final String msg) { + super(msg); + } + + /** + * Creates a new TaskSerializationException with the specified cause. + * + * @param cause the underlying cause + */ + public TaskSerializationException(final Throwable cause) { + super(cause); + } + + /** + * Creates a new TaskSerializationException with the specified message and cause. + * + * @param msg the exception message + * @param cause the underlying cause + */ + public TaskSerializationException(final String msg, final Throwable cause) { + super(msg, cause); + } + + /** + * Creates a new TaskSerializationException with the specified task ID and message. + * + * @param taskId the task identifier (may be null for operations not tied to a specific task) + * @param msg the exception message + */ + public TaskSerializationException(@Nullable final String taskId, final String msg) { + super(taskId, msg); + } + + /** + * Creates a new TaskSerializationException with the specified task ID, message, and cause. + * + * @param taskId the task identifier (may be null for operations not tied to a specific task) + * @param msg the exception message + * @param cause the underlying cause + */ + public TaskSerializationException(@Nullable final String taskId, final String msg, final Throwable cause) { + super(taskId, msg, cause); + } +} diff --git a/server-common/src/main/java/io/a2a/server/tasks/TaskStore.java b/server-common/src/main/java/io/a2a/server/tasks/TaskStore.java index 3df903f77..c0b7eb9ea 100644 --- a/server-common/src/main/java/io/a2a/server/tasks/TaskStore.java +++ b/server-common/src/main/java/io/a2a/server/tasks/TaskStore.java @@ -90,10 +90,119 @@ *
  • Consider caching for frequently-accessed task lists
    • * * + *

    Exception Contract

    + * All TaskStore methods may throw {@link TaskStoreException} or its subclasses to indicate + * persistence failures. Implementers must choose the appropriate exception type based on + * the failure cause: + *
      + *
    • {@link TaskSerializationException} - JSON/data format errors (non-transient)
      • + *
    • {@link TaskPersistenceException} - Database/storage failures (transient or non-transient)
      • + *
    + * + *

    When to Throw TaskSerializationException

    + * Use when task data cannot be serialized or deserialized: + *
      + *
    • JSON parsing errors during {@code get()} operations
      • + *
    • JSON serialization errors during {@code save()} operations
      • + *
    • Invalid enum values or missing required fields
      • + *
    • Schema version mismatches after upgrades
      • + *
    + * These failures are always non-transient - retry will not help. They require data + * migration, schema updates, or manual intervention. + * + *

    When to Throw TaskPersistenceException

    + * Use when the storage system fails: + *
      + *
    • Database connection timeouts (transient)
      • + *
    • Transaction deadlocks (transient)
      • + *
    • Connection pool exhausted (transient)
      • + *
    • Disk full / quota exceeded (non-transient)
      • + *
    • Database constraint violations (non-transient)
      • + *
    • Insufficient permissions (non-transient)
      • + *
    + * Set the {@code isTransient} flag appropriately: + *
      + *
    • {@code true} - Temporary failure, retry may succeed (network, timeout, deadlock)
      • + *
    • {@code false} - Permanent failure, requires intervention (disk full, constraints)
      • + *
    + * + *

    Implementer Example

    + * 
    {@code
+ * @Override
+ * public void save(Task task, boolean isReplicated) {
+ *     try {
+ *         String json = objectMapper.writeValueAsString(task);
+ *     } catch (JsonProcessingException e) {
+ *         throw new TaskSerializationException(task.id(), "Failed to serialize task", e);
+ *     }
+ *
+ *     try {
+ *         entityManager.merge(toEntity(json));
+ *     } catch (PersistenceException e) {
+ *         boolean transient = isTransientDatabaseError(e);
+ *         throw new TaskPersistenceException(task.id(), "Database save failed", e, transient);
+ *     }
+ * }
+ *
+ * @Override
+ * public Task get(String taskId) {
+ *     String json = database.retrieve(taskId);
+ *     try {
+ *         return objectMapper.readValue(json, Task.class);
+ *     } catch (JsonProcessingException e) {
+ *         throw new TaskSerializationException(taskId, "Failed to deserialize task", e);
+ *     }
+ * }
+ * }
    + * + *

    Caller Exception Handling

    + * Callers should distinguish between transient and permanent failures: + * 
    {@code
+ * try {
+ *     taskStore.save(task, false);
+ * } catch (TaskSerializationException e) {
+ *     // Non-transient: Log error, notify operations team
+ *     logger.error("Task data corruption for {}: {}", e.getTaskId(), e.getMessage(), e);
+ *     alerting.sendAlert("Task serialization failure", e);
+ *     // DO NOT RETRY - requires manual data repair
+ *
+ * } catch (TaskPersistenceException e) {
+ *     if (e.isTransient()) {
+ *         // Transient: Retry with exponential backoff
+ *         logger.warn("Transient persistence failure for {}: {}", e.getTaskId(), e.getMessage());
+ *         retryWithBackoff(() -> taskStore.save(task, false));
+ *     } else {
+ *         // Non-transient: Log error, alert operations
+ *         logger.error("Permanent persistence failure for {}: {}", e.getTaskId(), e.getMessage(), e);
+ *         alerting.sendAlert("Database capacity/constraint issue", e);
+ *         // DO NOT RETRY - requires manual intervention
+ *     }
+ * } catch (TaskStoreException e) {
+ *     // Generic fallback - treat as non-transient
+ *     logger.error("TaskStore failure for {}: {}", e.getTaskId(), e.getMessage(), e);
+ *     // DO NOT RETRY by default
+ * }
+ * }
    + * + *

    Current Exception Handling

    + * {@link io.a2a.server.events.MainEventBusProcessor} currently catches all TaskStore + * exceptions and wraps them in {@link io.a2a.spec.InternalError} events for client + * distribution. Future enhancements may distinguish transient failures for retry logic. + * + *

    Method-Specific Notes

    + *
      + *
    • {@code delete()} typically does not throw TaskSerializationException (no deserialization required)
      • + *
    • {@code list()} may encounter serialization errors for any task in the result set
      • + *
    + * * @see TaskManager * @see TaskStateProvider + * @see TaskStoreException + * @see TaskSerializationException + * @see TaskPersistenceException * @see InMemoryTaskStore * @see io.a2a.server.requesthandlers.DefaultRequestHandler + * @see io.a2a.server.events.MainEventBusProcessor */ public interface TaskStore { /** @@ -103,6 +212,11 @@ public interface TaskStore { * @param isReplicated true if this task update came from a replicated event, * false if it originated locally. Used to prevent feedback loops * in replicated scenarios (e.g., don't fire TaskFinalizedEvent for replicated updates) + * @throws TaskSerializationException if the task cannot be serialized to storage format (JSON parsing error, + * invalid field values, schema mismatch). Non-transient - retry will not help. + * @throws TaskPersistenceException if the storage system fails (database timeout, connection error, disk full). + * Check {@link TaskPersistenceException#isTransient()} to determine if retry is appropriate. + * @throws TaskStoreException for other persistence failures not covered by specific subclasses */ void save(Task task, boolean isReplicated); @@ -111,6 +225,11 @@ public interface TaskStore { * * @param taskId the task identifier * @return the task if found, null otherwise + * @throws TaskSerializationException if the persisted task data cannot be deserialized (corrupted JSON, + * schema incompatibility). Non-transient - indicates data corruption. + * @throws TaskPersistenceException if the storage system fails during retrieval (database connection error, + * query timeout). Check {@link TaskPersistenceException#isTransient()} for retry guidance. + * @throws TaskStoreException for other retrieval failures not covered by specific subclasses */ @Nullable Task get(String taskId); @@ -118,6 +237,10 @@ public interface TaskStore { * Deletes a task by its ID. * * @param taskId the task identifier + * @throws TaskPersistenceException if the storage system fails during deletion (database connection error, + * transaction timeout, constraint violation). Check {@link TaskPersistenceException#isTransient()} + * to determine if retry is appropriate. + * @throws TaskStoreException for other deletion failures not covered by specific subclasses */ void delete(String taskId); @@ -126,6 +249,12 @@ public interface TaskStore { * * @param params the filtering and pagination parameters * @return the list of tasks matching the criteria with pagination info + * @throws TaskSerializationException if any persisted task data cannot be deserialized during listing + * (corrupted JSON in database). Non-transient - indicates data corruption affecting + * one or more tasks. + * @throws TaskPersistenceException if the storage system fails during the list operation (database query timeout, + * connection error). Check {@link TaskPersistenceException#isTransient()} for retry guidance. + * @throws TaskStoreException for other listing failures not covered by specific subclasses */ ListTasksResult list(ListTasksParams params); } diff --git a/server-common/src/main/java/io/a2a/server/tasks/TaskStoreException.java b/server-common/src/main/java/io/a2a/server/tasks/TaskStoreException.java new file mode 100644 index 000000000..5865c5a80 --- /dev/null +++ b/server-common/src/main/java/io/a2a/server/tasks/TaskStoreException.java @@ -0,0 +1,113 @@ +package io.a2a.server.tasks; + +import io.a2a.spec.A2AServerException; +import org.jspecify.annotations.Nullable; + +/** + * Base exception for TaskStore persistence layer failures. + *

    + * Root exception for all task storage and retrieval errors. Specialized subclasses + * provide specific failure contexts: + *

      + *
    • {@link TaskSerializationException} - JSON/data format errors
      • + *
    • {@link TaskPersistenceException} - Database/storage system failures
      • + *
    + * + *

    Usage Context

    + * Thrown by {@link TaskStore} implementations during: + *
      + *
    • {@code save(Task, boolean)} - Task persistence failures
      • + *
    • {@code get(String)} - Task retrieval failures
      • + *
    • {@code delete(String)} - Task deletion failures
      • + *
    • {@code list(ListTasksParams)} - Task listing failures
      • + *
    + * + *

    Error Handling Pattern

    + * Caught by {@link io.a2a.server.events.MainEventBusProcessor} which: + *
      + *
    1. Logs the failure with full context (taskId, operation)
      2. + *
    2. Distributes {@link io.a2a.spec.InternalError} event to clients
      3. + *
    3. Preserves exception cause chain for diagnostics
      4. + *
    + * + * @see TaskSerializationException for data format errors + * @see TaskPersistenceException for storage system failures + * @see TaskStore + */ +public class TaskStoreException extends A2AServerException { + + @Nullable + private final String taskId; + + /** + * Creates a new TaskStoreException with no message or cause. + */ + public TaskStoreException() { + super(); + this.taskId = null; + } + + /** + * Creates a new TaskStoreException with the specified message. + * + * @param msg the exception message + */ + public TaskStoreException(final String msg) { + super(msg); + this.taskId = null; + } + + /** + * Creates a new TaskStoreException with the specified cause. + * + * @param cause the underlying cause + */ + public TaskStoreException(final Throwable cause) { + super(cause); + this.taskId = null; + } + + /** + * Creates a new TaskStoreException with the specified message and cause. + * + * @param msg the exception message + * @param cause the underlying cause + */ + public TaskStoreException(final String msg, final Throwable cause) { + super(msg, cause); + this.taskId = null; + } + + /** + * Creates a new TaskStoreException with the specified task ID and message. + * + * @param taskId the task identifier (may be null for operations not tied to a specific task) + * @param msg the exception message + */ + public TaskStoreException(@Nullable final String taskId, final String msg) { + super(msg); + this.taskId = taskId; + } + + /** + * Creates a new TaskStoreException with the specified task ID, message, and cause. + * + * @param taskId the task identifier (may be null for operations not tied to a specific task) + * @param msg the exception message + * @param cause the underlying cause + */ + public TaskStoreException(@Nullable final String taskId, final String msg, final Throwable cause) { + super(msg, cause); + this.taskId = taskId; + } + + /** + * Returns the task ID associated with this exception. + * + * @return the task ID, or null if not associated with a specific task + */ + @Nullable + public String getTaskId() { + return taskId; + } +} diff --git a/server-common/src/test/java/io/a2a/server/events/MainEventBusProcessorExceptionTest.java b/server-common/src/test/java/io/a2a/server/events/MainEventBusProcessorExceptionTest.java new file mode 100644 index 000000000..eb6c53d46 --- /dev/null +++ b/server-common/src/test/java/io/a2a/server/events/MainEventBusProcessorExceptionTest.java @@ -0,0 +1,312 @@ +package io.a2a.server.events; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import io.a2a.server.tasks.PushNotificationSender; +import io.a2a.server.tasks.TaskManager; +import io.a2a.server.tasks.TaskPersistenceException; +import io.a2a.server.tasks.TaskSerializationException; +import io.a2a.server.tasks.TaskStore; +import io.a2a.spec.Event; +import io.a2a.spec.InternalError; +import io.a2a.spec.Task; +import io.a2a.spec.TaskState; +import io.a2a.spec.TaskStatus; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; + +/** + * Integration tests for MainEventBusProcessor exception handling. + *

    + * Tests verify that TaskStore persistence failures are converted to InternalError events + * and distributed to clients with appropriate logging based on failure type: + *

      + *
    • TaskSerializationException → ERROR log + InternalError
      • + *
    • TaskPersistenceException (transient) → WARN log + InternalError
      • + *
    • TaskPersistenceException (permanent) → ERROR log + InternalError
      • + *
    + */ +public class MainEventBusProcessorExceptionTest { + + private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = task -> {}; + private static final String TASK_ID = "test-task-123"; + + private MainEventBus mainEventBus; + private MainEventBusProcessor mainEventBusProcessor; + private TaskStore mockTaskStore; + private InMemoryQueueManager queueManager; + private EventQueue eventQueue; + private ListAppender logAppender; + + @BeforeEach + public void setUp() { + // Set up mock TaskStore + mockTaskStore = mock(TaskStore.class); + + // Set up MainEventBus and processor with mock TaskStore + mainEventBus = new MainEventBus(); + queueManager = new InMemoryQueueManager(null, mainEventBus); // null TaskStateProvider for tests + mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, mockTaskStore, + NOOP_PUSHNOTIFICATION_SENDER, queueManager); + + // Set up log capture for verifying error messages + Logger logger = (Logger) LoggerFactory.getLogger(MainEventBusProcessor.class); + logAppender = new ListAppender<>(); + logAppender.start(); + logger.addAppender(logAppender); + + // Start processor + EventQueueUtil.start(mainEventBusProcessor); + + // Create event queue for testing + eventQueue = EventQueueUtil.getEventQueueBuilder(mainEventBus) + .taskId(TASK_ID) + .mainEventBus(mainEventBus) + .build().tap(); + } + + @AfterEach + public void tearDown() { + if (mainEventBusProcessor != null) { + mainEventBusProcessor.setCallback(null); + EventQueueUtil.stop(mainEventBusProcessor); + } + + // Clean up log appender + Logger logger = (Logger) LoggerFactory.getLogger(MainEventBusProcessor.class); + logger.detachAppender(logAppender); + } + + /** + * Test that TaskSerializationException is converted to InternalError with ERROR log. + * AC#1: Mock TaskStore throws TaskSerializationException → MainEventBusProcessor distributes InternalError + */ + @Test + public void testTaskSerializationException_ConvertsToInternalError() throws InterruptedException { + // Arrange: Mock TaskStore to throw TaskSerializationException + String exceptionMessage = "Failed to deserialize corrupted JSON"; + TaskSerializationException exception = new TaskSerializationException(TASK_ID, exceptionMessage); + when(mockTaskStore.get(any())).thenThrow(exception); + doThrow(exception).when(mockTaskStore).save(any(Task.class), anyBoolean()); + + Task testTask = createTestTask(); + + // Act: Enqueue event and wait for processing + List distributedEvents = captureDistributedEvent(testTask); + + // Assert: Verify InternalError was distributed + assertEquals(1, distributedEvents.size(), "Should distribute exactly one event"); + Event distributedEvent = distributedEvents.get(0); + assertInstanceOf(InternalError.class, distributedEvent, + "TaskSerializationException should convert to InternalError"); + + InternalError error = (InternalError) distributedEvent; + assertTrue(error.getMessage().contains(TASK_ID), + "Error message should contain task ID: " + error.getMessage()); + assertTrue(error.getMessage().contains("serialize"), + "Error message should mention serialization: " + error.getMessage()); + + // Assert: Verify ERROR level logging + boolean foundErrorLog = logAppender.list.stream() + .anyMatch(event -> event.getLevel() == Level.ERROR + && event.getFormattedMessage().contains(TASK_ID) + && event.getFormattedMessage().contains("serialization")); + assertTrue(foundErrorLog, "Should log TaskSerializationException at ERROR level"); + } + + /** + * Test that TaskPersistenceException (transient) is converted to InternalError with WARN log. + * AC#2: Mock TaskStore throws TaskPersistenceException(transient) → WARN log + InternalError + */ + @Test + public void testTaskPersistenceException_Transient_ConvertsToInternalError() throws InterruptedException { + // Arrange: Mock TaskStore to throw transient TaskPersistenceException + String exceptionMessage = "Connection timeout to database"; + TaskPersistenceException exception = new TaskPersistenceException( + TASK_ID, exceptionMessage, true // transient = true + ); + when(mockTaskStore.get(any())).thenThrow(exception); + doThrow(exception).when(mockTaskStore).save(any(Task.class), anyBoolean()); + + Task testTask = createTestTask(); + + // Act: Enqueue event and wait for processing + List distributedEvents = captureDistributedEvent(testTask); + + // Assert: Verify InternalError was distributed + assertEquals(1, distributedEvents.size(), "Should distribute exactly one event"); + Event distributedEvent = distributedEvents.get(0); + assertInstanceOf(InternalError.class, distributedEvent, + "TaskPersistenceException should convert to InternalError"); + + InternalError error = (InternalError) distributedEvent; + assertTrue(error.getMessage().contains(TASK_ID), + "Error message should contain task ID: " + error.getMessage()); + assertTrue(error.getMessage().contains("retry may succeed"), + "Transient error message should mention retry: " + error.getMessage()); + + // Assert: Verify WARN level logging for transient failures + boolean foundWarnLog = logAppender.list.stream() + .anyMatch(event -> event.getLevel() == Level.WARN + && event.getFormattedMessage().contains(TASK_ID) + && event.getFormattedMessage().contains("TRANSIENT")); + assertTrue(foundWarnLog, "Should log transient TaskPersistenceException at WARN level"); + } + + /** + * Test that TaskPersistenceException (permanent) is converted to InternalError with ERROR log. + * AC#3: Mock TaskStore throws TaskPersistenceException(permanent) → ERROR log + InternalError + */ + @Test + public void testTaskPersistenceException_Permanent_ConvertsToInternalError() throws InterruptedException { + // Arrange: Mock TaskStore to throw permanent TaskPersistenceException + String exceptionMessage = "Disk full - cannot write"; + TaskPersistenceException exception = new TaskPersistenceException( + TASK_ID, exceptionMessage, false // transient = false (permanent) + ); + when(mockTaskStore.get(any())).thenThrow(exception); + doThrow(exception).when(mockTaskStore).save(any(Task.class), anyBoolean()); + + Task testTask = createTestTask(); + + // Act: Enqueue event and wait for processing + List distributedEvents = captureDistributedEvent(testTask); + + // Assert: Verify InternalError was distributed + assertEquals(1, distributedEvents.size(), "Should distribute exactly one event"); + Event distributedEvent = distributedEvents.get(0); + assertInstanceOf(InternalError.class, distributedEvent, + "TaskPersistenceException should convert to InternalError"); + + InternalError error = (InternalError) distributedEvent; + assertTrue(error.getMessage().contains(TASK_ID), + "Error message should contain task ID: " + error.getMessage()); + assertTrue(error.getMessage().contains("requires intervention"), + "Permanent error message should mention intervention: " + error.getMessage()); + + // Assert: Verify ERROR level logging for permanent failures + boolean foundErrorLog = logAppender.list.stream() + .anyMatch(event -> event.getLevel() == Level.ERROR + && event.getFormattedMessage().contains(TASK_ID) + && event.getFormattedMessage().contains("PERMANENT")); + assertTrue(foundErrorLog, "Should log permanent TaskPersistenceException at ERROR level"); + } + + /** + * Test that taskId is preserved through exception chain and appears in error messages. + * AC#5: All tests validate error messages contain taskId and failure type + */ + @Test + public void testTaskIdPreservedInExceptionChain() throws InterruptedException { + // Arrange: Create exception with specific taskId + String specificTaskId = "task-with-unique-id-12345"; + TaskSerializationException exception = new TaskSerializationException( + specificTaskId, "Test exception with specific task ID" + ); + when(mockTaskStore.get(any())).thenThrow(exception); + doThrow(exception).when(mockTaskStore).save(any(Task.class), anyBoolean()); + + // Create event queue with specific taskId + EventQueue specificQueue = EventQueueUtil.getEventQueueBuilder(mainEventBus) + .taskId(specificTaskId) + .mainEventBus(mainEventBus) + .build().tap(); + + Task testTask = Task.builder() + .id(specificTaskId) + .contextId("test-context") + .status(new TaskStatus(TaskState.TASK_STATE_SUBMITTED)) + .build(); + + // Act: Enqueue event and wait for processing + List distributedEvents = new ArrayList<>(); + CountDownLatch latch = new CountDownLatch(1); + mainEventBusProcessor.setCallback(new MainEventBusProcessorCallback() { + @Override + public void onEventProcessed(String taskId, Event event) { + distributedEvents.add(event); + latch.countDown(); + } + + @Override + public void onTaskFinalized(String taskId) { + // No-op for this test + } + }); + + specificQueue.enqueueEvent(testTask); + assertTrue(latch.await(5, TimeUnit.SECONDS), "Event processing should complete within timeout"); + + // Assert: Verify specific taskId appears in distributed error + assertEquals(1, distributedEvents.size()); + InternalError error = (InternalError) distributedEvents.get(0); + assertTrue(error.getMessage().contains(specificTaskId), + "Error should contain specific task ID: " + error.getMessage()); + + // Assert: Verify specific taskId appears in logs + boolean foundTaskIdInLog = logAppender.list.stream() + .anyMatch(event -> event.getFormattedMessage().contains(specificTaskId)); + assertTrue(foundTaskIdInLog, "Logs should contain specific task ID"); + } + + /** + * Helper method to create a test Task. + */ + private Task createTestTask() { + return Task.builder() + .id(TASK_ID) + .contextId("test-context") + .status(new TaskStatus(TaskState.TASK_STATE_SUBMITTED)) + .build(); + } + + /** + * Helper method to enqueue an event and capture what gets distributed to clients. + * Uses MainEventBusProcessorCallback to wait for async processing. + * + * @param event the event to enqueue + * @return list of events distributed to ChildQueues (should be 1 event) + */ + private List captureDistributedEvent(Event event) throws InterruptedException { + List distributedEvents = new ArrayList<>(); + CountDownLatch latch = new CountDownLatch(1); + + mainEventBusProcessor.setCallback(new MainEventBusProcessorCallback() { + @Override + public void onEventProcessed(String taskId, Event processedEvent) { + distributedEvents.add(processedEvent); + latch.countDown(); + } + + @Override + public void onTaskFinalized(String taskId) { + // No-op for exception tests + } + }); + + eventQueue.enqueueEvent(event); + + assertTrue(latch.await(5, TimeUnit.SECONDS), + "Event processing should complete within timeout"); + + return distributedEvents; + } +} diff --git a/server-common/src/test/java/io/a2a/server/tasks/AbstractTaskStoreExceptionTest.java b/server-common/src/test/java/io/a2a/server/tasks/AbstractTaskStoreExceptionTest.java new file mode 100644 index 000000000..9646c24bf --- /dev/null +++ b/server-common/src/test/java/io/a2a/server/tasks/AbstractTaskStoreExceptionTest.java @@ -0,0 +1,168 @@ +package io.a2a.server.tasks; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; + +import org.junit.jupiter.api.Test; + +/** + * Base test class for TaskStore exception validation. + *

    + * Provides reusable test patterns for exception construction, field verification, + * and message formatting. Subclasses must implement {@link #createException} methods + * to test specific exception types. + *

    + * This class is designed to be extended by implementation tests (e.g., InMemoryTaskStore tests) + * to ensure consistent exception behavior across all TaskStore implementations. + * + * @param the exception type being tested (must extend TaskStoreException) + */ +public abstract class AbstractTaskStoreExceptionTest { + + /** + * Creates an exception with a message and taskId. + * Used for testing basic exception construction with task context. + * + * @param taskId the task identifier + * @param message the exception message + * @return the constructed exception + */ + protected abstract T createException(String taskId, String message); + + /** + * Creates an exception with a taskId, message, and cause. + * Used for testing exception chaining with task context. + * + * @param taskId the task identifier + * @param message the exception message + * @param cause the underlying cause + * @return the constructed exception + */ + protected abstract T createException(String taskId, String message, Throwable cause); + + // ========== Task ID Field Tests ========== + + @Test + void testTaskIdField_withTaskId() { + T exception = createException("task-123", "Test message"); + assertEquals("task-123", exception.getTaskId()); + } + + @Test + void testTaskIdField_nullTaskId() { + T exception = createException(null, "Test message"); + assertNull(exception.getTaskId()); + } + + @Test + void testTaskIdField_emptyTaskId() { + T exception = createException("", "Test message"); + assertEquals("", exception.getTaskId()); + } + + // ========== Message Field Tests ========== + + @Test + void testMessageField_nonNull() { + T exception = createException("task-123", "Failed to save task"); + assertNotNull(exception.getMessage()); + assertEquals("Failed to save task", exception.getMessage()); + } + + @Test + void testMessageField_withContext() { + T exception = createException("task-123", "Database connection timeout"); + assertNotNull(exception.getMessage()); + assertEquals("Database connection timeout", exception.getMessage()); + } + + // ========== Cause Chain Tests ========== + + @Test + void testCauseChain_withCause() { + RuntimeException cause = new RuntimeException("Root cause"); + T exception = createException("task-123", "Wrapper message", cause); + + assertNotNull(exception.getCause()); + assertSame(cause, exception.getCause()); + assertEquals("Root cause", exception.getCause().getMessage()); + } + + @Test + void testCauseChain_multipleLevels() { + RuntimeException rootCause = new RuntimeException("Database error"); + IllegalStateException intermediateCause = new IllegalStateException("Transaction failed", rootCause); + T exception = createException("task-123", "Save failed", intermediateCause); + + assertNotNull(exception.getCause()); + assertSame(intermediateCause, exception.getCause()); + assertNotNull(exception.getCause().getCause()); + assertSame(rootCause, exception.getCause().getCause()); + } + + // ========== Exception Inheritance Tests ========== + + @Test + void testInheritance_isTaskStoreException() { + T exception = createException("task-123", "Test message"); + assertNotNull(exception); + // Verified by generic type constraint: T extends TaskStoreException + } + + @Test + void testInheritance_isThrowable() { + T exception = createException("task-123", "Test message"); + Throwable throwable = exception; + assertNotNull(throwable); + } + + // ========== Message Clarity Tests ========== + + /** + * Verifies that exception messages are clear and actionable. + * Subclasses should override this to test domain-specific message patterns. + */ + @Test + void testMessageClarity_basicPattern() { + T exception = createException("task-123", "Operation failed"); + String message = exception.getMessage(); + + assertNotNull(message); + // Message should not be empty + assert !message.trim().isEmpty() : "Exception message should not be empty"; + // Message should not be too short (less than 5 characters is typically not helpful) + assert message.length() >= 5 : "Exception message should be descriptive"; + } + + // ========== Helper Assertions for Subclasses ========== + + /** + * Asserts that an exception message contains expected context information. + * Useful for implementation tests to verify TaskStore-specific message patterns. + * + * @param exception the exception to check + * @param expectedSubstring the expected substring in the message + */ + protected void assertMessageContains(T exception, String expectedSubstring) { + assertNotNull(exception.getMessage()); + assert exception.getMessage().contains(expectedSubstring) + : String.format("Expected message to contain '%s' but was: %s", + expectedSubstring, exception.getMessage()); + } + + /** + * Asserts that an exception has both taskId and cause properly set. + * Useful for implementation tests to verify complete exception context. + * + * @param exception the exception to check + * @param expectedTaskId the expected task ID + * @param expectedCause the expected cause + */ + protected void assertFullContext(T exception, String expectedTaskId, Throwable expectedCause) { + assertEquals(expectedTaskId, exception.getTaskId()); + assertSame(expectedCause, exception.getCause()); + assertNotNull(exception.getMessage()); + } +} diff --git a/server-common/src/test/java/io/a2a/server/tasks/TaskPersistenceExceptionTest.java b/server-common/src/test/java/io/a2a/server/tasks/TaskPersistenceExceptionTest.java new file mode 100644 index 000000000..bf0fe5388 --- /dev/null +++ b/server-common/src/test/java/io/a2a/server/tasks/TaskPersistenceExceptionTest.java @@ -0,0 +1,419 @@ +package io.a2a.server.tasks; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link TaskPersistenceException}. + *

    + * Tests the exception class for database/storage system failures, + * with special focus on the {@code isTransient} flag for retry logic. + */ +class TaskPersistenceExceptionTest extends AbstractTaskStoreExceptionTest { + + @Override + protected TaskPersistenceException createException(String taskId, String message) { + return new TaskPersistenceException(taskId, message); + } + + @Override + protected TaskPersistenceException createException(String taskId, String message, Throwable cause) { + return new TaskPersistenceException(taskId, message, cause); + } + + // ========== Constructor Tests - Default Non-Transient ========== + + @Test + void testConstructor_noArgs_defaultsToNonTransient() { + TaskPersistenceException exception = new TaskPersistenceException(); + assertNull(exception.getMessage()); + assertNull(exception.getCause()); + assertNull(exception.getTaskId()); + assertFalse(exception.isTransient(), "Default should be non-transient"); + } + + @Test + void testConstructor_messageOnly_defaultsToNonTransient() { + TaskPersistenceException exception = new TaskPersistenceException("Database error"); + assertEquals("Database error", exception.getMessage()); + assertNull(exception.getCause()); + assertNull(exception.getTaskId()); + assertFalse(exception.isTransient(), "Default should be non-transient"); + } + + @Test + void testConstructor_causeOnly_defaultsToNonTransient() { + RuntimeException cause = new RuntimeException("Connection failed"); + TaskPersistenceException exception = new TaskPersistenceException(cause); + + assertNotNull(exception.getMessage()); + assertSame(cause, exception.getCause()); + assertNull(exception.getTaskId()); + assertFalse(exception.isTransient(), "Default should be non-transient"); + } + + @Test + void testConstructor_messageAndCause_defaultsToNonTransient() { + RuntimeException cause = new RuntimeException("Timeout"); + TaskPersistenceException exception = new TaskPersistenceException("Operation failed", cause); + + assertEquals("Operation failed", exception.getMessage()); + assertSame(cause, exception.getCause()); + assertNull(exception.getTaskId()); + assertFalse(exception.isTransient(), "Default should be non-transient"); + } + + @Test + void testConstructor_taskIdAndMessage_defaultsToNonTransient() { + TaskPersistenceException exception = new TaskPersistenceException("task-123", "Save failed"); + + assertEquals("Save failed", exception.getMessage()); + assertEquals("task-123", exception.getTaskId()); + assertNull(exception.getCause()); + assertFalse(exception.isTransient(), "Default should be non-transient"); + } + + @Test + void testConstructor_taskIdMessageAndCause_defaultsToNonTransient() { + RuntimeException cause = new RuntimeException("Disk error"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-456", "Persistence failed", cause); + + assertEquals("Persistence failed", exception.getMessage()); + assertEquals("task-456", exception.getTaskId()); + assertSame(cause, exception.getCause()); + assertFalse(exception.isTransient(), "Default should be non-transient"); + } + + // ========== Constructor Tests - Explicit Transient Flag ========== + + @Test + void testConstructor_messageWithTransientTrue() { + TaskPersistenceException exception = new TaskPersistenceException("Connection timeout", true); + + assertEquals("Connection timeout", exception.getMessage()); + assertTrue(exception.isTransient(), "Should be transient when explicitly set"); + } + + @Test + void testConstructor_messageWithTransientFalse() { + TaskPersistenceException exception = new TaskPersistenceException("Disk full", false); + + assertEquals("Disk full", exception.getMessage()); + assertFalse(exception.isTransient(), "Should be non-transient when explicitly set"); + } + + @Test + void testConstructor_messageAndCauseWithTransientTrue() { + RuntimeException cause = new RuntimeException("Network partition"); + TaskPersistenceException exception = new TaskPersistenceException( + "Database unreachable", cause, true); + + assertEquals("Database unreachable", exception.getMessage()); + assertSame(cause, exception.getCause()); + assertTrue(exception.isTransient(), "Should be transient"); + } + + @Test + void testConstructor_messageAndCauseWithTransientFalse() { + RuntimeException cause = new RuntimeException("Constraint violation"); + TaskPersistenceException exception = new TaskPersistenceException( + "Duplicate key error", cause, false); + + assertEquals("Duplicate key error", exception.getMessage()); + assertSame(cause, exception.getCause()); + assertFalse(exception.isTransient(), "Should be non-transient"); + } + + @Test + void testConstructor_taskIdMessageWithTransientTrue() { + TaskPersistenceException exception = new TaskPersistenceException( + "task-123", "Lock timeout", true); + + assertEquals("task-123", exception.getTaskId()); + assertEquals("Lock timeout", exception.getMessage()); + assertTrue(exception.isTransient(), "Should be transient"); + } + + @Test + void testConstructor_taskIdMessageWithTransientFalse() { + TaskPersistenceException exception = new TaskPersistenceException( + "task-456", "Storage quota exceeded", false); + + assertEquals("task-456", exception.getTaskId()); + assertEquals("Storage quota exceeded", exception.getMessage()); + assertFalse(exception.isTransient(), "Should be non-transient"); + } + + @Test + void testConstructor_fullArgsWithTransientTrue() { + RuntimeException cause = new RuntimeException("Deadlock detected"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-789", "Transaction rolled back", cause, true); + + assertEquals("task-789", exception.getTaskId()); + assertEquals("Transaction rolled back", exception.getMessage()); + assertSame(cause, exception.getCause()); + assertTrue(exception.isTransient(), "Should be transient"); + } + + @Test + void testConstructor_fullArgsWithTransientFalse() { + RuntimeException cause = new RuntimeException("Permission denied"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-abc", "Access control violation", cause, false); + + assertEquals("task-abc", exception.getTaskId()); + assertEquals("Access control violation", exception.getMessage()); + assertSame(cause, exception.getCause()); + assertFalse(exception.isTransient(), "Should be non-transient"); + } + + // ========== Inheritance Verification ========== + + @Test + void testInheritance_extendsTaskStoreException() { + TaskPersistenceException exception = new TaskPersistenceException("test", "Test message"); + TaskStoreException baseException = exception; + assertNotNull(baseException); + } + + @Test + void testInheritance_taskIdFieldAccessible() { + TaskPersistenceException exception = new TaskPersistenceException("task-xyz", "Test message", true); + // Should be accessible via TaskStoreException parent class + assertEquals("task-xyz", exception.getTaskId()); + } + + // ========== Transient Flag Semantics Tests ========== + + @Test + void testTransientFlag_indicatesRetryability() { + TaskPersistenceException transientError = new TaskPersistenceException( + "task-123", "Connection timeout", true); + TaskPersistenceException permanentError = new TaskPersistenceException( + "task-456", "Disk full", false); + + assertTrue(transientError.isTransient(), "Timeout errors should be retryable"); + assertFalse(permanentError.isTransient(), "Disk full is not retryable"); + } + + // ========== Real-World Scenario Tests - Transient Failures ========== + + @Test + void testScenario_transient_connectionTimeout() { + RuntimeException timeout = new RuntimeException("Connection timeout after 30s"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-123", "Database connection timeout", timeout, true); + + assertFullContext(exception, "task-123", timeout); + assertTrue(exception.isTransient(), "Connection timeouts are transient"); + assertMessageContains(exception, "timeout"); + } + + @Test + void testScenario_transient_deadlock() { + RuntimeException deadlock = new RuntimeException("Deadlock detected"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-456", "Transaction deadlock, retry recommended", deadlock, true); + + assertTrue(exception.isTransient(), "Deadlocks are transient"); + assertMessageContains(exception, "deadlock"); + assertMessageContains(exception, "retry"); + } + + @Test + void testScenario_transient_lockTimeout() { + RuntimeException lockTimeout = new RuntimeException("Lock wait timeout exceeded"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-789", "Failed to acquire row lock", lockTimeout, true); + + assertTrue(exception.isTransient(), "Lock timeouts are transient"); + assertMessageContains(exception, "lock"); + } + + @Test + void testScenario_transient_networkPartition() { + RuntimeException networkError = new RuntimeException("Network unreachable"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-abc", "Database host unreachable due to network partition", networkError, true); + + assertTrue(exception.isTransient(), "Network errors are transient"); + assertMessageContains(exception, "network"); + } + + @Test + void testScenario_transient_poolExhausted() { + RuntimeException poolError = new RuntimeException("Connection pool exhausted"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-def", "No database connections available", poolError, true); + + assertTrue(exception.isTransient(), "Pool exhaustion is transient"); + assertMessageContains(exception, "connections available"); + } + + // ========== Real-World Scenario Tests - Non-Transient Failures ========== + + @Test + void testScenario_nonTransient_diskFull() { + RuntimeException diskError = new RuntimeException("No space left on device"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-123", "Cannot write task: disk full", diskError, false); + + assertFalse(exception.isTransient(), "Disk full is not transient"); + assertMessageContains(exception, "disk full"); + } + + @Test + void testScenario_nonTransient_uniqueConstraint() { + RuntimeException constraintError = new RuntimeException("Duplicate entry for key 'PRIMARY'"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-456", "Task ID already exists", constraintError, false); + + assertFalse(exception.isTransient(), "Unique constraint violations are not transient"); + assertMessageContains(exception, "already exists"); + } + + @Test + void testScenario_nonTransient_foreignKeyViolation() { + RuntimeException fkError = new RuntimeException("Cannot add or update child row"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-789", "Foreign key constraint violation", fkError, false); + + assertFalse(exception.isTransient(), "Foreign key violations are not transient"); + assertMessageContains(exception, "constraint"); + } + + @Test + void testScenario_nonTransient_permissionDenied() { + RuntimeException permError = new RuntimeException("Access denied for user 'app'"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-abc", "Insufficient database permissions", permError, false); + + assertFalse(exception.isTransient(), "Permission errors are not transient"); + assertMessageContains(exception, "permission"); + } + + @Test + void testScenario_nonTransient_schemaIncompatibility() { + RuntimeException schemaError = new RuntimeException("Column 'new_field' does not exist"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-def", "Database schema incompatible with application version", schemaError, false); + + assertFalse(exception.isTransient(), "Schema incompatibilities are not transient"); + assertMessageContains(exception, "schema"); + } + + @Test + void testScenario_nonTransient_quotaExceeded() { + RuntimeException quotaError = new RuntimeException("Storage quota exceeded"); + TaskPersistenceException exception = new TaskPersistenceException( + "task-ghi", "Database storage limit reached", quotaError, false); + + assertFalse(exception.isTransient(), "Quota violations are not transient"); + assertMessageContains(exception, "storage limit"); + } + + // ========== Message Quality Tests - Persistence Context ========== + + @Test + void testMessage_connectionError() { + TaskPersistenceException exception = new TaskPersistenceException( + "task-123", + "Failed to connect to database at jdbc:postgresql://localhost:5432/a2a", + true); + + assertMessageContains(exception, "connect to database"); + assertMessageContains(exception, "jdbc:postgresql"); + } + + @Test + void testMessage_transactionRollback() { + TaskPersistenceException exception = new TaskPersistenceException( + "task-123", + "Transaction rolled back due to deadlock. Retry operation.", + true); + + assertMessageContains(exception, "Transaction rolled back"); + assertMessageContains(exception, "Retry operation"); + } + + @Test + void testMessage_constraintViolation() { + TaskPersistenceException exception = new TaskPersistenceException( + "task-123", + "Unique constraint violation: task_id 'task-123' already exists in table 'tasks'", + false); + + assertMessageContains(exception, "Unique constraint"); + assertMessageContains(exception, "already exists"); + } + + // ========== Message Actionability Tests ========== + + @Test + void testMessageActionable_transientWithRetryGuidance() { + TaskPersistenceException exception = new TaskPersistenceException( + "task-123", + "Connection timeout. Retry with exponential backoff (attempt 1 of 3).", + true); + + assertMessageContains(exception, "Retry with exponential backoff"); + assertMessageContains(exception, "attempt 1 of 3"); + } + + @Test + void testMessageActionable_nonTransientWithResolution() { + TaskPersistenceException exception = new TaskPersistenceException( + "task-123", + "Disk full. Free up space or increase storage capacity to continue.", + false); + + assertMessageContains(exception, "Free up space"); + assertMessageContains(exception, "increase storage"); + } + + // ========== Cause Chain for Retry Logic ========== + + @Test + void testCauseChain_transientErrorWithContext() { + RuntimeException sqlError = new RuntimeException("SQL state: 08006 - Connection failure"); + RuntimeException jdbcError = new RuntimeException("JDBC connection error", sqlError); + TaskPersistenceException exception = new TaskPersistenceException( + "task-xyz", "Database operation failed", jdbcError, true); + + assertTrue(exception.isTransient()); + assertEquals("task-xyz", exception.getTaskId()); + assertSame(jdbcError, exception.getCause()); + assertSame(sqlError, exception.getCause().getCause()); + } + + // ========== Null Safety ========== + + @Test + void testNullSafety_nullTaskIdWithTransient() { + TaskPersistenceException exception = new TaskPersistenceException( + null, "Generic database error", true); + + assertNull(exception.getTaskId()); + assertTrue(exception.isTransient()); + assertEquals("Generic database error", exception.getMessage()); + } + + @Test + void testNullSafety_allNullsWithTransient() { + TaskPersistenceException exception = new TaskPersistenceException( + null, (String) null, (Throwable) null, true); + + assertNull(exception.getTaskId()); + assertNull(exception.getMessage()); + assertNull(exception.getCause()); + assertTrue(exception.isTransient()); + } +} diff --git a/server-common/src/test/java/io/a2a/server/tasks/TaskSerializationExceptionTest.java b/server-common/src/test/java/io/a2a/server/tasks/TaskSerializationExceptionTest.java new file mode 100644 index 000000000..e843adb68 --- /dev/null +++ b/server-common/src/test/java/io/a2a/server/tasks/TaskSerializationExceptionTest.java @@ -0,0 +1,272 @@ +package io.a2a.server.tasks; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; + +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link TaskSerializationException}. + *

    + * Tests the exception class for task serialization/deserialization failures, + * verifying all constructor variants and non-transient failure semantics. + */ +class TaskSerializationExceptionTest extends AbstractTaskStoreExceptionTest { + + @Override + protected TaskSerializationException createException(String taskId, String message) { + return new TaskSerializationException(taskId, message); + } + + @Override + protected TaskSerializationException createException(String taskId, String message, Throwable cause) { + return new TaskSerializationException(taskId, message, cause); + } + + // ========== Constructor Tests ========== + + @Test + void testConstructor_noArgs() { + TaskSerializationException exception = new TaskSerializationException(); + assertNull(exception.getMessage()); + assertNull(exception.getCause()); + assertNull(exception.getTaskId()); + } + + @Test + void testConstructor_messageOnly() { + TaskSerializationException exception = new TaskSerializationException("JSON parsing failed"); + assertEquals("JSON parsing failed", exception.getMessage()); + assertNull(exception.getCause()); + assertNull(exception.getTaskId()); + } + + @Test + void testConstructor_causeOnly() { + RuntimeException cause = new RuntimeException("Invalid JSON format"); + TaskSerializationException exception = new TaskSerializationException(cause); + + assertNotNull(exception.getMessage()); + // Exception message should contain cause class name + assert exception.getMessage().contains("RuntimeException"); + assertSame(cause, exception.getCause()); + assertNull(exception.getTaskId()); + } + + @Test + void testConstructor_messageAndCause() { + RuntimeException cause = new RuntimeException("Unexpected field type"); + TaskSerializationException exception = new TaskSerializationException("Deserialization failed", cause); + + assertEquals("Deserialization failed", exception.getMessage()); + assertSame(cause, exception.getCause()); + assertNull(exception.getTaskId()); + } + + @Test + void testConstructor_taskIdAndMessage() { + TaskSerializationException exception = new TaskSerializationException( + "task-123", "Failed to serialize task"); + + assertEquals("Failed to serialize task", exception.getMessage()); + assertEquals("task-123", exception.getTaskId()); + assertNull(exception.getCause()); + } + + @Test + void testConstructor_taskIdMessageAndCause() { + RuntimeException cause = new RuntimeException("Missing required field"); + TaskSerializationException exception = new TaskSerializationException( + "task-456", "Task deserialization failed", cause); + + assertEquals("Task deserialization failed", exception.getMessage()); + assertEquals("task-456", exception.getTaskId()); + assertSame(cause, exception.getCause()); + } + + // ========== Inheritance Verification ========== + + @Test + void testInheritance_extendsTaskStoreException() { + TaskSerializationException exception = new TaskSerializationException("test", "Test message"); + TaskStoreException baseException = exception; + assertNotNull(baseException); + } + + @Test + void testInheritance_taskIdFieldAccessible() { + TaskSerializationException exception = new TaskSerializationException("task-789", "Test message"); + // Should be accessible via TaskStoreException parent class + assertEquals("task-789", exception.getTaskId()); + } + + // ========== Message Quality Tests - Serialization Context ========== + + @Test + void testMessage_jsonParsingError() { + TaskSerializationException exception = new TaskSerializationException( + "task-123", + "Failed to deserialize task: unexpected token at line 42, column 15"); + + assertNotNull(exception.getMessage()); + assertMessageContains(exception, "deserialize"); + assertMessageContains(exception, "line 42"); + } + + @Test + void testMessage_invalidFieldType() { + TaskSerializationException exception = new TaskSerializationException( + "task-123", + "Field 'status' expected enum TaskState, got string 'INVALID'"); + + assertNotNull(exception.getMessage()); + assertMessageContains(exception, "Field 'status'"); + assertMessageContains(exception, "expected enum"); + } + + @Test + void testMessage_missingRequiredField() { + TaskSerializationException exception = new TaskSerializationException( + "task-123", + "Missing required field 'id' during task deserialization"); + + assertNotNull(exception.getMessage()); + assertMessageContains(exception, "Missing required field"); + assertMessageContains(exception, "'id'"); + } + + @Test + void testMessage_schemaVersionMismatch() { + TaskSerializationException exception = new TaskSerializationException( + "task-123", + "Task schema version 2.0 not compatible with current version 1.0"); + + assertNotNull(exception.getMessage()); + assertMessageContains(exception, "schema version"); + assertMessageContains(exception, "not compatible"); + } + + // ========== Real-World Scenario Tests ========== + + @Test + void testScenario_jsonProcessingException() { + // Simulate Jackson JsonProcessingException wrapping + RuntimeException jsonError = new RuntimeException( + "Unrecognized field \"unknownField\" (class Task), not marked as ignorable"); + TaskSerializationException exception = new TaskSerializationException( + "task-abc", "Failed to deserialize task from JSON", jsonError); + + assertFullContext(exception, "task-abc", jsonError); + assertMessageContains(exception, "Failed to deserialize"); + } + + @Test + void testScenario_enumConversionError() { + RuntimeException enumError = new RuntimeException( + "Cannot deserialize value of type `TaskState` from String \"INVALID_STATE\""); + TaskSerializationException exception = new TaskSerializationException( + "task-def", "Invalid enum value in task JSON", enumError); + + assertFullContext(exception, "task-def", enumError); + assertMessageContains(exception, "Invalid enum value"); + } + + @Test + void testScenario_nullValueError() { + TaskSerializationException exception = new TaskSerializationException( + "task-ghi", + "Required field 'taskId' cannot be null during deserialization"); + + assertEquals("task-ghi", exception.getTaskId()); + assertMessageContains(exception, "cannot be null"); + } + + // ========== Cause Chain for Debugging ========== + + @Test + void testCauseChain_multiLevelSerializationError() { + // Simulate nested serialization error with multiple layers + RuntimeException rootCause = new RuntimeException("Invalid UTF-8 sequence at byte 1024"); + IllegalArgumentException parseError = new IllegalArgumentException("Cannot parse JSON", rootCause); + TaskSerializationException exception = new TaskSerializationException( + "task-xyz", "Task deserialization failed", parseError); + + // Verify full chain for debugging + assertEquals("task-xyz", exception.getTaskId()); + assertSame(parseError, exception.getCause()); + assertSame(rootCause, exception.getCause().getCause()); + assertMessageContains(exception, "deserialization failed"); + } + + // ========== Edge Cases ========== + + @Test + void testEdgeCase_emptyJsonError() { + TaskSerializationException exception = new TaskSerializationException( + "task-empty", "Cannot deserialize empty JSON string"); + + assertEquals("task-empty", exception.getTaskId()); + assertMessageContains(exception, "empty JSON"); + } + + @Test + void testEdgeCase_circularReferenceError() { + TaskSerializationException exception = new TaskSerializationException( + "task-circular", + "Infinite recursion detected during serialization (StackOverflowError)"); + + assertEquals("task-circular", exception.getTaskId()); + assertMessageContains(exception, "Infinite recursion"); + } + + @Test + void testEdgeCase_characterEncodingError() { + RuntimeException encodingError = new RuntimeException("Invalid character encoding"); + TaskSerializationException exception = new TaskSerializationException( + "task-encoding", "Failed to encode task JSON", encodingError); + + assertFullContext(exception, "task-encoding", encodingError); + } + + // ========== Message Actionability Tests ========== + + @Test + void testMessageActionable_providesContext() { + TaskSerializationException exception = new TaskSerializationException( + "task-123", + "Failed to deserialize task: field 'createdAt' expects ISO-8601 timestamp, got '2024-13-45'"); + + // Message should help developer understand the problem + assertMessageContains(exception, "field 'createdAt'"); + assertMessageContains(exception, "expects ISO-8601"); + assertMessageContains(exception, "got '2024-13-45'"); + } + + @Test + void testMessageActionable_suggestsResolution() { + TaskSerializationException exception = new TaskSerializationException( + "task-123", + "Schema version mismatch. Run database migration to update task format to v2.0"); + + assertMessageContains(exception, "Schema version"); + assertMessageContains(exception, "Run database migration"); + } + + // ========== Null Safety ========== + + @Test + void testNullSafety_nullTaskIdWithMessage() { + TaskSerializationException exception = new TaskSerializationException(null, "Generic serialization error"); + assertNull(exception.getTaskId()); + assertEquals("Generic serialization error", exception.getMessage()); + } + + @Test + void testNullSafety_nullMessageWithTaskId() { + TaskSerializationException exception = new TaskSerializationException("task-123", (String) null); + assertEquals("task-123", exception.getTaskId()); + assertNull(exception.getMessage()); + } +} diff --git a/server-common/src/test/java/io/a2a/server/tasks/TaskStoreExceptionTest.java b/server-common/src/test/java/io/a2a/server/tasks/TaskStoreExceptionTest.java new file mode 100644 index 000000000..351e4332f --- /dev/null +++ b/server-common/src/test/java/io/a2a/server/tasks/TaskStoreExceptionTest.java @@ -0,0 +1,191 @@ +package io.a2a.server.tasks; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; + +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link TaskStoreException}. + *

    + * Tests the base exception class for TaskStore persistence layer failures, + * verifying all constructor variants and field behavior. + */ +class TaskStoreExceptionTest extends AbstractTaskStoreExceptionTest { + + @Override + protected TaskStoreException createException(String taskId, String message) { + return new TaskStoreException(taskId, message); + } + + @Override + protected TaskStoreException createException(String taskId, String message, Throwable cause) { + return new TaskStoreException(taskId, message, cause); + } + + // ========== Constructor Tests ========== + + @Test + void testConstructor_noArgs() { + TaskStoreException exception = new TaskStoreException(); + assertNull(exception.getMessage()); + assertNull(exception.getCause()); + assertNull(exception.getTaskId()); + } + + @Test + void testConstructor_messageOnly() { + TaskStoreException exception = new TaskStoreException("Failed to persist task"); + assertEquals("Failed to persist task", exception.getMessage()); + assertNull(exception.getCause()); + assertNull(exception.getTaskId()); + } + + @Test + void testConstructor_causeOnly() { + RuntimeException cause = new RuntimeException("Database error"); + TaskStoreException exception = new TaskStoreException(cause); + + assertNotNull(exception.getMessage()); + // Exception message should contain cause class name + assert exception.getMessage().contains("RuntimeException"); + assertSame(cause, exception.getCause()); + assertNull(exception.getTaskId()); + } + + @Test + void testConstructor_messageAndCause() { + RuntimeException cause = new RuntimeException("Database error"); + TaskStoreException exception = new TaskStoreException("Persistence failed", cause); + + assertEquals("Persistence failed", exception.getMessage()); + assertSame(cause, exception.getCause()); + assertNull(exception.getTaskId()); + } + + @Test + void testConstructor_taskIdAndMessage() { + TaskStoreException exception = new TaskStoreException("task-123", "Failed to save"); + + assertEquals("Failed to save", exception.getMessage()); + assertEquals("task-123", exception.getTaskId()); + assertNull(exception.getCause()); + } + + @Test + void testConstructor_taskIdMessageAndCause() { + RuntimeException cause = new RuntimeException("Connection timeout"); + TaskStoreException exception = new TaskStoreException("task-456", "Save operation failed", cause); + + assertEquals("Save operation failed", exception.getMessage()); + assertEquals("task-456", exception.getTaskId()); + assertSame(cause, exception.getCause()); + } + + // ========== Task ID Edge Cases ========== + + @Test + void testTaskId_uuid() { + String uuid = "550e8400-e29b-41d4-a716-446655440000"; + TaskStoreException exception = new TaskStoreException(uuid, "Test message"); + assertEquals(uuid, exception.getTaskId()); + } + + @Test + void testTaskId_numericString() { + TaskStoreException exception = new TaskStoreException("12345", "Test message"); + assertEquals("12345", exception.getTaskId()); + } + + @Test + void testTaskId_specialCharacters() { + String taskId = "task-123_v2.0"; + TaskStoreException exception = new TaskStoreException(taskId, "Test message"); + assertEquals(taskId, exception.getTaskId()); + } + + // ========== Inheritance Verification ========== + + @Test + void testInheritance_extendsA2AServerException() { + TaskStoreException exception = new TaskStoreException("test", "Test message"); + assertNotNull(exception); + // TaskStoreException extends A2AServerException (verified by compilation) + } + + @Test + void testInheritance_isException() { + TaskStoreException exception = new TaskStoreException("test", "Test message"); + Exception e = exception; + assertNotNull(e); + } + + // ========== Message Quality Tests ========== + + @Test + void testMessage_descriptive() { + TaskStoreException exception = new TaskStoreException( + "task-123", + "Failed to persist task to database: connection timeout after 30s"); + + assertNotNull(exception.getMessage()); + assertMessageContains(exception, "Failed to persist"); + assertMessageContains(exception, "database"); + assertMessageContains(exception, "connection timeout"); + } + + @Test + void testMessage_actionable() { + TaskStoreException exception = new TaskStoreException( + "task-123", + "Task not found in store. Verify taskId and retry operation."); + + assertNotNull(exception.getMessage()); + assertMessageContains(exception, "Task not found"); + assertMessageContains(exception, "Verify taskId"); + } + + // ========== Cause Chain Preservation ========== + + @Test + void testCausePreservation_multipleWrapping() { + RuntimeException rootCause = new RuntimeException("Disk full"); + IllegalStateException level1 = new IllegalStateException("Write failed", rootCause); + IllegalArgumentException level2 = new IllegalArgumentException("Validation failed", level1); + TaskStoreException exception = new TaskStoreException("task-789", "Complete failure", level2); + + // Verify full chain + assertEquals("Complete failure", exception.getMessage()); + assertEquals("task-789", exception.getTaskId()); + assertSame(level2, exception.getCause()); + assertSame(level1, exception.getCause().getCause()); + assertSame(rootCause, exception.getCause().getCause().getCause()); + } + + // ========== Null Safety Tests ========== + + @Test + void testNullSafety_nullMessage() { + TaskStoreException exception = new TaskStoreException("task-123", (String) null); + assertEquals("task-123", exception.getTaskId()); + assertNull(exception.getMessage()); + } + + @Test + void testNullSafety_nullCause() { + TaskStoreException exception = new TaskStoreException("task-123", "Message", (Throwable) null); + assertEquals("task-123", exception.getTaskId()); + assertEquals("Message", exception.getMessage()); + assertNull(exception.getCause()); + } + + @Test + void testNullSafety_allNulls() { + TaskStoreException exception = new TaskStoreException(null, (String) null, (Throwable) null); + assertNull(exception.getTaskId()); + assertNull(exception.getMessage()); + assertNull(exception.getCause()); + } +} From 1cbacc82f3d4712c63757984b3d78b5a67516a0d Mon Sep 17 00:00:00 2001 From: Kabir Khan Date: Tue, 24 Feb 2026 15:53:53 +0000 Subject: [PATCH 2/5] Remove transient flag from TaskPersistenceException At the moment it is just noise --- .../database/jpa/JpaDatabaseTaskStore.java | 80 +----- .../server/events/MainEventBusProcessor.java | 15 +- .../tasks/TaskPersistenceException.java | 93 +------ .../MainEventBusProcessorExceptionTest.java | 62 +---- .../tasks/TaskPersistenceExceptionTest.java | 243 +++--------------- 5 files changed, 68 insertions(+), 425 deletions(-) diff --git a/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java b/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java index 6693ab733..f55d119df 100644 --- a/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java +++ b/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java @@ -66,66 +66,6 @@ void initConfig() { gracePeriodSeconds = Long.parseLong(configProvider.getValue(A2A_REPLICATION_GRACE_PERIOD_SECONDS)); } - /** - * Determines if a database exception represents a transient failure that may succeed on retry. - * - *

    Transient Errors (return true):

    - *
      - *
    • Connection timeouts and network partitions
      • - *
    • Transaction deadlocks and lock timeouts
      • - *
    • Connection pool exhausted
      • - *
    • Query timeouts
      • - *
    - * - *

    Non-Transient Errors (return false):

    - *
      - *
    • Constraint violations (unique key, foreign key)
      • - *
    • Disk full / quota exceeded
      • - *
    • Permission denied
      • - *
    • Schema incompatibilities
      • - *
    - * - * @param e the persistence exception to analyze - * @return true if retry may succeed, false if manual intervention required - */ - private static boolean isTransientDatabaseError(PersistenceException e) { - // Check exception class hierarchy for transient indicators - Throwable cause = e; - while (cause != null) { - String className = cause.getClass().getName(); - String message = cause.getMessage() != null ? cause.getMessage().toLowerCase() : ""; - - // Transient exception types - if (className.contains("Timeout") || - className.contains("LockTimeout") || - className.contains("QueryTimeout")) { - return true; - } - - // Transient message patterns - if (message.contains("timeout") || - message.contains("deadlock") || - message.contains("connection") && (message.contains("refused") || message.contains("closed")) || - message.contains("pool") && message.contains("exhausted")) { - return true; - } - - // Non-transient indicators - if (className.contains("ConstraintViolation") || - className.contains("IntegrityConstraint") || - message.contains("unique constraint") || - message.contains("foreign key") || - message.contains("disk full") || - message.contains("permission denied")) { - return false; - } - - cause = cause.getCause(); - } - - // Default to non-transient for safety (don't retry unless confident) - return false; - } @Transactional @Override @@ -152,10 +92,9 @@ public void save(Task task, boolean isReplicated) { throw new TaskSerializationException(task.id(), "Failed to serialize task for persistence", e); } catch (PersistenceException e) { - boolean isTransient = isTransientDatabaseError(e); - LOGGER.error("Database save failed for task with ID: {} (transient: {})", task.id(), isTransient, e); + LOGGER.error("Database save failed for task with ID: {}", task.id(), e); throw new TaskPersistenceException(task.id(), - "Database save failed for task", e, isTransient); + "Database save failed for task", e); } } @@ -181,10 +120,9 @@ public Task get(String taskId) { } } catch (PersistenceException e) { - boolean isTransient = isTransientDatabaseError(e); - LOGGER.error("Database retrieval failed for task with ID: {} (transient: {})", taskId, isTransient, e); + LOGGER.error("Database retrieval failed for task with ID: {}", taskId, e); throw new TaskPersistenceException(taskId, - "Database retrieval failed for task", e, isTransient); + "Database retrieval failed for task", e); } } @@ -201,10 +139,9 @@ public void delete(String taskId) { LOGGER.debug("Task not found for deletion with ID: {}", taskId); } } catch (PersistenceException e) { - boolean isTransient = isTransientDatabaseError(e); - LOGGER.error("Database deletion failed for task with ID: {} (transient: {})", taskId, isTransient, e); + LOGGER.error("Database deletion failed for task with ID: {}", taskId, e); throw new TaskPersistenceException(taskId, - "Database deletion failed for task", e, isTransient); + "Database deletion failed for task", e); } } @@ -436,10 +373,9 @@ public ListTasksResult list(ListTasksParams params) { throw e; } catch (PersistenceException e) { // Database errors from query creation, execution, or count - boolean isTransient = isTransientDatabaseError(e); - LOGGER.error("Database query failed during list operation (transient: {})", isTransient, e); + LOGGER.error("Database query failed during list operation", e); throw new TaskPersistenceException(null, // No single taskId for list operation - "Database query failed during list operation", e, isTransient); + "Database query failed during list operation", e); } } diff --git a/server-common/src/main/java/io/a2a/server/events/MainEventBusProcessor.java b/server-common/src/main/java/io/a2a/server/events/MainEventBusProcessor.java index c8f7db106..9005765eb 100644 --- a/server-common/src/main/java/io/a2a/server/events/MainEventBusProcessor.java +++ b/server-common/src/main/java/io/a2a/server/events/MainEventBusProcessor.java @@ -316,18 +316,9 @@ private boolean updateTaskStore(String taskId, Event event, boolean isReplicated throw new InternalError("Failed to serialize task " + taskId + ": " + e.getMessage()); } catch (TaskPersistenceException e) { - // Database failure - check if transient for retry guidance - if (e.isTransient()) { - LOGGER.warn("Task {} event persistence failed (TRANSIENT) - retry may succeed: {}", - taskId, e.getMessage()); - throw new InternalError("Temporary storage failure for task " + taskId + - " (retry may succeed): " + e.getMessage()); - } else { - LOGGER.error("Task {} event persistence failed (PERMANENT) - manual intervention required: {}", - taskId, e.getMessage(), e); - throw new InternalError("Permanent storage failure for task " + taskId + - " (requires intervention): " + e.getMessage()); - } + // Database/storage failure + LOGGER.error("Task {} event persistence failed: {}", taskId, e.getMessage(), e); + throw new InternalError("Storage failure for task " + taskId + ": " + e.getMessage()); } catch (InternalError e) { // Already an InternalError from TaskManager validation - pass through diff --git a/server-common/src/main/java/io/a2a/server/tasks/TaskPersistenceException.java b/server-common/src/main/java/io/a2a/server/tasks/TaskPersistenceException.java index 0b40b4310..d5a218d77 100644 --- a/server-common/src/main/java/io/a2a/server/tasks/TaskPersistenceException.java +++ b/server-common/src/main/java/io/a2a/server/tasks/TaskPersistenceException.java @@ -6,36 +6,27 @@ * Exception for database/storage system failures during task persistence operations. *

    * Indicates failures in the underlying storage system (database, filesystem, etc.) rather - * than data format issues. Includes a {@link #isTransient()} flag to distinguish between - * temporary failures (retry recommended) and permanent failures (manual intervention required). - * - *

    Transient Failures (isTransient = true)

    - * Temporary issues that may resolve with retry: + * than data format issues. + *

    + *

    + * Examples of persistence failures: *

      *
    • Database connection timeout or network partition
      • *
    • Transaction deadlock or lock wait timeout
      • *
    • Connection pool exhausted
      • - *
    • Temporary storage system overload
      • - *
    - * Recovery: Exponential backoff retry, circuit breaker patterns - * - *

    Non-Transient Failures (isTransient = false)

    - * Persistent issues requiring intervention: - *
      *
    • Disk full / storage quota exceeded
      • *
    • Database constraint violations (unique key, foreign key)
      • *
    • Insufficient permissions or authentication failures
      • *
    • Database schema incompatibilities
      • *
    - * Recovery: Manual intervention, capacity planning, configuration fixes + *

    * *

    Usage Example

    * 
    {@code
  * try {
  *     em.merge(jpaTask);
  * } catch (PersistenceException e) {
- *     boolean transient = isTransientDatabaseError(e);
- *     throw new TaskPersistenceException(taskId, "Database save failed", e, transient);
+ *     throw new TaskPersistenceException(taskId, "Database save failed", e);
  * }
  * }
    * @@ -44,66 +35,53 @@ */ public class TaskPersistenceException extends TaskStoreException { - private final boolean isTransientFailure; - /** * Creates a new TaskPersistenceException with no message or cause. - * The failure is assumed to be non-transient. */ public TaskPersistenceException() { super(); - this.isTransientFailure = false; } /** * Creates a new TaskPersistenceException with the specified message. - * The failure is assumed to be non-transient. * * @param msg the exception message */ public TaskPersistenceException(final String msg) { super(msg); - this.isTransientFailure = false; } /** * Creates a new TaskPersistenceException with the specified cause. - * The failure is assumed to be non-transient. * * @param cause the underlying cause */ public TaskPersistenceException(final Throwable cause) { super(cause); - this.isTransientFailure = false; } /** * Creates a new TaskPersistenceException with the specified message and cause. - * The failure is assumed to be non-transient. * * @param msg the exception message * @param cause the underlying cause */ public TaskPersistenceException(final String msg, final Throwable cause) { super(msg, cause); - this.isTransientFailure = false; } /** * Creates a new TaskPersistenceException with the specified task ID and message. - * The failure is assumed to be non-transient. * * @param taskId the task identifier (may be null for operations not tied to a specific task) * @param msg the exception message */ public TaskPersistenceException(@Nullable final String taskId, final String msg) { super(taskId, msg); - this.isTransientFailure = false; } /** * Creates a new TaskPersistenceException with the specified task ID, message, and cause. - * The failure is assumed to be non-transient. * * @param taskId the task identifier (may be null for operations not tied to a specific task) * @param msg the exception message @@ -111,64 +89,5 @@ public TaskPersistenceException(@Nullable final String taskId, final String msg) */ public TaskPersistenceException(@Nullable final String taskId, final String msg, final Throwable cause) { super(taskId, msg, cause); - this.isTransientFailure = false; - } - - /** - * Creates a new TaskPersistenceException with the specified message and transient flag. - * - * @param msg the exception message - * @param isTransient true if the failure is transient and may resolve with retry, false otherwise - */ - public TaskPersistenceException(final String msg, final boolean isTransient) { - super(msg); - this.isTransientFailure = isTransient; - } - - /** - * Creates a new TaskPersistenceException with the specified message, cause, and transient flag. - * - * @param msg the exception message - * @param cause the underlying cause - * @param isTransient true if the failure is transient and may resolve with retry, false otherwise - */ - public TaskPersistenceException(final String msg, final Throwable cause, final boolean isTransient) { - super(msg, cause); - this.isTransientFailure = isTransient; - } - - /** - * Creates a new TaskPersistenceException with the specified task ID, message, and transient flag. - * - * @param taskId the task identifier (may be null for operations not tied to a specific task) - * @param msg the exception message - * @param isTransient true if the failure is transient and may resolve with retry, false otherwise - */ - public TaskPersistenceException(@Nullable final String taskId, final String msg, final boolean isTransient) { - super(taskId, msg); - this.isTransientFailure = isTransient; - } - - /** - * Creates a new TaskPersistenceException with the specified task ID, message, cause, and transient flag. - * - * @param taskId the task identifier (may be null for operations not tied to a specific task) - * @param msg the exception message - * @param cause the underlying cause - * @param isTransient true if the failure is transient and may resolve with retry, false otherwise - */ - public TaskPersistenceException(@Nullable final String taskId, final String msg, final Throwable cause, - final boolean isTransient) { - super(taskId, msg, cause); - this.isTransientFailure = isTransient; - } - - /** - * Indicates whether this failure is transient (retry may help). - * - * @return true if transient (network, timeout, deadlock), false if permanent (disk full, constraint) - */ - public boolean isTransient() { - return isTransientFailure; } } diff --git a/server-common/src/test/java/io/a2a/server/events/MainEventBusProcessorExceptionTest.java b/server-common/src/test/java/io/a2a/server/events/MainEventBusProcessorExceptionTest.java index eb6c53d46..8bb212601 100644 --- a/server-common/src/test/java/io/a2a/server/events/MainEventBusProcessorExceptionTest.java +++ b/server-common/src/test/java/io/a2a/server/events/MainEventBusProcessorExceptionTest.java @@ -40,8 +40,7 @@ * and distributed to clients with appropriate logging based on failure type: *
      *
    • TaskSerializationException → ERROR log + InternalError
      • - *
    • TaskPersistenceException (transient) → WARN log + InternalError
      • - *
    • TaskPersistenceException (permanent) → ERROR log + InternalError
      • + *
    • TaskPersistenceException → ERROR log + InternalError
      • *
    */ public class MainEventBusProcessorExceptionTest { @@ -133,15 +132,15 @@ public void testTaskSerializationException_ConvertsToInternalError() throws Inte } /** - * Test that TaskPersistenceException (transient) is converted to InternalError with WARN log. - * AC#2: Mock TaskStore throws TaskPersistenceException(transient) → WARN log + InternalError + * Test that TaskPersistenceException is converted to InternalError with ERROR log. + * AC#2: Mock TaskStore throws TaskPersistenceException → ERROR log + InternalError */ @Test - public void testTaskPersistenceException_Transient_ConvertsToInternalError() throws InterruptedException { - // Arrange: Mock TaskStore to throw transient TaskPersistenceException - String exceptionMessage = "Connection timeout to database"; + public void testTaskPersistenceException_ConvertsToInternalError() throws InterruptedException { + // Arrange: Mock TaskStore to throw TaskPersistenceException + String exceptionMessage = "Database operation failed"; TaskPersistenceException exception = new TaskPersistenceException( - TASK_ID, exceptionMessage, true // transient = true + TASK_ID, exceptionMessage ); when(mockTaskStore.get(any())).thenThrow(exception); doThrow(exception).when(mockTaskStore).save(any(Task.class), anyBoolean()); @@ -160,54 +159,13 @@ public void testTaskPersistenceException_Transient_ConvertsToInternalError() thr InternalError error = (InternalError) distributedEvent; assertTrue(error.getMessage().contains(TASK_ID), "Error message should contain task ID: " + error.getMessage()); - assertTrue(error.getMessage().contains("retry may succeed"), - "Transient error message should mention retry: " + error.getMessage()); - // Assert: Verify WARN level logging for transient failures - boolean foundWarnLog = logAppender.list.stream() - .anyMatch(event -> event.getLevel() == Level.WARN - && event.getFormattedMessage().contains(TASK_ID) - && event.getFormattedMessage().contains("TRANSIENT")); - assertTrue(foundWarnLog, "Should log transient TaskPersistenceException at WARN level"); - } - - /** - * Test that TaskPersistenceException (permanent) is converted to InternalError with ERROR log. - * AC#3: Mock TaskStore throws TaskPersistenceException(permanent) → ERROR log + InternalError - */ - @Test - public void testTaskPersistenceException_Permanent_ConvertsToInternalError() throws InterruptedException { - // Arrange: Mock TaskStore to throw permanent TaskPersistenceException - String exceptionMessage = "Disk full - cannot write"; - TaskPersistenceException exception = new TaskPersistenceException( - TASK_ID, exceptionMessage, false // transient = false (permanent) - ); - when(mockTaskStore.get(any())).thenThrow(exception); - doThrow(exception).when(mockTaskStore).save(any(Task.class), anyBoolean()); - - Task testTask = createTestTask(); - - // Act: Enqueue event and wait for processing - List distributedEvents = captureDistributedEvent(testTask); - - // Assert: Verify InternalError was distributed - assertEquals(1, distributedEvents.size(), "Should distribute exactly one event"); - Event distributedEvent = distributedEvents.get(0); - assertInstanceOf(InternalError.class, distributedEvent, - "TaskPersistenceException should convert to InternalError"); - - InternalError error = (InternalError) distributedEvent; - assertTrue(error.getMessage().contains(TASK_ID), - "Error message should contain task ID: " + error.getMessage()); - assertTrue(error.getMessage().contains("requires intervention"), - "Permanent error message should mention intervention: " + error.getMessage()); - - // Assert: Verify ERROR level logging for permanent failures + // Assert: Verify ERROR level logging boolean foundErrorLog = logAppender.list.stream() .anyMatch(event -> event.getLevel() == Level.ERROR && event.getFormattedMessage().contains(TASK_ID) - && event.getFormattedMessage().contains("PERMANENT")); - assertTrue(foundErrorLog, "Should log permanent TaskPersistenceException at ERROR level"); + && event.getFormattedMessage().contains("persistence failed")); + assertTrue(foundErrorLog, "Should log TaskPersistenceException at ERROR level"); } /** diff --git a/server-common/src/test/java/io/a2a/server/tasks/TaskPersistenceExceptionTest.java b/server-common/src/test/java/io/a2a/server/tasks/TaskPersistenceExceptionTest.java index bf0fe5388..51ad86dc3 100644 --- a/server-common/src/test/java/io/a2a/server/tasks/TaskPersistenceExceptionTest.java +++ b/server-common/src/test/java/io/a2a/server/tasks/TaskPersistenceExceptionTest.java @@ -1,19 +1,16 @@ package io.a2a.server.tasks; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; -import static org.junit.jupiter.api.Assertions.assertTrue; import org.junit.jupiter.api.Test; /** * Unit tests for {@link TaskPersistenceException}. *

    - * Tests the exception class for database/storage system failures, - * with special focus on the {@code isTransient} flag for retry logic. + * Tests the exception class for database/storage system failures. */ class TaskPersistenceExceptionTest extends AbstractTaskStoreExceptionTest { @@ -27,60 +24,55 @@ protected TaskPersistenceException createException(String taskId, String message return new TaskPersistenceException(taskId, message, cause); } - // ========== Constructor Tests - Default Non-Transient ========== + // ========== Constructor Tests ========== @Test - void testConstructor_noArgs_defaultsToNonTransient() { + void testConstructor_noArgs() { TaskPersistenceException exception = new TaskPersistenceException(); assertNull(exception.getMessage()); assertNull(exception.getCause()); assertNull(exception.getTaskId()); - assertFalse(exception.isTransient(), "Default should be non-transient"); } @Test - void testConstructor_messageOnly_defaultsToNonTransient() { + void testConstructor_messageOnly() { TaskPersistenceException exception = new TaskPersistenceException("Database error"); assertEquals("Database error", exception.getMessage()); assertNull(exception.getCause()); assertNull(exception.getTaskId()); - assertFalse(exception.isTransient(), "Default should be non-transient"); } @Test - void testConstructor_causeOnly_defaultsToNonTransient() { + void testConstructor_causeOnly() { RuntimeException cause = new RuntimeException("Connection failed"); TaskPersistenceException exception = new TaskPersistenceException(cause); assertNotNull(exception.getMessage()); assertSame(cause, exception.getCause()); assertNull(exception.getTaskId()); - assertFalse(exception.isTransient(), "Default should be non-transient"); } @Test - void testConstructor_messageAndCause_defaultsToNonTransient() { + void testConstructor_messageAndCause() { RuntimeException cause = new RuntimeException("Timeout"); TaskPersistenceException exception = new TaskPersistenceException("Operation failed", cause); assertEquals("Operation failed", exception.getMessage()); assertSame(cause, exception.getCause()); assertNull(exception.getTaskId()); - assertFalse(exception.isTransient(), "Default should be non-transient"); } @Test - void testConstructor_taskIdAndMessage_defaultsToNonTransient() { + void testConstructor_taskIdAndMessage() { TaskPersistenceException exception = new TaskPersistenceException("task-123", "Save failed"); assertEquals("Save failed", exception.getMessage()); assertEquals("task-123", exception.getTaskId()); assertNull(exception.getCause()); - assertFalse(exception.isTransient(), "Default should be non-transient"); } @Test - void testConstructor_taskIdMessageAndCause_defaultsToNonTransient() { + void testConstructor_taskIdMessageAndCause() { RuntimeException cause = new RuntimeException("Disk error"); TaskPersistenceException exception = new TaskPersistenceException( "task-456", "Persistence failed", cause); @@ -88,91 +80,6 @@ void testConstructor_taskIdMessageAndCause_defaultsToNonTransient() { assertEquals("Persistence failed", exception.getMessage()); assertEquals("task-456", exception.getTaskId()); assertSame(cause, exception.getCause()); - assertFalse(exception.isTransient(), "Default should be non-transient"); - } - - // ========== Constructor Tests - Explicit Transient Flag ========== - - @Test - void testConstructor_messageWithTransientTrue() { - TaskPersistenceException exception = new TaskPersistenceException("Connection timeout", true); - - assertEquals("Connection timeout", exception.getMessage()); - assertTrue(exception.isTransient(), "Should be transient when explicitly set"); - } - - @Test - void testConstructor_messageWithTransientFalse() { - TaskPersistenceException exception = new TaskPersistenceException("Disk full", false); - - assertEquals("Disk full", exception.getMessage()); - assertFalse(exception.isTransient(), "Should be non-transient when explicitly set"); - } - - @Test - void testConstructor_messageAndCauseWithTransientTrue() { - RuntimeException cause = new RuntimeException("Network partition"); - TaskPersistenceException exception = new TaskPersistenceException( - "Database unreachable", cause, true); - - assertEquals("Database unreachable", exception.getMessage()); - assertSame(cause, exception.getCause()); - assertTrue(exception.isTransient(), "Should be transient"); - } - - @Test - void testConstructor_messageAndCauseWithTransientFalse() { - RuntimeException cause = new RuntimeException("Constraint violation"); - TaskPersistenceException exception = new TaskPersistenceException( - "Duplicate key error", cause, false); - - assertEquals("Duplicate key error", exception.getMessage()); - assertSame(cause, exception.getCause()); - assertFalse(exception.isTransient(), "Should be non-transient"); - } - - @Test - void testConstructor_taskIdMessageWithTransientTrue() { - TaskPersistenceException exception = new TaskPersistenceException( - "task-123", "Lock timeout", true); - - assertEquals("task-123", exception.getTaskId()); - assertEquals("Lock timeout", exception.getMessage()); - assertTrue(exception.isTransient(), "Should be transient"); - } - - @Test - void testConstructor_taskIdMessageWithTransientFalse() { - TaskPersistenceException exception = new TaskPersistenceException( - "task-456", "Storage quota exceeded", false); - - assertEquals("task-456", exception.getTaskId()); - assertEquals("Storage quota exceeded", exception.getMessage()); - assertFalse(exception.isTransient(), "Should be non-transient"); - } - - @Test - void testConstructor_fullArgsWithTransientTrue() { - RuntimeException cause = new RuntimeException("Deadlock detected"); - TaskPersistenceException exception = new TaskPersistenceException( - "task-789", "Transaction rolled back", cause, true); - - assertEquals("task-789", exception.getTaskId()); - assertEquals("Transaction rolled back", exception.getMessage()); - assertSame(cause, exception.getCause()); - assertTrue(exception.isTransient(), "Should be transient"); - } - - @Test - void testConstructor_fullArgsWithTransientFalse() { - RuntimeException cause = new RuntimeException("Permission denied"); - TaskPersistenceException exception = new TaskPersistenceException( - "task-abc", "Access control violation", cause, false); - - assertEquals("task-abc", exception.getTaskId()); - assertEquals("Access control violation", exception.getMessage()); - assertSame(cause, exception.getCause()); - assertFalse(exception.isTransient(), "Should be non-transient"); } // ========== Inheritance Verification ========== @@ -186,148 +93,120 @@ void testInheritance_extendsTaskStoreException() { @Test void testInheritance_taskIdFieldAccessible() { - TaskPersistenceException exception = new TaskPersistenceException("task-xyz", "Test message", true); + TaskPersistenceException exception = new TaskPersistenceException("task-xyz", "Test message"); // Should be accessible via TaskStoreException parent class assertEquals("task-xyz", exception.getTaskId()); } - // ========== Transient Flag Semantics Tests ========== + // ========== Real-World Scenario Tests ========== @Test - void testTransientFlag_indicatesRetryability() { - TaskPersistenceException transientError = new TaskPersistenceException( - "task-123", "Connection timeout", true); - TaskPersistenceException permanentError = new TaskPersistenceException( - "task-456", "Disk full", false); - - assertTrue(transientError.isTransient(), "Timeout errors should be retryable"); - assertFalse(permanentError.isTransient(), "Disk full is not retryable"); - } - - // ========== Real-World Scenario Tests - Transient Failures ========== - - @Test - void testScenario_transient_connectionTimeout() { + void testScenario_connectionTimeout() { RuntimeException timeout = new RuntimeException("Connection timeout after 30s"); TaskPersistenceException exception = new TaskPersistenceException( - "task-123", "Database connection timeout", timeout, true); + "task-123", "Database connection timeout", timeout); assertFullContext(exception, "task-123", timeout); - assertTrue(exception.isTransient(), "Connection timeouts are transient"); assertMessageContains(exception, "timeout"); } @Test - void testScenario_transient_deadlock() { + void testScenario_deadlock() { RuntimeException deadlock = new RuntimeException("Deadlock detected"); TaskPersistenceException exception = new TaskPersistenceException( - "task-456", "Transaction deadlock, retry recommended", deadlock, true); + "task-456", "Transaction deadlock", deadlock); - assertTrue(exception.isTransient(), "Deadlocks are transient"); assertMessageContains(exception, "deadlock"); - assertMessageContains(exception, "retry"); } @Test - void testScenario_transient_lockTimeout() { + void testScenario_lockTimeout() { RuntimeException lockTimeout = new RuntimeException("Lock wait timeout exceeded"); TaskPersistenceException exception = new TaskPersistenceException( - "task-789", "Failed to acquire row lock", lockTimeout, true); + "task-789", "Failed to acquire row lock", lockTimeout); - assertTrue(exception.isTransient(), "Lock timeouts are transient"); assertMessageContains(exception, "lock"); } @Test - void testScenario_transient_networkPartition() { + void testScenario_networkPartition() { RuntimeException networkError = new RuntimeException("Network unreachable"); TaskPersistenceException exception = new TaskPersistenceException( - "task-abc", "Database host unreachable due to network partition", networkError, true); + "task-abc", "Database host unreachable due to network partition", networkError); - assertTrue(exception.isTransient(), "Network errors are transient"); assertMessageContains(exception, "network"); } @Test - void testScenario_transient_poolExhausted() { + void testScenario_poolExhausted() { RuntimeException poolError = new RuntimeException("Connection pool exhausted"); TaskPersistenceException exception = new TaskPersistenceException( - "task-def", "No database connections available", poolError, true); + "task-def", "No database connections available", poolError); - assertTrue(exception.isTransient(), "Pool exhaustion is transient"); assertMessageContains(exception, "connections available"); } - // ========== Real-World Scenario Tests - Non-Transient Failures ========== - @Test - void testScenario_nonTransient_diskFull() { + void testScenario_diskFull() { RuntimeException diskError = new RuntimeException("No space left on device"); TaskPersistenceException exception = new TaskPersistenceException( - "task-123", "Cannot write task: disk full", diskError, false); + "task-123", "Cannot write task: disk full", diskError); - assertFalse(exception.isTransient(), "Disk full is not transient"); assertMessageContains(exception, "disk full"); } @Test - void testScenario_nonTransient_uniqueConstraint() { + void testScenario_uniqueConstraint() { RuntimeException constraintError = new RuntimeException("Duplicate entry for key 'PRIMARY'"); TaskPersistenceException exception = new TaskPersistenceException( - "task-456", "Task ID already exists", constraintError, false); + "task-456", "Task ID already exists", constraintError); - assertFalse(exception.isTransient(), "Unique constraint violations are not transient"); assertMessageContains(exception, "already exists"); } @Test - void testScenario_nonTransient_foreignKeyViolation() { + void testScenario_foreignKeyViolation() { RuntimeException fkError = new RuntimeException("Cannot add or update child row"); TaskPersistenceException exception = new TaskPersistenceException( - "task-789", "Foreign key constraint violation", fkError, false); + "task-789", "Foreign key constraint violation", fkError); - assertFalse(exception.isTransient(), "Foreign key violations are not transient"); assertMessageContains(exception, "constraint"); } @Test - void testScenario_nonTransient_permissionDenied() { + void testScenario_permissionDenied() { RuntimeException permError = new RuntimeException("Access denied for user 'app'"); TaskPersistenceException exception = new TaskPersistenceException( - "task-abc", "Insufficient database permissions", permError, false); + "task-abc", "Insufficient database permissions", permError); - assertFalse(exception.isTransient(), "Permission errors are not transient"); assertMessageContains(exception, "permission"); } @Test - void testScenario_nonTransient_schemaIncompatibility() { + void testScenario_schemaIncompatibility() { RuntimeException schemaError = new RuntimeException("Column 'new_field' does not exist"); TaskPersistenceException exception = new TaskPersistenceException( - "task-def", "Database schema incompatible with application version", schemaError, false); + "task-def", "Database schema incompatible with application version", schemaError); - assertFalse(exception.isTransient(), "Schema incompatibilities are not transient"); assertMessageContains(exception, "schema"); } @Test - void testScenario_nonTransient_quotaExceeded() { + void testScenario_quotaExceeded() { RuntimeException quotaError = new RuntimeException("Storage quota exceeded"); TaskPersistenceException exception = new TaskPersistenceException( - "task-ghi", "Database storage limit reached", quotaError, false); + "task-ghi", "Database storage limit reached", quotaError); - assertFalse(exception.isTransient(), "Quota violations are not transient"); assertMessageContains(exception, "storage limit"); } - // ========== Message Quality Tests - Persistence Context ========== + // ========== Message Quality Tests ========== @Test void testMessage_connectionError() { TaskPersistenceException exception = new TaskPersistenceException( "task-123", - "Failed to connect to database at jdbc:postgresql://localhost:5432/a2a", - true); + "Failed to connect to database at jdbc:postgresql://localhost:5432/a2a"); assertMessageContains(exception, "connect to database"); assertMessageContains(exception, "jdbc:postgresql"); @@ -337,58 +216,30 @@ void testMessage_connectionError() { void testMessage_transactionRollback() { TaskPersistenceException exception = new TaskPersistenceException( "task-123", - "Transaction rolled back due to deadlock. Retry operation.", - true); + "Transaction rolled back due to deadlock"); assertMessageContains(exception, "Transaction rolled back"); - assertMessageContains(exception, "Retry operation"); } @Test void testMessage_constraintViolation() { TaskPersistenceException exception = new TaskPersistenceException( "task-123", - "Unique constraint violation: task_id 'task-123' already exists in table 'tasks'", - false); + "Unique constraint violation: task_id 'task-123' already exists in table 'tasks'"); assertMessageContains(exception, "Unique constraint"); assertMessageContains(exception, "already exists"); } - // ========== Message Actionability Tests ========== + // ========== Cause Chain Tests ========== @Test - void testMessageActionable_transientWithRetryGuidance() { - TaskPersistenceException exception = new TaskPersistenceException( - "task-123", - "Connection timeout. Retry with exponential backoff (attempt 1 of 3).", - true); - - assertMessageContains(exception, "Retry with exponential backoff"); - assertMessageContains(exception, "attempt 1 of 3"); - } - - @Test - void testMessageActionable_nonTransientWithResolution() { - TaskPersistenceException exception = new TaskPersistenceException( - "task-123", - "Disk full. Free up space or increase storage capacity to continue.", - false); - - assertMessageContains(exception, "Free up space"); - assertMessageContains(exception, "increase storage"); - } - - // ========== Cause Chain for Retry Logic ========== - - @Test - void testCauseChain_transientErrorWithContext() { + void testCauseChain_withContext() { RuntimeException sqlError = new RuntimeException("SQL state: 08006 - Connection failure"); RuntimeException jdbcError = new RuntimeException("JDBC connection error", sqlError); TaskPersistenceException exception = new TaskPersistenceException( - "task-xyz", "Database operation failed", jdbcError, true); + "task-xyz", "Database operation failed", jdbcError); - assertTrue(exception.isTransient()); assertEquals("task-xyz", exception.getTaskId()); assertSame(jdbcError, exception.getCause()); assertSame(sqlError, exception.getCause().getCause()); @@ -397,23 +248,11 @@ void testCauseChain_transientErrorWithContext() { // ========== Null Safety ========== @Test - void testNullSafety_nullTaskIdWithTransient() { + void testNullSafety_nullTaskId() { TaskPersistenceException exception = new TaskPersistenceException( - null, "Generic database error", true); + null, "Generic database error"); assertNull(exception.getTaskId()); - assertTrue(exception.isTransient()); assertEquals("Generic database error", exception.getMessage()); } - - @Test - void testNullSafety_allNullsWithTransient() { - TaskPersistenceException exception = new TaskPersistenceException( - null, (String) null, (Throwable) null, true); - - assertNull(exception.getTaskId()); - assertNull(exception.getMessage()); - assertNull(exception.getCause()); - assertTrue(exception.isTransient()); - } } From 450baf3deeaf3ddef617d3e902318c45d33bd478 Mon Sep 17 00:00:00 2001 From: Kabir Khan Date: Tue, 24 Feb 2026 16:05:28 +0000 Subject: [PATCH 3/5] docs: Remove recovery strategy and transient/non-transient references from exception JavaDoc Simplified JavaDoc for both TaskPersistenceException and TaskSerializationException to remove all mentions of transient/non-transient failures and recovery strategies. Now just documents common scenarios and usage examples. Co-Authored-By: Claude Sonnet 4.5 --- .../server/events/MainEventBusProcessor.java | 7 +- .../a2a/server/tasks/InMemoryTaskStore.java | 3 +- .../server/tasks/PushNotificationSender.java | 1 - .../tasks/TaskPersistenceException.java | 6 +- .../tasks/TaskSerializationException.java | 10 -- .../java/io/a2a/server/tasks/TaskStore.java | 98 ++++--------------- .../tasks/TaskSerializationExceptionTest.java | 2 +- 7 files changed, 27 insertions(+), 100 deletions(-) diff --git a/server-common/src/main/java/io/a2a/server/events/MainEventBusProcessor.java b/server-common/src/main/java/io/a2a/server/events/MainEventBusProcessor.java index 9005765eb..d7eece30b 100644 --- a/server-common/src/main/java/io/a2a/server/events/MainEventBusProcessor.java +++ b/server-common/src/main/java/io/a2a/server/events/MainEventBusProcessor.java @@ -47,11 +47,10 @@ *

    Exception Handling

    * TaskStore persistence failures are caught and handled gracefully: *
      - *
    • {@link TaskSerializationException} - Data corruption or schema mismatch (permanent). + *
    • {@link TaskSerializationException} - Data corruption or schema mismatch. + * Logged at ERROR level, distributed as {@link InternalError} to clients.
      • + *
    • {@link TaskPersistenceException} - Database/storage system failure. * Logged at ERROR level, distributed as {@link InternalError} to clients.
      • - *
    • {@link TaskPersistenceException} - Database failure (transient or permanent). - * Transient failures logged at WARN level, permanent at ERROR level. - * InternalError message indicates retry viability.
      • *
    * *

    Processing continues after errors - the failed event is distributed as InternalError diff --git a/server-common/src/main/java/io/a2a/server/tasks/InMemoryTaskStore.java b/server-common/src/main/java/io/a2a/server/tasks/InMemoryTaskStore.java index de8774eb8..7c443b1cc 100644 --- a/server-common/src/main/java/io/a2a/server/tasks/InMemoryTaskStore.java +++ b/server-common/src/main/java/io/a2a/server/tasks/InMemoryTaskStore.java @@ -46,8 +46,7 @@ *

  • No serialization step exists - tasks stored as Java objects
    • *
  • No I/O or network operations that can fail
    • *
  • ConcurrentHashMap guarantees thread-safe operations without checked exceptions
    • - *
  • Memory exhaustion (OutOfMemoryError) is an unrecoverable system failure, not a transient - * storage error
    • + *
  • Memory exhaustion (OutOfMemoryError) is an unrecoverable system failure
    • * * *

    Comparison to Database Implementations

    diff --git a/server-common/src/main/java/io/a2a/server/tasks/PushNotificationSender.java b/server-common/src/main/java/io/a2a/server/tasks/PushNotificationSender.java index f8b7b018d..093241182 100644 --- a/server-common/src/main/java/io/a2a/server/tasks/PushNotificationSender.java +++ b/server-common/src/main/java/io/a2a/server/tasks/PushNotificationSender.java @@ -63,7 +63,6 @@ * Implementations should handle errors gracefully: *
      *
    • Log failures but don't throw exceptions (notifications are best-effort)
      • - *
    • Consider retry logic for transient failures
      • *
    • Don't block on network I/O - execute asynchronously if needed
      • *
    • Circuit breaker patterns for repeatedly failing endpoints
      • *
    diff --git a/server-common/src/main/java/io/a2a/server/tasks/TaskPersistenceException.java b/server-common/src/main/java/io/a2a/server/tasks/TaskPersistenceException.java index d5a218d77..97007cab9 100644 --- a/server-common/src/main/java/io/a2a/server/tasks/TaskPersistenceException.java +++ b/server-common/src/main/java/io/a2a/server/tasks/TaskPersistenceException.java @@ -7,9 +7,8 @@ *

    * Indicates failures in the underlying storage system (database, filesystem, etc.) rather * than data format issues. - *

    - *

    - * Examples of persistence failures: + * + *

    Common Scenarios

    *
      *
    • Database connection timeout or network partition
      • *
    • Transaction deadlock or lock wait timeout
      • @@ -19,7 +18,6 @@ *
    • Insufficient permissions or authentication failures
      • *
    • Database schema incompatibilities
      • *
    - *

    * *

    Usage Example

    * 
    {@code
diff --git a/server-common/src/main/java/io/a2a/server/tasks/TaskSerializationException.java b/server-common/src/main/java/io/a2a/server/tasks/TaskSerializationException.java
index ac960f91f..5916ce530 100644
--- a/server-common/src/main/java/io/a2a/server/tasks/TaskSerializationException.java
+++ b/server-common/src/main/java/io/a2a/server/tasks/TaskSerializationException.java
@@ -6,8 +6,6 @@
  * Exception for task serialization/deserialization failures.
  * 
    
  * Indicates failures converting between Task domain objects and persistent storage format (JSON).
- * These failures are typically non-transient - they indicate data corruption, schema
- * mismatches, or invalid field values that require manual intervention.
  *
  * 
    Common Scenarios
    
  * 
      
@@ -17,14 +15,6 @@
  *   
    • Data format version mismatches after upgrades
      • 
  * 
    
  *
- * 
    Recovery Strategy
    
- * Non-Transient: Retry will not help. Requires:
- * 
      
- *   
    • Data migration or repair for corrupted tasks
      • 
- *   
    • Schema updates for version mismatches
      • 
- *   
    • Manual intervention to fix invalid data
      • 
- * 
    
- *
  * 
    Usage Example
    
  * 
    {@code
  * try {
diff --git a/server-common/src/main/java/io/a2a/server/tasks/TaskStore.java b/server-common/src/main/java/io/a2a/server/tasks/TaskStore.java
index c0b7eb9ea..7ae7a46bf 100644
--- a/server-common/src/main/java/io/a2a/server/tasks/TaskStore.java
+++ b/server-common/src/main/java/io/a2a/server/tasks/TaskStore.java
@@ -92,11 +92,10 @@
  *
  * 
    Exception Contract
    
  * All TaskStore methods may throw {@link TaskStoreException} or its subclasses to indicate
- * persistence failures. Implementers must choose the appropriate exception type based on
- * the failure cause:
+ * persistence failures:
  * 
      
- *   
    • {@link TaskSerializationException} - JSON/data format errors (non-transient)
      • 
- *   
    • {@link TaskPersistenceException} - Database/storage failures (transient or non-transient)
      • 
+ *   
    • {@link TaskSerializationException} - JSON/data format errors
      • 
+ *   
    • {@link TaskPersistenceException} - Database/storage system failures
      • 
  * 
    
  *
  * 
    When to Throw TaskSerializationException
    
@@ -107,23 +106,16 @@
  *   
  • Invalid enum values or missing required fields
    • 
  *   
  • Schema version mismatches after upgrades
    • 
  * 
- * These failures are always non-transient - retry will not help. They require data
- * migration, schema updates, or manual intervention.
  *
  * 
    When to Throw TaskPersistenceException
    
  * Use when the storage system fails:
  * 
      
- *   
    • Database connection timeouts (transient)
      • 
- *   
    • Transaction deadlocks (transient)
      • 
- *   
    • Connection pool exhausted (transient)
      • 
- *   
    • Disk full / quota exceeded (non-transient)
      • 
- *   
    • Database constraint violations (non-transient)
      • 
- *   
    • Insufficient permissions (non-transient)
      • 
- * 
    
- * Set the {@code isTransient} flag appropriately:
- * 
      
- *   
    • {@code true} - Temporary failure, retry may succeed (network, timeout, deadlock)
      • 
- *   
    • {@code false} - Permanent failure, requires intervention (disk full, constraints)
      • 
+ *   
    • Database connection timeouts
      • 
+ *   
    • Transaction deadlocks
      • 
+ *   
    • Connection pool exhausted
      • 
+ *   
    • Disk full / quota exceeded
      • 
+ *   
    • Database constraint violations
      • 
+ *   
    • Insufficient permissions
      • 
  * 
    
  *
  * 
    Implementer Example
    
@@ -139,61 +131,14 @@
  *     try {
  *         entityManager.merge(toEntity(json));
  *     } catch (PersistenceException e) {
- *         boolean transient = isTransientDatabaseError(e);
- *         throw new TaskPersistenceException(task.id(), "Database save failed", e, transient);
- *     }
- * }
- *
- * @Override
- * public Task get(String taskId) {
- *     String json = database.retrieve(taskId);
- *     try {
- *         return objectMapper.readValue(json, Task.class);
- *     } catch (JsonProcessingException e) {
- *         throw new TaskSerializationException(taskId, "Failed to deserialize task", e);
+ *         throw new TaskPersistenceException(task.id(), "Database save failed", e);
  *     }
  * }
  * }
    
  *
- * 
    Caller Exception Handling
    
- * Callers should distinguish between transient and permanent failures:
- * 
    {@code
- * try {
- *     taskStore.save(task, false);
- * } catch (TaskSerializationException e) {
- *     // Non-transient: Log error, notify operations team
- *     logger.error("Task data corruption for {}: {}", e.getTaskId(), e.getMessage(), e);
- *     alerting.sendAlert("Task serialization failure", e);
- *     // DO NOT RETRY - requires manual data repair
- *
- * } catch (TaskPersistenceException e) {
- *     if (e.isTransient()) {
- *         // Transient: Retry with exponential backoff
- *         logger.warn("Transient persistence failure for {}: {}", e.getTaskId(), e.getMessage());
- *         retryWithBackoff(() -> taskStore.save(task, false));
- *     } else {
- *         // Non-transient: Log error, alert operations
- *         logger.error("Permanent persistence failure for {}: {}", e.getTaskId(), e.getMessage(), e);
- *         alerting.sendAlert("Database capacity/constraint issue", e);
- *         // DO NOT RETRY - requires manual intervention
- *     }
- * } catch (TaskStoreException e) {
- *     // Generic fallback - treat as non-transient
- *     logger.error("TaskStore failure for {}: {}", e.getTaskId(), e.getMessage(), e);
- *     // DO NOT RETRY by default
- * }
- * }
    
- *
- * 
    Current Exception Handling
    
- * {@link io.a2a.server.events.MainEventBusProcessor} currently catches all TaskStore
- * exceptions and wraps them in {@link io.a2a.spec.InternalError} events for client
- * distribution. Future enhancements may distinguish transient failures for retry logic.
- *
- * 
    Method-Specific Notes
    
- * 
      
- *   
    • {@code delete()} typically does not throw TaskSerializationException (no deserialization required)
      • 
- *   
    • {@code list()} may encounter serialization errors for any task in the result set
      • 
- * 
    
+ * 
    Exception Handling
    
+ * {@link io.a2a.server.events.MainEventBusProcessor} catches TaskStore exceptions and
+ * wraps them in {@link io.a2a.spec.InternalError} events for client distribution.
  *
  * @see TaskManager
  * @see TaskStateProvider
@@ -213,9 +158,8 @@ public interface TaskStore {
      *                     false if it originated locally. Used to prevent feedback loops
      *                     in replicated scenarios (e.g., don't fire TaskFinalizedEvent for replicated updates)
      * @throws TaskSerializationException if the task cannot be serialized to storage format (JSON parsing error,
-     *                                    invalid field values, schema mismatch). Non-transient - retry will not help.
-     * @throws TaskPersistenceException if the storage system fails (database timeout, connection error, disk full).
-     *                                  Check {@link TaskPersistenceException#isTransient()} to determine if retry is appropriate.
+     *                                    invalid field values, schema mismatch)
+     * @throws TaskPersistenceException if the storage system fails (database timeout, connection error, disk full)
      * @throws TaskStoreException for other persistence failures not covered by specific subclasses
      */
     void save(Task task, boolean isReplicated);
@@ -226,9 +170,9 @@ public interface TaskStore {
      * @param taskId the task identifier
      * @return the task if found, null otherwise
      * @throws TaskSerializationException if the persisted task data cannot be deserialized (corrupted JSON,
-     *                                    schema incompatibility). Non-transient - indicates data corruption.
+     *                                    schema incompatibility)
      * @throws TaskPersistenceException if the storage system fails during retrieval (database connection error,
-     *                                  query timeout). Check {@link TaskPersistenceException#isTransient()} for retry guidance.
+     *                                  query timeout)
      * @throws TaskStoreException for other retrieval failures not covered by specific subclasses
      */
     @Nullable Task get(String taskId);
@@ -238,8 +182,7 @@ public interface TaskStore {
      *
      * @param taskId the task identifier
      * @throws TaskPersistenceException if the storage system fails during deletion (database connection error,
-     *                                  transaction timeout, constraint violation). Check {@link TaskPersistenceException#isTransient()}
-     *                                  to determine if retry is appropriate.
+     *                                  transaction timeout, constraint violation)
      * @throws TaskStoreException for other deletion failures not covered by specific subclasses
      */
     void delete(String taskId);
@@ -250,10 +193,9 @@ public interface TaskStore {
      * @param params the filtering and pagination parameters
      * @return the list of tasks matching the criteria with pagination info
      * @throws TaskSerializationException if any persisted task data cannot be deserialized during listing
-     *                                    (corrupted JSON in database). Non-transient - indicates data corruption affecting
-     *                                    one or more tasks.
+     *                                    (corrupted JSON in database)
      * @throws TaskPersistenceException if the storage system fails during the list operation (database query timeout,
-     *                                  connection error). Check {@link TaskPersistenceException#isTransient()} for retry guidance.
+     *                                  connection error)
      * @throws TaskStoreException for other listing failures not covered by specific subclasses
      */
     ListTasksResult list(ListTasksParams params);
diff --git a/server-common/src/test/java/io/a2a/server/tasks/TaskSerializationExceptionTest.java b/server-common/src/test/java/io/a2a/server/tasks/TaskSerializationExceptionTest.java
index e843adb68..df356ef14 100644
--- a/server-common/src/test/java/io/a2a/server/tasks/TaskSerializationExceptionTest.java
+++ b/server-common/src/test/java/io/a2a/server/tasks/TaskSerializationExceptionTest.java
@@ -11,7 +11,7 @@
  * Unit tests for {@link TaskSerializationException}.
  * 
    
  * Tests the exception class for task serialization/deserialization failures,
- * verifying all constructor variants and non-transient failure semantics.
+ * verifying all constructor variants.
  */
 class TaskSerializationExceptionTest extends AbstractTaskStoreExceptionTest {
 

From 38ddca442b04bb55154543f38300c0dd76a23f2a Mon Sep 17 00:00:00 2001
From: Kabir Khan 
Date: Tue, 24 Feb 2026 16:23:45 +0000
Subject: [PATCH 4/5] Fix indentation

---
 .../database/jpa/JpaDatabaseTaskStore.java    | 144 +++++++++---------
 1 file changed, 72 insertions(+), 72 deletions(-)

diff --git a/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java b/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java
index f55d119df..a75378fe9 100644
--- a/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java
+++ b/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java
@@ -264,76 +264,76 @@ public ListTasksResult list(ListTasksParams params) {
             StringBuilder queryBuilder = new StringBuilder("SELECT t FROM JpaTask t WHERE 1=1");
             StringBuilder countQueryBuilder = new StringBuilder("SELECT COUNT(t) FROM JpaTask t WHERE 1=1");
 
-        // Apply contextId filter using denormalized column
-        if (params.contextId() != null) {
-            queryBuilder.append(" AND t.contextId = :contextId");
-            countQueryBuilder.append(" AND t.contextId = :contextId");
-        }
+            // Apply contextId filter using denormalized column
+            if (params.contextId() != null) {
+                queryBuilder.append(" AND t.contextId = :contextId");
+                countQueryBuilder.append(" AND t.contextId = :contextId");
+            }
 
-        // Apply status filter using denormalized column
-        if (params.status() != null) {
-            queryBuilder.append(" AND t.state = :state");
-            countQueryBuilder.append(" AND t.state = :state");
-        }
+            // Apply status filter using denormalized column
+            if (params.status() != null) {
+                queryBuilder.append(" AND t.state = :state");
+                countQueryBuilder.append(" AND t.state = :state");
+            }
 
-        // Apply statusTimestampAfter filter using denormalized timestamp column
-        if (params.statusTimestampAfter() != null) {
-            queryBuilder.append(" AND t.statusTimestamp > :statusTimestampAfter");
-            countQueryBuilder.append(" AND t.statusTimestamp > :statusTimestampAfter");
-        }
+            // Apply statusTimestampAfter filter using denormalized timestamp column
+            if (params.statusTimestampAfter() != null) {
+                queryBuilder.append(" AND t.statusTimestamp > :statusTimestampAfter");
+                countQueryBuilder.append(" AND t.statusTimestamp > :statusTimestampAfter");
+            }
 
-        // Apply pagination cursor using keyset pagination for composite sort (timestamp DESC, id ASC)
-        if (tokenTimestamp != null) {
-            // Keyset pagination: get tasks where timestamp < tokenTimestamp OR (timestamp = tokenTimestamp AND id > tokenId)
-            queryBuilder.append(" AND (t.statusTimestamp < :tokenTimestamp OR (t.statusTimestamp = :tokenTimestamp AND t.id > :tokenId))");
-        }
+            // Apply pagination cursor using keyset pagination for composite sort (timestamp DESC, id ASC)
+            if (tokenTimestamp != null) {
+                // Keyset pagination: get tasks where timestamp < tokenTimestamp OR (timestamp = tokenTimestamp AND id > tokenId)
+                queryBuilder.append(" AND (t.statusTimestamp < :tokenTimestamp OR (t.statusTimestamp = :tokenTimestamp AND t.id > :tokenId))");
+            }
 
-        // Sort by status timestamp descending (most recent first), then by ID for stable ordering
-        queryBuilder.append(" ORDER BY t.statusTimestamp DESC, t.id ASC");
+            // Sort by status timestamp descending (most recent first), then by ID for stable ordering
+            queryBuilder.append(" ORDER BY t.statusTimestamp DESC, t.id ASC");
 
-        // Create and configure the main query
-        TypedQuery query = em.createQuery(queryBuilder.toString(), JpaTask.class);
+            // Create and configure the main query
+            TypedQuery query = em.createQuery(queryBuilder.toString(), JpaTask.class);
 
-        // Set filter parameters
-        if (params.contextId() != null) {
-            query.setParameter("contextId", params.contextId());
-        }
-        if (params.status() != null) {
-            query.setParameter("state", params.status().name());
-        }
-        if (params.statusTimestampAfter() != null) {
-            query.setParameter("statusTimestampAfter", params.statusTimestampAfter());
-        }
-        if (tokenTimestamp != null) {
-            query.setParameter("tokenTimestamp", tokenTimestamp);
-            query.setParameter("tokenId", tokenId);
-        }
+            // Set filter parameters
+            if (params.contextId() != null) {
+                query.setParameter("contextId", params.contextId());
+            }
+            if (params.status() != null) {
+                query.setParameter("state", params.status().name());
+            }
+            if (params.statusTimestampAfter() != null) {
+                query.setParameter("statusTimestampAfter", params.statusTimestampAfter());
+            }
+            if (tokenTimestamp != null) {
+                query.setParameter("tokenTimestamp", tokenTimestamp);
+                query.setParameter("tokenId", tokenId);
+            }
 
-        // Apply page size limit (+1 to check for next page)
-        int pageSize = params.getEffectivePageSize();
-        query.setMaxResults(pageSize + 1);
+            // Apply page size limit (+1 to check for next page)
+            int pageSize = params.getEffectivePageSize();
+            query.setMaxResults(pageSize + 1);
 
-        // Execute query and deserialize tasks
-        List jpaTasksPage = query.getResultList();
+            // Execute query and deserialize tasks
+            List jpaTasksPage = query.getResultList();
 
-        // Determine if there are more results
-        boolean hasMore = jpaTasksPage.size() > pageSize;
-        if (hasMore) {
-            jpaTasksPage = jpaTasksPage.subList(0, pageSize);
-        }
+            // Determine if there are more results
+            boolean hasMore = jpaTasksPage.size() > pageSize;
+            if (hasMore) {
+                jpaTasksPage = jpaTasksPage.subList(0, pageSize);
+            }
 
-        // Get total count of matching tasks
-        TypedQuery countQuery = em.createQuery(countQueryBuilder.toString(), Long.class);
-        if (params.contextId() != null) {
-            countQuery.setParameter("contextId", params.contextId());
-        }
-        if (params.status() != null) {
-            countQuery.setParameter("state", params.status().name());
-        }
-        if (params.statusTimestampAfter() != null) {
-            countQuery.setParameter("statusTimestampAfter", params.statusTimestampAfter());
-        }
-        int totalSize = countQuery.getSingleResult().intValue();
+            // Get total count of matching tasks
+            TypedQuery countQuery = em.createQuery(countQueryBuilder.toString(), Long.class);
+            if (params.contextId() != null) {
+                countQuery.setParameter("contextId", params.contextId());
+            }
+            if (params.status() != null) {
+                countQuery.setParameter("state", params.status().name());
+            }
+            if (params.statusTimestampAfter() != null) {
+                countQuery.setParameter("statusTimestampAfter", params.statusTimestampAfter());
+            }
+            int totalSize = countQuery.getSingleResult().intValue();
 
             // Deserialize tasks from JSON
             List tasks = new ArrayList<>();
@@ -347,19 +347,19 @@ public ListTasksResult list(ListTasksParams params) {
                 }
             }
 
-        // Determine next page token (timestamp:ID of last task if there are more results)
-        // Format: "timestamp_millis:taskId" for keyset pagination
-        String nextPageToken = null;
-        if (hasMore && !tasks.isEmpty()) {
-            Task lastTask = tasks.get(tasks.size() - 1);
-            // All tasks have timestamps (TaskStatus canonical constructor ensures this)
-            Instant timestamp = lastTask.status().timestamp().toInstant();
-            nextPageToken = new PageToken(timestamp, lastTask.id()).toString();
-        }
+            // Determine next page token (timestamp:ID of last task if there are more results)
+            // Format: "timestamp_millis:taskId" for keyset pagination
+            String nextPageToken = null;
+            if (hasMore && !tasks.isEmpty()) {
+                Task lastTask = tasks.get(tasks.size() - 1);
+                // All tasks have timestamps (TaskStatus canonical constructor ensures this)
+                Instant timestamp = lastTask.status().timestamp().toInstant();
+                nextPageToken = new PageToken(timestamp, lastTask.id()).toString();
+            }
 
-        // Apply post-processing transformations (history limiting, artifact removal)
-        int historyLength = params.getEffectiveHistoryLength();
-        boolean includeArtifacts = params.shouldIncludeArtifacts();
+            // Apply post-processing transformations (history limiting, artifact removal)
+            int historyLength = params.getEffectiveHistoryLength();
+            boolean includeArtifacts = params.shouldIncludeArtifacts();
 
             List transformedTasks = tasks.stream()
                     .map(task -> transformTask(task, historyLength, includeArtifacts))

From 47c8c8d057a6c95727579cf8b8972991b2ab919a Mon Sep 17 00:00:00 2001
From: Kabir Khan 
Date: Tue, 24 Feb 2026 16:24:31 +0000
Subject: [PATCH 5/5] Remove unnecessary rethrow

---
 .../extras/taskstore/database/jpa/JpaDatabaseTaskStore.java    | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java b/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java
index a75378fe9..3a704ee2a 100644
--- a/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java
+++ b/extras/task-store-database-jpa/src/main/java/io/a2a/extras/taskstore/database/jpa/JpaDatabaseTaskStore.java
@@ -368,9 +368,6 @@ public ListTasksResult list(ListTasksParams params) {
             LOGGER.debug("Returning {} tasks out of {} total", transformedTasks.size(), totalSize);
             return new ListTasksResult(transformedTasks, totalSize, transformedTasks.size(), nextPageToken);
 
-        } catch (TaskSerializationException e) {
-            // Re-throw TaskSerializationException from inner loop
-            throw e;
         } catch (PersistenceException e) {
             // Database errors from query creation, execution, or count
             LOGGER.error("Database query failed during list operation", e);