DanielLeens opened a new issue, #11558:
URL: https://github.com/apache/seatunnel/issues/11558
### Search before asking
- [x] 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.
- [x] 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
```text
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:
```java
@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:
```text
SourceCommandEnvelope
protocolVersion
jobId
sourceRuntimeId
coordinatorEpoch
senderAttemptId
targetAttemptId
senderSequence
commandId
commandKind
durability
payloadVersion
payloadChecksum
payloadBytes
```
Admission returns:
```text
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:
```text
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:
```java
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:
```text
CREATED -> RESTORING -> RUNNING -> DRAINING -> CLOSED
\-> CANCELLING -> CLOSED
\-> FAILED -> CLOSED
```
Schema change is an orthogonal sub-state:
```text
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?
- [x] Yes, I am willing to submit a PR.
### Code of Conduct
- [x] I agree to follow the Apache Software Foundation Code of 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]
