nzw921rx opened a new issue, #12457: URL: https://github.com/apache/seatunnel/issues/12457
### 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 proposal introduces a native **Zeta Application Mode** for YARN and Kubernetes. One submission creates one isolated Zeta cluster for one SeaTunnel job: an application master starts the Zeta master, provisions a fixed number of Zeta workers, submits the job after all workers register, and releases the resources when the job terminates. This STIP turns the umbrella discussion in #11857 into a concrete architecture and MVP boundary. ## Background Zeta currently supports long-running standalone clusters. That model works well when one cluster hosts multiple jobs, but it does not fit teams that use YARN or Kubernetes as their resource control plane and want job-level isolation. Native Application Mode allows these teams to: - allocate resources for one job without deploying a long-running SeaTunnel cluster; - isolate dependencies, resources, lifecycle, and failures by job; - release batch resources immediately after completion; - run a streaming job as an independently observable platform application; - recover a failed execution from a durable Zeta checkpoint. The implementation reuses the existing Zeta scheduler, worker registration, slot service, connector runtime, and checkpoint service. ## Goals 1. One platform application contains exactly one Zeta job. 2. One application master coordinates one Zeta master and a fixed number of workers. 3. YARN containers and Kubernetes Pods host native Zeta workers. 4. Platform resource provisioning remains separate from Engine slot scheduling. 5. Existing standalone, Flink, and Spark behavior remains compatible. 6. YARN and Kubernetes providers are optional components of the standard SeaTunnel distribution. 7. A new application can recover from a completed checkpoint written by an earlier application. ## Non-goals for the first phase - Automatic master failover - Automatic worker replacement - Autoscaling - Kerberos, delegation token, or keytab support - Multiple jobs in one application - Savepoint-based application upgrades - Replacing Hazelcast membership - Replacing the existing Zeta slot scheduler ## Execution model ```mermaid flowchart LR client["SeaTunnel Application CLI"] platform["YARN / Kubernetes"] subgraph app["One platform application"] master["Application Master<br/>Zeta Master"] worker1["Zeta Worker 1"] workerN["Zeta Worker N"] end checkpoint[("Persistent checkpoint storage")] client -->|submit / status / cancel| platform platform --> master master -->|request fixed workers| platform platform --> worker1 platform --> workerN worker1 -->|Hazelcast TCP join| master workerN -->|Hazelcast TCP join| master master --> checkpoint ``` The master coordinates the application but does not provide task slots. Workers execute task groups. One master can coordinate three or more workers. Fixed capacity is: ```text application.worker-count × application.worker.slots ``` Effective job parallelism still depends on the source, transform, and sink topology. ## Proposed architecture ### Deployment contracts A shared resource-manager core defines: - immutable application and worker specifications; - application IDs, status, and terminal results; - deployment, status, cancellation, and provider discovery contracts; - stable user-facing options; - distribution-local connector path resolution. These contracts do not belong in Engine client. Engine client remains responsible for native Zeta job communication over Hazelcast. External platform deployment is a separate lifecycle. ### Two resource-management layers | Layer | Owner | Unit managed | | --- | --- | --- | | Platform provisioning | Resource manager driver | YARN containers or Kubernetes Pods with CPU and memory | | Engine scheduling | Master ResourceManager and worker SlotService | Task-group slots on registered workers | The platform driver creates and releases worker processes. After a worker joins the isolated Hazelcast cluster, the existing Zeta resource manager registers its fixed slots. The existing scheduler requests and assigns those slots without knowing whether the worker runs in a YARN container or Kubernetes Pod. ### Runtime lifecycle ```mermaid sequenceDiagram participant CLI participant Deployer participant Platform as YARN / Kubernetes participant Runtime as Application Runtime participant Driver as Resource Manager Driver participant Worker as Zeta Worker participant Engine as Zeta Scheduler CLI->>Deployer: submit job and deployment options Deployer->>Platform: create application Platform->>Runtime: start application master Runtime->>Runtime: start isolated Zeta master Runtime->>Driver: request fixed workers Driver->>Platform: allocate containers or Pods Platform->>Worker: start N workers Worker->>Runtime: join and register slots Runtime->>Engine: submit one native job Engine-->>Runtime: success, failure, or cancellation Runtime->>Driver: release workers Runtime-->>Platform: publish result and exit ``` The deployer owns rollback until the platform accepts the application. After startup, the application runtime owns the Zeta master and job lifecycle. The driver owns every requested worker, including requests still pending when cleanup starts. Closing a client handle must not cancel the remote application. Cancellation is explicit so detached submission, later status retrieval, and later cancellation work predictably. ### Cluster discovery Every submission creates a unique Hazelcast cluster name. Workers are lite members and join only the advertised master address through TCP discovery. Application Mode does not inherit multicast, cloud discovery, or standalone peer lists. The application master is the only non-lite member in the MVP. Engine `backup-count` is therefore zero. ## Platform mapping | Concern | YARN | Kubernetes | | --- | --- | --- | | Master | ApplicationMaster container | `batch/v1` Job Pod | | Worker | YARN container started through AMRMClient/NMClient | Worker Pod created by the master | | Distribution | Archive localized from a shared filesystem | Immutable image shared by master and workers | | Discovery endpoint | Actual AM host and selected Hazelcast port | Headless Service and master Pod address | | External ID | YARN application ID | Generated Job name | | Durable result | YARN final application state | Job `Complete` or `Failed` condition | | Cleanup | Release requests and containers, then remove staging | Delete labeled Pods and owner-referenced resources | YARN uses one application attempt in the MVP. Kubernetes uses a Job without retry. Losing the master or any worker fails the current application. ## Checkpoint recovery and HA boundary Live master replication and persistent job checkpoints solve different problems. | Mechanism | Purpose | | --- | --- | | Hazelcast backup replicas | Allow a surviving eligible master to take over live coordination state | | Persistent checkpoints | Store source progress and operator or sink state outside application processes for a new execution | Setting `backup-count` above zero does not provide HA when no standby master process exists. Complete HA also requires: - a standby or replacement master process; - leadership fencing; - platform process replacement; - durable application ownership; - reconciliation between platform and Engine state. Checkpoint recovery does not require Hazelcast backup replicas. Before the first submission, users configure the existing native checkpoint storage in `seatunnel.yaml`. The complete storage configuration is preserved, including HDFS, OSS, S3, COS, namespace, endpoint, filesystem, and credential options. ```mermaid sequenceDiagram participant Old as Application A / Job A participant Store as Persistent checkpoint storage participant CLI participant New as Application B / Job B Old->>Store: write completed checkpoints Old--xOld: master or worker failure CLI->>New: submit with restore job ID = Job A New->>Store: find the latest eligible checkpoint Store-->>New: source and task state New->>New: restore through the native Zeta checkpoint service New->>Store: write later checkpoints under Job B ``` Recovery creates a new platform application ID and a new native Zeta job ID. The restore source is the historical job ID. If no eligible checkpoint exists, submission fails instead of silently starting from the beginning. Checkpoint storage must outlive both applications: - A YARN checkpoint directory must be outside application staging. - A Kubernetes PVC is caller-owned and must not receive an application owner reference. - Remote object storage does not require a PVC. ## Configuration contract Common options include: - application name; - native job ID and restore job ID; - worker count, CPU, memory, and slots; - master CPU, memory, and port; - startup timeout. YARN adds: - distribution archive; - Hadoop configuration directory; - staging directory; - queue. Kubernetes adds: - namespace; - image and image pull policy; - service account; - SeaTunnel home; - kubeconfig; - finished Job retention; - optional checkpoint PVC. All user-facing settings use SeaTunnel `Option` definitions. Their names and defaults are compatibility contracts. ## Distribution and dependency isolation Application Mode uses the standard SeaTunnel binary distribution. It does not create another release archive. - The shared resource-manager core is built by default. - The `yarn` profile adds the YARN provider. - The `kubernetes` profile adds the Kubernetes provider. - Selected providers are placed under `resource-managers/yarn` and `resource-managers/kubernetes`. - The launcher loads only the requested provider. YARN reuses the Hadoop runtime already shipped by SeaTunnel. The Kubernetes provider shades and relocates SDK dependencies that may conflict with Engine libraries while preserving service-provider metadata. Master and workers always use the same localized distribution or image, including connector and checkpoint storage dependencies. ## Failure semantics | Failure | Required behavior | | --- | --- | | Invalid configuration | Reject before platform submission | | Partial platform submission | Roll back resources created by the submission | | Worker request failure | Fail the application and release requested workers | | Worker registration timeout | Fail within the configured startup deadline | | Worker exit or cluster departure | Fail the application without replacing the worker | | Job failure or cancellation | Preserve the native result and begin cleanup | | Master exit | Let the platform record application failure | | Cleanup failure | Continue remaining cleanup and preserve diagnostics | No in-process cleanup can run after `SIGKILL` or host loss. Platform ownership handles process termination. Staging directories, retained Kubernetes Job metadata, and persistent checkpoints follow separate retention policies. ## Security and operations - Job and deployment configuration may contain credentials and must remain private. - YARN staging directories use submitter-only permissions. - Kubernetes job configuration is localized in a namespace-scoped ConfigMap mounted only by the application master. - Kubernetes RBAC is namespace-scoped. - The master receives the ServiceAccount token; workers do not require Kubernetes API credentials. - Private image credentials are managed by the cluster or ServiceAccount. - Secrets must not be written to logs. ## Compatibility This proposal does not rename existing options or change standalone defaults. Existing Zeta resource scheduling remains in place. Existing Flink and Spark entrypoints and distributions remain unchanged. Provider discovery uses SPI, allowing another platform provider to be added without introducing a dependency from Engine or starter to that provider. ## Validation The implementation should include: - unit tests for option validation, serialization, provider discovery, platform resource construction, rollback, cancellation, and status mapping; - runtime tests for successful jobs, invalid jobs, startup timeouts, worker loss, checkpoint recovery, and cleanup failures; - YARN E2E on MiniYARN and MiniDFS using real master and worker JVMs; - Kubernetes E2E on a disposable K3s cluster using real Job and worker Pods; - the same FakeSource, Console, and Assert job configurations on both platforms; - checkpoint recovery with HDFS on YARN and a persistent local volume on Kubernetes. ### Usage Scenario _No response_ ### Related issues _No response_ ### 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) -- 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]
