goutamadwant opened a new issue, #12268: URL: https://github.com/apache/seatunnel/issues/12268
# Bounded job-level lineage REST API for Zeta Status: **Under Review**. Proposed STIP-39; not accepted or implemented. Mailing-list discussion is pending. Author: Goutam Adwant. Feature request: [#11484](https://github.com/apache/seatunnel/issues/11484). ## Background Zeta already exposes DAG information through its job-detail API. The request in #11484 is for a supported REST lineage contract that a downstream operations tool can consume without depending on the UI response. The initial scope is the Source/Transform/Sink graph for one batch or streaming Zeta job, not column lineage. An earlier [design-only PR #11941](https://github.com/apache/seatunnel/pull/11941) was closed without merging, with direction to put the proposal on an issue and follow the STIP/dev-list process. This proposal reuses that research, corrects assumptions against current source, and reconciles the newer [OpenLineage proposal #12207](https://github.com/apache/seatunnel/issues/12207). Neither proposal is treated as accepted implementation. ## Motivation An operator investigating an unexpected destination should be able to ask: “Which source and transform stages feed this sink in this job?” A platform integration should not need raw job configuration, connector credentials, or sampled row traces to answer that topology question. Before this change, the consumer must interpret the existing job-detail representation. After an accepted implementation, it could request a bounded, documented graph with explicit missing-metadata and retention behavior. This is execution-topology visibility, not proof of which individual records or columns reached a table. ## Goals - Expose a deterministic, read-only projection of an available `JobDAGInfo` for one Zeta job. - Support batch and streaming jobs through the same response schema. - Represent pending/running jobs when a usable DAG exists, and finished jobs while the required history remains available. - Preserve node and pipeline connectivity without inventing table-to-table mappings. - Bound projection work, response bytes and concurrent request resource use. - Keep the existing job-detail API, checkpoint formats, connector contracts and job execution behavior unchanged. ## Non-Goals - Column lineage, SQL/transform semantic analysis, or record-level provenance. - A cross-job dataset catalog, OpenLineage event delivery, or an external receiver dependency. - Flink/Spark support, a UI screen, or a new persistent lineage store. - Live dynamic-table discovery, DDL history, or a promise that reported table paths are exhaustive. - Per-job access control, runtime metrics, or OpenLineage run/attempt identifiers. ## Design ### 1. Source of truth and ownership Use `JobDAGInfo` as the input to a pure, bounded mapper. Whitelist topology fields from `VertexInfo` and pipeline edges; never serialize `JobDAGInfo` wholesale. Its existing JSON conversion includes environment options, which are not part of this API. Current `JobMaster.getJobDAGInfo()` lazily builds and caches DAG information. `CoordinatorService.getJobInfo()` can also recover it from retained history, an active/pending master, or reconstruction from job information. Therefore, “reuse the existing lookup” is not equivalent to “no reconstruction or connector class loading.” The implementation must distinguish these paths and bound their costs before the endpoint is considered ready. The REST path must not independently rebuild a plan on every request, invoke connectors to enrich metadata, scan StainTrace, or run graph projection/serialization on a scheduling or checkpoint thread. Reuse an available cached snapshot; if recovery cannot provide one within the agreed resource/deadline contract, return temporary unavailability. Any required narrow lookup helper must leave the existing job-detail behavior unchanged. ### Proposed request sequence This shows the proposed responsibility boundaries, not an endpoint already implemented. The lookup must reconcile current execution ownership with retained history; it must not simply accept an old historical record while the same job ID is active again. ```mermaid sequenceDiagram autonumber participant Client as REST consumer participant Auth as Existing HTTP auth boundary participant API as Proposed lineage endpoint participant Lookup as Bounded DAG lookup participant State as Existing master and retained state participant Mapper as Bounded graph projection Client->>Auth: GET /job-lineage/{jobId} alt Authentication enabled and credentials rejected Auth-->>Client: Existing 401 behavior else Admitted under existing HTTP configuration Auth->>API: Forward request API->>API: Validate job ID and acquire capacity alt Invalid ID or no capacity API-->>Client: Fixed 400 or 503 error else Capacity acquired API->>Lookup: Resolve DAG within lookup deadline Lookup->>State: Read available DAG and reconcile lifecycle State-->>Lookup: Snapshot, absence, or unavailable state Lookup-->>API: Classified lookup result alt No usable consistent snapshot API-->>Client: Fixed 404, 409, or 503 error else Usable snapshot API->>Mapper: Validate and project whitelisted fields Mapper->>Mapper: Sort and serialize into capped buffer Mapper-->>API: Complete bytes or classified failure API-->>Client: 200 graph or fixed application error end API->>API: Release request resources end end ``` Deadline expiry releases the HTTP request, not necessarily underlying work. Any still-running lookup retains its work-capacity accounting until completion/cancellation; otherwise retries could create unbounded background work. Every application exit releases its owned buffers/permits, including failures and client disconnects. This feature introduces no call to an OpenLineage receiver and no checkpoint/lifecycle event delivery. ### 2. Proposed REST response Proposed route: `GET /job-lineage/{jobId}`, under the existing Zeta REST base path. Exact route and error status choices remain review decisions. Illustrative response, not a runtime capture: ```json { "schemaVersion": 1, "jobId": "733584788375093248", "graphKind": "EXECUTION", "idScope": "JOB", "nodes": [ {"id": "1", "kind": "SOURCE", "name": "pipeline-1 [source]", "tablePaths": ["sales.orders"], "datasetMetadata": "REPORTED"}, {"id": "2", "kind": "TRANSFORM", "name": "pipeline-1 [transform]", "tablePaths": [], "datasetMetadata": "NOT_APPLICABLE"}, {"id": "3", "kind": "SINK", "name": "pipeline-1 [sink]", "tablePaths": ["warehouse.orders"], "datasetMetadata": "REPORTED"} ], "edges": [ {"pipelineId": 1, "sourceNodeId": "1", "targetNodeId": "2"}, {"pipelineId": 1, "sourceNodeId": "2", "targetNodeId": "3"} ], "warnings": [] } ``` - `jobId` and node IDs are decimal strings to avoid JSON integer precision loss. `pipelineId` is the existing integer pipeline identifier. - `graphKind=EXECUTION` describes the execution-oriented representation built by the current job-info path; it does not promise one node per original configuration block. Unknown legacy representation must not silently be relabeled as this contract. - `name` is a display label from the vertex metadata. Despite the Java field name `connectorType`, its producer uses the action name, which can include pipeline prefixes and transform-chain labels. It is not a canonical connector identifier; do not parse it to infer identity. - `tablePaths` are distinct, sorted reported path strings, not globally unique dataset IDs or OpenLineage namespace/name pairs. Do not infer catalog/server identity or parse connector configuration to complete them. - `datasetMetadata=REPORTED` means at least one usable non-default path was reported, not that every dynamic table is known. `UNAVAILABLE` means no usable paths. Transforms use `NOT_APPLICABLE`. - Ignore null/default table-path entries. For an unavailable source/sink or when entries are omitted, emit `DATASET_METADATA_UNAVAILABLE`, even if other valid paths remain. A single warning per affected node is sufficient. Do not turn missing metadata into missing topology. - Sort nodes numerically by ID; edges by pipeline ID, source ID and target ID; warnings by code then numeric node ID; paths by case-sensitive string order. Do not lowercase names or paths. - Multi-table sources/sinks remain one execution node with multiple reported paths. Transform chains remain as represented in the DAG. Do not manufacture a source-table-to-sink-table cross-product. The mapper rejects missing required topology, mismatched map key/vertex ID, duplicate vertex identity, null edge endpoints, dangling references and inconsistent snapshots. Duplicate identical edges within a pipeline are deduplicated; distinct pipeline membership must be preserved. Cycles and self-loops are rejected as inconsistent topology, using bounded, non-recursive validation. A missing/null/unknown node kind is invalid; V1 supports only `SOURCE`, `TRANSFORM` and `SINK`. Empty graph metadata is unavailable rather than a successful empty lineage result. Warning shape, for a source with no usable reported path: ```json {"code": "DATASET_METADATA_UNAVAILABLE", "nodeId": "1"} ``` The response field names and enum values above are proposal-level contracts, not existing Java API additions. The implementation should keep its response DTOs and mapper internal to the engine REST implementation, rather than introducing a new connector-facing SPI. ### 3. Identity, restore and history The proposed identity is `(jobId, nodeId)`, scoped to the available job DAG; it is not an execution-attempt or OpenLineage run identity. Deterministic serialization of a supplied snapshot is straightforward, but stability after reconstructing that snapshot must be tested independently. Before promising job-lifetime identity, test repeated DAG construction, transform chaining, pipeline partitioning, pipeline retry, active-master failover, same-ID savepoint resume and new-ID restore. If same-ID reconstruction can change the identity/topology, resolve the contract explicitly: either prove a bounded correction using existing state, or revise the proposal to a graph-revision identity. Do not add persistent state or silently weaken the contract within an implementation PR. Separately submitted jobs with new IDs have independent graphs. V1 adds no restore edges across jobs and does not correlate node IDs across jobs. The current coordinator lookup checks retained DAG history before active/pending masters. The implementation must test coexistence of retained state and a same-ID resumed job; it must not serve an earlier retained graph as the current execution without validating that choice. This is an integration risk to resolve, not a newly reproduced bug claim. Finished state and DAG records are separate retained entries, not an atomic snapshot. Return a graph only while sufficient consistent metadata exists. Do not claim exact simultaneous expiry or attempt to recover expired history by replaying configuration. No additional retention setting is proposed. ### 4. Resource limits and failure behavior Proposed initial review values, not benchmark-backed defaults: - 10,000 nodes, 50,000 input edges, and 50,000 reported path entries across the graph, counted before deduplication/copying. - 4 KiB UTF-8 per display label or reported path; reject an over-limit graph rather than silently truncating identifiers. - 8 MiB total uncompressed UTF-8 JSON response, including JSON escaping and warnings. - At most four concurrent lineage projections/serializations per REST-serving process, with immediate busy rejection rather than an unbounded queue. Bound distributed lookup work too; a timed-out request must not leave unlimited background work. - A proposed five-second lookup deadline. A timeout is not proof that an underlying rebuild was cancelled; admission and lifecycle handling must account for work still in flight. Use bounded traversal and immutable request-owned output. Preflight count/string limits before constructing sorted copies; serialize once to a capped output buffer and commit the HTTP response only after success. Do not mutate shared `JobDAGInfo`, cache a stale graph indefinitely, or hold a scheduling lock while sorting/writing a response. These caps bound new endpoint work, not the memory already occupied by an engine DAG. Sorting means the projection is not strictly linear: expect `O(V log V + E log E + D log D)` ordering work, plus string comparison/encoding costs, within the accepted input limits. Measure accepted-limit requests and concurrent overload before finalizing defaults. Proposed application-owned responses: - `200`: complete valid topology; advisory metadata may carry warnings. - `400 / INVALID_JOB_ID`: malformed or non-positive/out-of-range ID. - `404 / JOB_NOT_FOUND`: no known active/pending job or usable retained history after the bounded lookup. - `409 / LINEAGE_UNAVAILABLE`: known job but missing or inconsistent DAG metadata. - `413 / LINEAGE_GRAPH_TOO_LARGE`: a structural, string or serialized-byte limit is exceeded. This use of 413 for a response limit needs explicit REST-contract agreement. - `503 / LINEAGE_TEMPORARILY_UNAVAILABLE`: recovery, ownership transition, deadline or capacity prevents a safe response. - `500 / LINEAGE_INTERNAL_ERROR`: unexpected non-fatal application failure. Application errors use fixed `{ "code": "...", "message": "..." }` fields, without raw request values, exception text, configuration, topology names or stack frames. Restore thread interruption where applicable and do not swallow fatal JVM errors. Preserve existing authentication/container error behavior; the endpoint cannot promise its JSON envelope for failures that occur before its handler runs or after a client disconnects. ### 5. Security and compatibility boundary Use the existing HTTP authentication boundary. Basic Auth is configurable and may be disabled; this feature does not introduce authentication where none is configured, nor does it add per-job authorization. Document network/TLS controls and the existing cluster-wide visibility model. Return only whitelisted graph fields. Exclude environment options, configuration, addresses, JAR locations, SQL text extracted from config, traces, credentials and exception details. Display labels and table names can themselves contain sensitive user-supplied text, so do not promise that arbitrary configured labels are secret-free. Use private/no-store response caching policy and ensure documentation treats topology as restricted operational metadata. The shared exception filter now omits stack frames but still returns its own status/message envelope. No shared-filter security fix is included in this proposal. Use endpoint-owned fixed error translation where possible without changing other REST behavior. ### 6. Relationship to OpenLineage work [#12207](https://github.com/apache/seatunnel/issues/12207) and PRs [#12208](https://github.com/apache/seatunnel/pull/12208), [#12209](https://github.com/apache/seatunnel/pull/12209), [#12210](https://github.com/apache/seatunnel/pull/12210) cover opt-in outbound table-level lifecycle reporting for Zeta/Flink. This proposal covers an on-demand read-only job topology view for Zeta, including intermediate transform nodes. There is no proposed dependency on enabling OpenLineage or deploying a receiver. Do not copy its transport, backend SPI, token configuration, lifecycle hooks, statistics or retry queues into this work. Coordinate with that author and reviewers on: - Terminology: job/node identity versus run/attempt identity; neither is a substitute for the other. - Dataset semantics: raw reported paths here versus normalized namespace/name there. No universal join key is promised. - Metadata extraction: reuse a compatible approved helper if appropriate; do not introduce a competing connector-metadata SPI. The REST projection should remain usable without an outbound-reporting module. - Test fixtures for multi-table jobs, missing metadata and restores, while keeping acceptance criteria separate. If maintainers prefer a shared identity model before either public contract, record that dependency explicitly and pause the affected API slice. Related open PRs do not establish acceptance or require this proposal to inherit their full scope. ## Compatibility, Deprecation, and Migration Plan The intended change is additive: a new endpoint only, with existing `/job-info` responses and consumers unchanged. No change to serialized `JobDAGInfo`/`VertexInfo`, checkpoints, savepoints, connector SPI, job options, scheduler behavior or history storage is proposed. No new runtime library is planned. `schemaVersion=1` is an API version, not a graph generation counter. Clients must tolerate unknown optional fields. Removing fields, redefining identities or changing enum semantics requires an explicit compatibility/version decision. Do not advertise stable semantics before the STIP is accepted and corresponding tests pass. ## Test Plan No Java 8/11 runtime validation is claimed for this draft. Both JDKs are required for the eventual implementation. - Mapper: branching and joining graphs; multiple pipelines; chained transforms; large/signed numeric IDs; shuffled maps/edges; multi-table and missing/default/mixed metadata; invalid references and cycles; immutability under caller mutation. - Bounds: exact and one-over boundaries for every limit, multibyte characters, escaping, repeated entries before deduplication, warning growth, concurrent admission saturation, bounded timeout cleanup and a slow/disconnected client. Confirm no partial success body is committed on rejection. - Service: pending/running/finished/expired/unknown jobs, failed initialization, separate history-entry expiry, active-to-finished transition, follower routing, unavailable master, same-ID active/history coexistence, repeated reconstruction and failover. - Identity: compare node/edge identities across the restore/retry cases above using actual DAG-generation paths, not only mocked lookup responses. Retain explicit evidence or revise the design before implementation approval. - Security: enabled/disabled/invalid Basic Auth behavior; missing auth header; unexpected exceptions containing sentinel credentials; absence of configuration/address/JAR fields; unchanged shared endpoint/filter behavior. - Compatibility: existing REST fixtures, serialized-class compatibility checks and relevant engine tests on Java 8/11; no connector/receiver/trace dependency introduced by lookup or projection. - E2E: representative batch and streaming jobs with source/transform/sink paths, branching/multi-table coverage, retained-history lookup and master transition. Run source and result checks rather than asserting only HTTP 200. - Performance: record cold lookup separately from cached projection, accepted-limit allocations/latency, and mixed normal/overload traffic. Choose final limits from measurements, not this draft's provisional numbers. ## Implementation and Decision Slices - [ ] A — Discuss this proposal on the issue/dev list; reconcile #12207; record explicit decisions on identity, lookup cost, route/errors, limits and access model. No implementation PR is implied by publishing a draft. - [ ] B — After agreement, implement the pure bounded mapper and immutable response with focused tests on Java 8/11. No route, connector calls or new state in this slice. - [ ] C — Add the read service and endpoint, bounded lookup/admission, history/race handling and security tests. Recheck upstream overlap before touching shared engine paths. - [ ] D — Complete real engine lifecycle validation, performance measurements, compatibility coverage and EN/ZH user documentation; link implementation PRs to the accepted STIP and #11484. Each slice has its own reviewable PR once authorized and validated. Do not open another design-only PR. No implementation issue numbers or PR links are invented in this draft. ## Decisions Requested Before Implementation All decisions below are **pending**. An issue assignment, an open proposal, or positive direction on scope is not acceptance of these contracts. 1. **D1 — Scope and source:** accept the available execution-oriented `JobDAGInfo` as the V1 topology source, with reported paths only and no column/table-mapping precision implied. 2. **D2 — Identity:** agree the required stability across retry, same-ID resume and master recovery. Require real reconstruction evidence; revisit graph-revision identity explicitly if existing state cannot support the guarantee. 3. **D3 — Lookup and history:** agree current-versus-retained precedence and temporary-unavailability behavior without unbounded plan rebuilding, new HA state, or job execution changes. 4. **D4 — Public REST contract:** agree route, schema versioning, response fields and status/error mapping, including whether 413 is appropriate for response limits. 5. **D5 — Resource budget:** review the provisional structural/string/byte limits, per-process admission and lookup deadline. Require measurements and bounds on in-flight master work before finalizing values. 6. **D6 — Access and overlap:** accept the existing HTTP access model with explicit topology-sensitivity documentation; reconcile terminology/metadata reuse with #12207 without creating a receiver dependency. Decision records should link the specific issue comment or dev-list message and record any agreed changes in this canonical body. When a dev-list thread is sent, add its archive link here. Implementation PRs should map their acceptance tests to these decisions and the slices above. ## Alternatives Considered - Keep consuming `/job-info`: no new endpoint, but no separate bounded lineage contract and unnecessarily broad payload coupling. Community may prefer documenting a safe subset; discuss before committing to a route. - Use OpenLineage events as the REST source: introduces receiver/storage/delivery dependencies and does not directly preserve transform topology. Keep export and on-demand graph inspection complementary. - Use StainTrace: sampled record traces are not an authoritative complete topology source. - Add a new persistent logical-plan graph: may provide stronger identity semantics, but changes retention and compatibility scope. Return to design review if existing DAG state cannot meet the agreed contract. - Export column lineage now: requires transform semantic analysis and is not a small extension of this proposal. ## References - [Feature request and scoped design discussion](https://github.com/apache/seatunnel/issues/11484) - [Direction to use issue/STIP/email rather than a design-only PR](https://github.com/apache/seatunnel/pull/11941#issuecomment-5492740257) - [Historical design branch](https://github.com/goutamadwant/seatunnel/tree/feature/GH-11484-job-lineage-design) — predates this revision; not the canonical updated contract. - [OpenLineage umbrella and review requirements](https://github.com/apache/seatunnel/issues/12207) - [STIP guide](https://seatunnel.apache.org/community/contribution_guide/STIP/) - [Source baseline reviewed](https://github.com/apache/seatunnel/commit/bbde250914346068f50131e7907bef2c9ecbc448) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
