[STIP-33][Feature][Zeta] Define task exception and failure history contract

### Search before asking

- [x] I had searched in the [feature](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22Feature%22) and found no similar feature requirement.

### Description

## Summary

This STIP proposes a bounded and structured task failure history for Zeta Engine.

Today, SeaTunnel retains and exposes a single formatted error message for a job. This is insufficient when a pipeline is restored multiple times or when different task groups fail during the same job.

The proposal introduces a job-scoped failure history that:

- groups failures by pipeline execution attempt
- identifies the failing pipeline and task group
- preserves worker and task information when available
- records timestamps, exception type, message, and stack trace
- remains available for running and finished jobs
- survives active master changes
- expires through the existing finished-job history policy
- preserves the existing `errorMsg` contract for compatibility

## Current behavior

`PhysicalPlan` currently retains only the first error reported by a sub-plan.

`TaskExecutionState` transports a formatted throwable message, but the engine does not preserve structured failure records across retries or restores.

`JobHistoryService` retains the final job error for finished jobs, but not the sequence of failures that led to the terminal state.

Pipeline and task-group state timestamps already exist internally, but they are not exposed as part of a failure-history read path.

## Goals

1. Retain multiple failures for one job using a bounded history.
2. Group failures by pipeline execution attempt.
3. Preserve task-group and worker attribution where available.
4. Support running and finished jobs through the same read contract.
5. Preserve history across active master changes.
6. Define deterministic retention, ordering, and deduplication behavior.
7. Keep existing job-detail clients backward compatible.
8. Provide a stable backend contract before implementing the Web UI.

## Non-goals

The first version will not provide:

- automatic root-cause classification
- cross-job failure analytics
- indefinite stack-trace retention
- distributed log aggregation
- task-level attribution when only task-group information is available
- changes to checkpoint or savepoint payloads

## Attempt model

An attempt belongs to a pipeline rather than an individual task.

- Initial pipeline execution uses attempt `0`.
- The attempt number increments before pipeline restore is scheduled.
- Failures from the same pipeline execution carry the same attempt number.
- Diagnostic attempt state must survive active master changes.
- A restored pipeline receives a new attempt even when an individual task index is reused.

`SubPlan.pipelineRestoreNum` currently participates in the `job.retry.times` decision. Making it durable and reusing it for history would also make the retry budget survive an active-master failover, which is a separate behavior change. The first implementation must keep the existing retry counter and retry-limit behavior unchanged.

Failure history stores a separate durable diagnostic attempt identity in the job-scoped history state. The initial identity is created as attempt `0`. Before a restore starts a new execution, the history entry atomically advances that pipeline's diagnostic attempt and records its start time. A new active master reads this identity before recording another failure. This counter is used only for failure correlation and REST output; it must not participate in restore eligibility or retry-limit checks.

This keeps the diagnostic model aligned with the existing pipeline restore boundary without changing retry semantics.

## Failure record

A failure record contains:

- sequence
- timestamp
- job ID
- pipeline ID
- pipeline attempt
- attempt start time
- task-group ID
- task ID when available
- task name when available
- worker address when available
- exception type when transported structurally
- concise exception message
- message truncation flag
- stack trace
- stack-trace truncation flag

The required fields are:

- sequence
- timestamp
- job ID
- pipeline ID
- attempt
- task-group ID

`attemptStartedAt` comes from the durable diagnostic attempt metadata. Attempt `0` is initialized when the pipeline execution is created for deployment, and a restored attempt is initialized when its diagnostic identity advances before restore scheduling. The field is optional for legacy or synthetic paths that cannot resolve this metadata. It has the same value for every record with the same pipeline ID and attempt and is distinct from the per-record failure timestamp.

Task ID, task name, worker, exception type, message, and stack trace remain optional because not every existing failure path provides them. The two truncation flags are required booleans.

The implementation must not derive exception type or task identity by parsing formatted stack traces.

The stored UTF-8 representation is limited to 4 KiB for the message and 64 KiB for the stack trace. Truncation must preserve valid UTF-8. A truncated message keeps its prefix. A truncated stack trace keeps both its beginning and end so the exception and deepest cause remain available.

## Capture and deduplication

`TaskExecutionState` remains the structured worker-to-master failure transport, but it is not the only way a task group can fail. The common capture point for terminal worker-reported failures is the `PhysicalVertex` state transition after `updateStateByExecutionService` accepts a `FAILED` state. This covers both normal worker reports and node-loss state updates that are routed directly to the physical vertex.

Deployment failures do not carry a `TaskExecutionState`; they enter through `makeTaskGroupFailing`. That path must create a failure record from the deployment exception and the known pipeline, task-group, slot, and worker metadata. `TaskDeployState.failed(Throwable)` must extract the original failure's class name, message, and bounded stack trace as strings while the `Throwable` is still available. For `deployOnRemote`, these strings are captured on the worker before the response crosses the Hazelcast RPC boundary; the live `Throwable` must not be included because its connector-specific class may not be available to the master. The resulting record reads those structured fields directly, so its `exceptionType` identifies the original cause rather than `TaskGroupDeployException`. The same deduplication key prevents a later terminal delivery for that attempt from creating another record. Cancellation without a failure cause is not recorded as an exception.

Repeated delivery of the same terminal state must not create duplicate records while the original record is retained. The first implementation will deduplicate atomically using:

`pipelineId + attempt + taskGroupId`

The first implementation uses a Hazelcast `EntryProcessor` on the job-scoped HA entry to perform the deduplication check, sequence allocation, append, and oldest-record eviction atomically. It must be submitted asynchronously from the task-status operation path. Completion handling may log a store failure, but must not wait on or re-enter a Hazelcast operation thread.

The first terminal delivery creates the record and receives a sequence number. Later deliveries with the same key are ignored without consuming another sequence number. A different task group in the same attempt remains separate. A task group that fails again after restore has a new attempt and therefore produces a separate record.

Deduplication uses the retained records as its bounded key set. After a record is evicted by either the 100-record limit or the 1 MiB aggregate-text limit, a delayed duplicate for that key can be recorded again. The first version does not keep a separate unbounded set of every key seen during the job lifetime.

Failure-history persistence is diagnostic and best effort. A history-store failure must not block the original failure or restore processing.

## Storage and retention

Failure history should use a dedicated HA-backed engine state entry keyed by job ID. The first implementation can use a dedicated Hazelcast `IMap` with the same default Hazelcast `MapConfig` baseline as the existing job-state maps. It does not add backups, persistence, or an external history backend. The public REST representation remains independent of that storage choice.

Initial retention behavior:

- retain at most 100 records per job
- retain at most 1 MiB of combined UTF-8 message and stack-trace content per job
- evict the oldest records until both limits are satisfied
- do not expire records while a job is active
- give the dedicated finished-history entry its own `history-job-expire-minutes` TTL

The initial maximum is a fixed implementation bound. A configurable limit can be considered later based on operational evidence.

## Large exception handling

Exception payloads must be bounded independently from the number of records.

The first version defines:

- a 4 KiB UTF-8 maximum for the message
- a 64 KiB UTF-8 maximum for the stored stack trace
- explicit truncation metadata
- truncation at a valid UTF-8 boundary
- preservation of the beginning and end of a truncated stack trace
- a 1 MiB aggregate retained message and stack-trace budget per job

The REST response should expose whether a field was truncated.

The first version will not paginate an individual stack trace. Future versions may add a separate detail endpoint without changing the failure summary contract.

## Finished-job history

The public REST contract must not depend directly on Hazelcast `IMap`.

Running-job history uses a dedicated HA engine state entry. When a job reaches a terminal state, `JobHistoryService` writes the retained records and diagnostic attempt metadata to a dedicated finished-history entry. That entry receives the same `history-job-expire-minutes` TTL as the corresponding finished-job record, so expiration does not depend on a cleanup listener. A listener may still remove it eagerly. This reuses the existing lifecycle without introducing a pluggable history-store abstraction.

Support for broader external history backends requires a separate storage design and is outside the first implementation.

The terminal history write is best effort. Write or cleanup failures must be logged and must not change the job terminal state.

Active master failover preserves records already acknowledged by the HA history store and resumes sequence and attempt numbering from that persisted state. Because history submission is asynchronous and best effort, a submission still in flight when the active master fails may be lost. The diagnostic path does not delay failure or restore processing to wait for history acknowledgement.

## REST contract

Proposed endpoint:

`GET /job-info/{jobId}/failures?limit=100`

Behavior:

- return records in descending sequence order
- default `limit` to 100
- reject non-positive limits
- cap the requested limit at the retained maximum
- return an empty list for a known job with no recorded failures
- return a controlled `404` for an unknown or expired job
- use the same response model for running and finished jobs

`JobInfoServlet` currently treats all path information after `/job-info/` as one numeric job ID. The REST implementation must extend that routing, or add an equivalent dedicated handler, so `/job-info/{jobId}` keeps its current behavior while `/job-info/{jobId}/failures` is routed to failure history. Routing must match only these exact path shapes. Additional segments, prefixes, or substring matches must use the existing not-found behavior.

The current `/job-info/{jobId}` behavior and its `errorMsg` field remain unchanged, including its existing response for an unknown job. The new failure-history endpoint defines its own explicit `404` response so callers can distinguish an unknown job from a known job with no failures.

## Security and input validation

The endpoint uses the same `BasicAuthFilter` boundary as the existing engine REST API. It does not introduce endpoint-specific authentication. Documentation must explain that exception text and worker addresses can contain operationally sensitive information when REST authentication is disabled.

Messages and stack traces are sanitized and bounded before HA or finished-history persistence. The implementation should extract the redaction patterns from `DryRunConnectFailureMessageSanitizer` into a shared utility rather than storing raw connector exception content. Failure history keeps its own 4 KiB message and 64 KiB stack-trace limits and truncation flags; it does not inherit the dry-run utility's 2 KiB display limit.

The route handler validates `jobId` and `limit`. Malformed identifiers and non-numeric or non-positive limits return a controlled `400` without a stack trace or reflected input. Values above the retained maximum are capped. Worker addresses remain optional and are limited to the same metadata already exposed by `/pending-jobs`: the `address` value in `host:port` form.

## Web UI

The Web UI will be implemented separately after the backend contract is stable.

The Exception view should:

- group failures by pipeline attempt
- display timestamp, pipeline, task group, worker, exception type, and message
- collapse stack traces by default
- indicate truncated fields
- link to task logs when a reliable log link is available
- display missing optional fields as unavailable
- avoid claiming task-level precision for task-group-level failures

## Compatibility

This proposal is additive:

- existing jobs require no configuration changes
- existing REST fields remain unchanged
- the existing final `errorMsg` remains available
- checkpoint and savepoint formats remain unchanged
- existing `job.retry.times` and active-master failover behavior remain unchanged
- old failure paths may populate only the fields available to them

`TaskExecutionState` is Java-serialized between workers and the master. Before adding optional failure fields, the implementation must capture and explicitly declare the serial UID generated for the current class. Keeping that UID preserves the existing wire form.

## Implementation slices

1. Add the internal failure record and dedicated HA-backed running and finished history entries.
2. Add diagnostic-attempt persistence and structured task failure transport.
3. Add capture, deduplication, retention, truncation, restore, and failover tests.
4. Add finished-job lifecycle and expiration integration.
5. Add the REST routing and endpoint with backward-compatibility and running/finished job API tests.
6. Add the Web UI history view separately.

## Acceptance criteria

1. A first-attempt failure is recorded with attempt `0`.
2. A failure after restore is recorded under the incremented attempt.
3. Duplicate terminal-state delivery does not create duplicate records while the original record remains in the bounded history; a delayed duplicate may be recorded after eviction.
4. Different task-group failures remain separate.
5. Active master failover preserves records acknowledged by the HA history store and the next attempt number; an asynchronous submission still in flight at failover may be lost.
6. Running and finished jobs expose the same response model.
7. More than 100 records, or more than 1 MiB of retained UTF-8 message and stack-trace content, evicts the oldest records deterministically until both limits are satisfied.
8. Large exception fields are bounded and marked as truncated.
9. A terminal job writes one bounded failure-history entry that expires with its corresponding finished-job record.
10. History-store failures do not alter job failure or restore behavior.
11. Existing clients continue receiving the current `errorMsg`.
12. Concurrent duplicate deliveries create one record and allocate one sequence through the atomic job-entry update.
13. Secrets in messages and stack traces are redacted before HA and finished-history persistence.
14. Malformed `jobId` or `limit` input returns a controlled `400` without exposing a stack trace or reflecting the invalid value.
15. The endpoint uses the same configured REST authentication boundary as existing job-detail endpoints.
16. Failure-history updates are submitted asynchronously and do not block a Hazelcast operation thread.
17. Failure-history state uses the same default Hazelcast map configuration as existing job-state maps and does not add backups or persistence.
18. Only exact `/job-info/{jobId}` and `/job-info/{jobId}/failures` paths are accepted; additional path segments use existing not-found behavior.
19. Existing serialized `TaskExecutionState` values remain readable after optional structured failure fields are added.
20. A deployment failure that enters through `makeTaskGroupFailing` creates one bounded record even though no `TaskExecutionState` exists.
21. Persisting diagnostic attempt identity does not change `job.retry.times`, restore eligibility, or retry behavior after active-master failover.
22. Every record that exposes `attemptStartedAt` reads it from the durable metadata created for that pipeline attempt.
23. English and Chinese documentation describe the final contract.

## Usage Scenario

This feature is intended for operators diagnosing Zeta jobs that fail or restore more than once.

It should help answer:

- Which pipeline attempt failed?
- Which task group and worker reported the failure?
- Is the same root cause recurring across restores?
- How long had the attempt been running before it failed?
- Which failures occurred before the final job error?
- Is the full stack trace available or was it truncated?
- Can the same history be inspected after the job finishes?

## Related issues

Feature umbrella: #11667

Design pull request: #11734

Related work:

- #9105
- #9050
- #10039
- #11351
- #11662

## Are you willing to submit a PR?

- [x] Yes I am willing to submit a PR!

## Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)