[STIP-31][Feature][Zeta] Engine-managed Source runtime serialization and concurrency contract

Search before asking

• I searched open and closed SeaTunnel issues and STIPs for SourceReader mailbox, checkpoint lock, split assignment, Source coordinator event loop, and engine-managed connector concurrency.
• Existing issues such as #5694 and #2936 describe individual checkpoint-lock or split/checkpoint atomicity failures, but no existing STIP defines the complete engine-owned Source concurrency and recovery contract proposed here.

Summary

This STIP proposes an opt-in, engine-managed Source runtime for Zeta that makes the engine, rather than each connector, responsible for callback serialization, checkpoint cuts, command admission, split-assignment recovery, and lifecycle ordering.

The proposal introduces two explicit execution lanes:

• Lane A: managed runtime. Eligible connectors run under a bounded Reader event loop and a Coordinator event loop. The engine owns all checkpoint-visible state transitions.
• Lane B: legacy runtime. Existing connectors retain the current SourceReader SPI, checkpoint monitor identity, private locks, and behavior. No connector is migrated silently.

The delivery scope is Phase 0 through Phase 3 plus Phase 4-lite. There is no mandatory repository-wide Phase 5 migration.

Motivation

Source concurrency boundaries are currently distributed across:

• SourceFlowLifeCycle.collect() and SourceReader.pollNext(...);
• Collector.getCheckpointLock() and SourceReader.snapshotState(...);
• split assignment and source-event Hazelcast operations;
• SourceSplitEnumeratorTask callbacks;
• connector-owned locks, background fetchers, and scheduler threads;
• schema-change checkpoint callbacks and close handling.

As a result, a connector author must reason about multiple engine threads, remote operations, checkpoint ordering, restore, and custom locks. This has caused long checkpoint-lock stalls, ambiguous split ownership, operation-thread blocking, and implementation-specific concurrency patterns.

Replacing every lock with a queue is not sufficient. The production contract must answer:

1. Which execution context owns Reader and Enumerator state?
2. When is a command admitted, applied, and included in a checkpoint?
3. How are duplicate delivery, lost acknowledgements, stale attempts, failover, and rescale handled?
4. How are queues bounded without blocking Hazelcast operation threads?
5. How do existing connectors remain compatible?

Goals

1. Make the engine the single owner of checkpoint-visible Source state for Lane A.
2. Remove connector callbacks from Hazelcast operation threads.
3. Define a strict and recoverable ordering between controls, polling, and checkpoints.
4. Provide at-least-once command delivery with idempotent apply and attempt fencing.
5. Track split assignment until checkpoint inclusion is proven.
6. Move slow Enumerator discovery to engine-managed workers while applying results only on the Coordinator event loop.
7. Preserve all existing connector behavior through Lane B.
8. Keep resource usage bounded by command count, payload bytes, and worker-wide budgets.
9. Version wire commands, connector payloads, capabilities, and checkpoint metadata independently.
10. Provide a production kill switch, metrics, conformance tests, and fault-injection coverage.

Non-goals

1. This STIP does not redesign Sink concurrency.
2. It does not replace Hazelcast.
3. It does not remove SourceReader.pollNext(...) or Collector.getCheckpointLock().
4. It does not migrate all existing connectors.
5. It does not persist and replay arbitrary connector SourceEvent instances.
6. It does not add one mailbox thread per task.
7. It does not provide synchronous Mailbox.call(...) or synchronous cross-thread Coordinator calls.

Required invariants

1. Single owner: one event loop is the only owner of a Lane A Reader or Enumerator's checkpoint-visible state.
2. No operation blocking: Hazelcast operation threads validate and offer commands only; they never invoke connector callbacks or wait for application.
3. No event-loop RPC wait: Reader and Coordinator event loops never call .get() or .join() on remote futures.
4. Strict checkpoint cut: every ordered command admitted before a barrier is applied before the snapshot; later commands belong to a later checkpoint.
5. Atomic state and emit: emitted records and their split-state advancement are complete within one poll turn before a snapshot can run.
6. At-least-once transport: commands may be delivered more than once, but one command has one business effect within an attempt.
7. Attempt fencing: stale Reader attempts, Coordinator epochs, and async-worker results cannot mutate the current attempt.
8. Bounded resources: command count, payload bytes, and worker-wide mailbox bytes are all bounded.
9. Bounded control latency: an ordered control waits for at most one compliant poll turn.
10. Terminal preemption: cancel and fatal failure use an independent terminal signal and invoke wakeup().
11. Versioned recovery: wire and checkpoint state use explicit versioned serializers, never long-term Java native serialization.
12. No silent lane switch: restore uses the lane and capability digest recorded in runtime metadata.
13. Legacy identity: Lane B retains the existing checkpoint-monitor identity.
14. No per-task thread: the Reader event loop is driven by the existing task execution resource.

Architecture

Hazelcast operation
  -> SourceCommandTransportAdapter
  -> bounded ReaderMailbox
  -> ReaderEventLoop
  -> ManagedSourceReader
  -> connector state + runtime checkpoint metadata

CoordinatorEventLoop
  -> SourceAssignmentTracker
  -> versioned command transport
  -> ReaderMailbox

CoordinatorEventLoop
  -> immutable discovery input
  -> shared SourceAsyncWorkerPool
  -> epoch-fenced result
  -> CoordinatorEventLoop

The Reader event loop reuses the existing SeaTunnelTask execution resource. Polling is a coalesced turn, not a command per record. The loop always checks terminal state and ordered controls before starting another poll turn.

Managed Reader capability

The additive API is experimental until it has survived at least one release cycle:

@Experimental
public interface ManagedSourceReader<T, SplitT extends SourceSplit>
        extends SourceReader<T, SplitT> {

    ManagedSourceCapability managedSourceCapability();

    CompletableFuture<Void> isAvailable();

    PollStatus pollNextManaged(Collector<T> output, PollContext context) throws Exception;

    void wakeup();
}

PollStatus contains MORE_AVAILABLE, NOTHING_AVAILABLE, and END_OF_INPUT.

PollContext exposes remaining record and byte budgets, a deadline, and shouldYield().

Lane A requires:

• bounded, non-blocking polling;
• no lost availability wakeup;
• idempotent and thread-safe wakeup();
• stable split IDs;
• no background thread mutation of checkpoint-visible state;
• versioned state and durable event serializers;
• passing the managed Source conformance suite.

A versioned capability descriptor is used instead of a boolean. It includes runtime protocol version, bounded-poll support, wakeup support, attempt-fencing support, source-event support, async-enumerator support, and a stable digest.

Reader scheduling

Scheduling categories are:

Category	Examples	Ordering
TERMINAL	cancel, fatal failure, force close	independent atomic signal; preempts normal work
ORDERED_CONTROL	split assignment, source event, checkpoint, graceful close	strict FIFO
ASYNC_COMPLETION	async I/O and scheduler completion	returned to the owner loop, bounded and coalescible
POLL	record production	low priority, at most one pending turn
OBSERVABILITY	metric and diagnostic sampling	lowest priority, droppable or coalescible

A poll turn yields when any configured records, bytes, or time budget is reached. A soft-budget violation records diagnostics. A hard-budget violation invokes wakeup(). Failure to return within the cancellation timeout fails the task; the runtime never changes lane online.

Versioned command protocol

Every cross-thread or cross-node ordered command uses an immutable envelope:

SourceCommandEnvelope
  protocolVersion
  jobId
  sourceRuntimeId
  coordinatorEpoch
  senderAttemptId
  targetAttemptId
  senderSequence
  commandId
  commandKind
  durability
  payloadVersion
  payloadChecksum
  payloadBytes

Admission returns:

AdmissionAck
  status
  targetAttemptId
  mailboxSequence
  retryAfterMillis
  reasonCode

Statuses are ACCEPTED, DUPLICATE, RETRY_LATER, STALE_TARGET,
TERMINAL_REJECTED, UNSUPPORTED_PROTOCOL, and INVALID_PAYLOAD.

ACCEPTED means that validation passed, capacity was reserved, a mailbox sequence was allocated, and the command is visible to the current target attempt. It does not mean the command was applied or checkpointed.

Commands are classified as:

• CHECKPOINT_COUPLED: split assignments and explicitly versioned state events;
• RECONSTRUCTABLE: registration, split request, and no-more-splits control;
• EPHEMERAL: availability and diagnostics;
• TERMINAL: cancellation and fatal failure.

Phase 1 only permits built-in split assignment and no-more-splits as durable engine commands. Arbitrary durable custom SourceEvent requires a separate reviewed serializer and recovery contract.

Admission and overload

Admission is linearized only after:

1. protocol, attempt, header, checksum, and payload-size validation;
2. command-count and byte-capacity reservation;
3. mailbox-sequence allocation;
4. publication to the Reader event loop.

The operation thread does not deserialize connector state, invoke connector code, sleep, spin, or wait.

Capacity is bounded at three levels:

• per-mailbox command count;
• per-mailbox payload bytes;
• per-worker Source mailbox bytes.

Checkpoint, graceful-close, checkpoint-callback, and assignment-ack controls have reserved capacity. Terminal state has independent capacity. A normal full queue returns RETRY_LATER; exhausted reserved capacity is an invariant violation and fails the task.

Checkpoint ordering

Local barriers enter the same target sequencer as ordered controls.

1. Commands with a lower sequence are applied first.
2. Poll turns do not consume a sequence and run only between ordered controls.
3. At the barrier, there is no accepted-before-barrier command left unapplied.
4. Reader state and engine runtime metadata are snapshotted.
5. State registration, task acknowledgement, and downstream barrier forwarding happen only after a successful snapshot.
6. A terminal preemption prevents the checkpoint from succeeding.

The mailbox queue itself is never checkpointed.

Split assignment ownership

The engine tracks each assignment:

State	Meaning
UNASSIGNED	still owned by the Enumerator
DISPATCHED	command created, admission not yet proven
ADMITTED	current Reader attempt accepted the command
APPLIED	Reader invoked addSplits once
CHECKPOINT_INCLUDED	Reader runtime metadata proves inclusion
GC_ELIGIBLE	the corresponding global checkpoint completed

An apply acknowledgement is useful for flow control but is not durability proof. The Coordinator retains assignment ownership until checkpoint inclusion and global checkpoint completion are reconciled.

Reader checkpoint metadata contains:

SourceRuntimeCheckpointMetadata
  metadataVersion
  commandProtocolVersion
  managedLaneVersion
  readerAttemptLineage
  appliedSenderWatermarks
  boundedSequenceGaps
  noMoreSplitsGeneration
  capabilityDigest

Reader failover creates a new attempt and restores connector state plus runtime metadata. Coordinator reconciliation returns assignments after the restored watermark to the Enumerator. Commands addressed to the old attempt are rejected.

Coordinator failover increments the epoch, restores Enumerator and tracker state in one checkpoint cut, cancels old async tasks, and reconciles registered Readers before new assignment.

Rescale keys ownership by stable split identity, not old subtask index.

Coordinator event loop and scheduler

The Coordinator event loop serializes:

• open and close;
• Reader registration;
• split requests and returned splits;
• source events;
• Enumerator snapshot;
• checkpoint complete and abort callbacks;
• async discovery results and scheduler ticks.

The scheduler exposes asynchronous APIs only:

public interface CoordinatorScheduler {

    <T> Cancellable callAsync(
            AsyncTaskKey key,
            Callable<T> callable,
            BiConsumer<T, Throwable> resultHandler,
            AsyncTaskOptions options);

    Cancellable scheduleInCoordinatorThread(
            AsyncTaskKey key,
            Duration delay,
            Runnable task);
}

Slow work receives immutable input and runs in a shared bounded engine worker pool. The result handler always returns to the Coordinator event loop and is fenced by epoch. Fixed-delay scheduling, no implicit re-entry, COALESCE_ONE by default, timeout, cancellation, failure policy, worker class, and context classloader are mandatory lifecycle properties.

Lifecycle and schema change

The main lifecycle is:

CREATED -> RESTORING -> RUNNING -> DRAINING -> CLOSED
                             \-> CANCELLING -> CLOSED
                             \-> FAILED -> CLOSED

Schema change is an orthogonal sub-state:

IDLE -> QUIESCING -> TRIGGER_REQUESTED -> WAITING_END -> IDLE

Only the bound schema checkpoint ID and request epoch may advance the sub-state. During schema change, no new poll turn runs. A matching completion returns to RUNNING, or to DRAINING if graceful close/end-of-input was latched. Abort, timeout, overlapping schema changes, and trigger failure fail the task. Stale and duplicate callbacks are ignored and counted. Cancel/fatal failure fences the request and cancels its future.

Compatibility and upgrades

Lane B remains fully independent and is the default for existing connectors.

Lane selection is recorded in:

• job plan;
• task deployment descriptor;
• checkpoint runtime metadata.

Restore never chooses a lane from the currently installed connector alone. A capability-digest mismatch fails before execution unless an explicit, tested migration exists.

The command protocol, connector payload version, and runtime-metadata version evolve independently. Unknown major versions fail fast. The first release may require a homogeneous cluster; mixed-version support is enabled only after a permanent compatibility matrix exists.

Initial production defaults

Setting	Initial value
Reader mailbox maximum commands	1024
Reader mailbox maximum payload bytes	4 MiB
Reserved control commands	64
Reserved control bytes	256 KiB
Worker Source mailbox maximum bytes	256 MiB
Maximum command payload	512 KiB
Poll maximum records per turn	64
Poll soft duration	5 ms
Poll hard duration	1 s
Poll cancellation timeout	30 s
Admission p99 budget	5 ms
Retry initial/max backoff	10 ms / 1 s
Coordinator async maximum concurrency	4
Assignment tracker maximum entries	100000
Assignment tracker maximum bytes	64 MiB

All settings use engine configuration options, have tests and documentation, and are not connector-overridable. Correctness-critical settings are not hot-reloadable.

Observability

Reader metrics cover admission status, mailbox count/bytes, reserved usage, oldest age, queue wait, command service time, poll records/bytes/time, wakeup, checkpoint stages, applied watermarks, and output-blocked time.

Coordinator metrics cover mailbox usage, assignment states, oldest uncheckpointed assignment, async queue/execution time, overlap/coalescing, stale results, registration reconciliation, tracker size/compaction, and assignment backpressure.

Metric labels are bounded. Command IDs, split IDs, table paths, payloads, credentials, and exception text are never labels or logs.

Validation

Deterministic event-loop tests cover all adjacent command/barrier permutations, availability wakeups, poll yielding, terminal preemption, duplicates, stale attempts/epochs, checkpoint callback ordering, and every schema transition.

Fault injection covers:

1. command creation;
2. operation send;
3. capacity reservation;
4. admission acknowledgement;
5. connector apply before return;
6. apply completion;
7. apply acknowledgement;
8. Reader snapshot before state registration;
9. state registration before task acknowledgement;
10. task acknowledgement before downstream forwarding;
11. global checkpoint completion before tracker garbage collection.

Model-based tests randomly generate assignment, duplicate delivery, ACK loss, Reader and Coordinator failover, checkpoint complete/abort, rescale, no-more-splits, disabled checkpoints, and sustained checkpoint failure.

The conformance suite validates non-blocking poll, budgets, no-lost-wakeup, cancellation, state/emit atomicity, duplicate split assignment, callback idempotence, background-thread confinement, source-event durability declaration, and context-classloader restoration.

The performance matrix covers Fake Source, Kafka/Pulsar, JDBC Lane B control, Iceberg discovery, split storms, and blocking controls at parallelism 1/16/128/512. Exit criteria are no more than 3% steady-state throughput regression, 5% CPU and heap increase, checkpoint p99 increase no greater than max(5%, 100 ms), admission p99 at most 5 ms, bounded mailbox memory, and no monotonic tracker growth in a 24-hour soak.

Implementation plan in one Draft PR

The implementation remains reviewable through phase-gated commits, but all work is delivered in one Draft PR:

Phase 0: protocol and measurement

• runtime contention metrics;
• temporary Lane A allowlist, defaulting every connector to Lane B;
• versioned runtime-metadata serializer;
• deterministic fault-injection fixtures and managed Fake Source.

Phase 1: Reader managed runtime

• command envelope, admission acknowledgement, deduplication, attempt fencing;
• bounded Reader mailbox and cooperative event loop;
• minimal assignment tracker and Reader checkpoint metadata;
• Fake Source and an asynchronous streaming Reader pilot;
• production kill switch.

Phase 2: Coordinator event loop

• serialized Enumerator callbacks;
• attempt-aware registration;
• bounded async worker and scheduler lifecycle;
• assignment tracker checkpoint and reconciliation;
• Iceberg streaming Enumerator pilot.

Phase 3: lifecycle and schema change

• executable main and schema lifecycle state machines;
• event-loop checkpoint callbacks;
• close latch, cancellation, timeout, stale callback, and failover handling.

Phase 4-lite: stable capability and connector adoption

• versioned capability descriptor;
• connector-common managed Reader base;
• conformance suite;
• at least two different connector types using Lane A;
• maintainer policy for new connectors, with an explicit Lane B exception path.

There is no mandatory all-connector migration phase.

Rollout and rollback

The feature is disabled by default and enabled by allowlist/capability. Initial production rollout is canary-only.

A global kill switch affects only new deployments and restores. Running tasks never switch lane. A rollback must preserve checkpoint readers for every published runtime-metadata version. If an older binary cannot read the current managed metadata, downgrade is rejected and the job must restore with a compatible binary or from a compatible savepoint.

Acceptance criteria

• Existing connectors remain in Lane B unless explicitly eligible.
• Operation threads never execute connector callbacks for Lane A.
• Reader and Enumerator checkpoint-visible state have one owner each.
• Commands are bounded, versioned, fenced, deduplicated, and checkpoint ordered.
• Split ownership survives lost ACK, Reader failover, Coordinator failover, and rescale.
• Mailbox queues and arbitrary SourceEvents are not persisted as generic state.
• Slow discovery uses engine-managed workers and epoch-fenced event-loop application.
• Schema change, close, cancellation, timeout, duplicate callback, and failover transitions are deterministic.
• Runtime metadata, lane, and capability are restore compatible and never switch silently.
• At least two different connector types pass the Lane A conformance suite.
• Required fault-injection, compatibility, performance, and soak gates pass before the PR leaves Draft.
• English and Chinese documentation describe configuration, compatibility, migration, metrics, and rollback.

Usage Scenario

Connector authors should be able to implement a Source without designing their own checkpoint lock or coordinating framework callbacks from multiple engine threads. Operators should gain bounded control latency, explicit overload behavior, deterministic recovery, and enough diagnostics to identify a non-compliant connector without changing existing jobs.

Related issues and implementation

• Historical checkpoint-lock stall: #5694
• Historical split/checkpoint atomicity issue: #2936
• Schema-change ordering discussion: #11402
• Implementation Draft PR: #11557

Are you willing to submit a PR?

• Yes, I am willing to submit a PR.

Code of Conduct

• I agree to follow the Apache Software Foundation Code of Conduct.