DanielLeens commented on code in PR #12165: URL: https://github.com/apache/seatunnel/pull/12165#discussion_r4017783941
########## seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/scheduler/SharedCheckpointScheduler.java: ########## @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.engine.server.checkpoint.scheduler; + +import org.apache.seatunnel.shade.com.google.common.util.concurrent.ThreadFactoryBuilder; + +import lombok.extern.slf4j.Slf4j; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * Node-wide scheduler shared by every checkpoint coordinator on this member. + * + * <p>Before this existed, each {@code CheckpointCoordinator} built its own two-thread scheduled + * pool, so timer threads grew with the number of active pipelines. The coordinators only ever used + * that pool as a timer: to re-arm the periodic checkpoint trigger and to arm the checkpoint-timeout + * watchdog. This class keeps a single small timer pool for the whole member and hands each + * coordinator a {@link PipelineCheckpointScheduler} lease. + * + * <p>Timing and execution are deliberately split across two pools. A timer thread does nothing but + * hand the task to the dispatch pool, so a checkpoint body that blocks on an RPC delays neither its + * own pipeline's next timer nor any other pipeline's. The dispatch pool is elastic and reaps idle + * threads, so it is not a fixed cost per pipeline either. + * + * <p>Checkpoint work is intentionally not dispatched onto the coordinator executor: that pool is + * bounded over a {@code SynchronousQueue} and rejects work when saturated, which would silently + * drop a checkpoint trigger or a timeout watchdog. + * + * <p>One instance is held per member by {@code SeaTunnelEngineContext}. It is not a singleton + * because several members run in one JVM during tests and E2E runs. + */ +@Slf4j +public class SharedCheckpointScheduler implements AutoCloseable { + + /** + * Timer threads only dispatch, so a small fixed pool is enough regardless of pipeline count. + * Two threads keep one slow dispatch hand-off from stalling the other timers. + */ + private static final int TIMER_THREAD_NUM = 2; + + private final ScheduledThreadPoolExecutor timer; + private final ExecutorService dispatcher; + private volatile boolean closed = false; + + public SharedCheckpointScheduler() { + this.timer = + new ScheduledThreadPoolExecutor( + TIMER_THREAD_NUM, + new ThreadFactoryBuilder() + .setNameFormat("checkpoint-timer-%d") + .setDaemon(true) + .build()); + // Coordinators cancel the timeout watchdog on every acknowledged checkpoint. Without this, + // cancelled entries would sit in the shared queue until their delay elapsed. + this.timer.setRemoveOnCancelPolicy(true); + this.dispatcher = + Executors.newCachedThreadPool( + new ThreadFactoryBuilder() + .setNameFormat("checkpoint-dispatcher-%d") + .setDaemon(true) + .build()); + } + + /** + * Creates a timer lease for one pipeline. + * + * <p>The lease is the cancellation unit: it is identified by {@code (jobId, pipelineId)} and + * holds only its own outstanding tasks. Nothing is registered on this object, so dropping a + * lease when its job ends leaks nothing. + * + * @param jobId the owning job + * @param pipelineId the owning pipeline + * @return a lease that borrows this member's shared timer and dispatch threads + */ + public PipelineCheckpointScheduler lease(long jobId, int pipelineId) { + return new PipelineCheckpointScheduler(this, jobId, pipelineId); + } + + /** + * Arms {@code task} on the shared timer so that it is handed to the dispatch pool after {@code + * delay}. + * + * @return the timer-side future, or {@code null} if this scheduler is already shutting down + */ + ScheduledFuture<?> dispatchAfter(PipelineCheckpointTask task, long delay, TimeUnit unit) { + try { + return timer.schedule(() -> submit(task), delay, unit); + } catch (RejectedExecutionException e) { + // Only reachable once close() has begun; the coordinator is stopping anyway. + log.debug("Shared checkpoint scheduler is shut down, dropping timer task", e); + task.cancel(false); + return null; + } + } + + private void submit(PipelineCheckpointTask task) { + try { + dispatcher.execute(task); + } catch (RejectedExecutionException e) { + log.debug("Shared checkpoint dispatcher is shut down, dropping checkpoint task", e); Review Comment: Continuing from my reply above: in this specific path, "fails" means "the scheduler is already closing," and `task.cancel(false)` (line 111/121) marks the `PipelineCheckpointTask` cancelled and completes its internal future via `cancel()` (`PipelineCheckpointTask.java:82-93`) — genuinely cancelled, not exceptionally-failed. That's the correct terminal state for "we stopped because we're shutting down," and nothing downstream treats it as a checkpoint failure, because by the time `close()` runs, the owning `CheckpointCoordinator`/pipeline is also being torn down — there's no live consumer left to silently miss a checkpoint. There IS a real "silent no longer triggering checkpoint" risk in this design, but it's not in this rejection path — it's in the checkpoint body itself. `tryTriggerPendingCheckpoint()` (`CheckpointCoordinator.java:722`) is not wrapped in try/catch, and it's the same method that re-arms the next periodic trigger (`scheduleTriggerPendingCheckpoint(...)` at line 802) before returning. If it throws for an unrelated reason (bug, NPE, etc.) before reaching that line, the periodic loop breaks and checkpointing silently stops for that one pipeline. I checked whether this PR introduces or worsens that, and it doesn't — I diffed against the pre-PR code on `dev`, where `CheckpointCoordinator` called `scheduler.schedule(runnable, ...)` directly on a real `ScheduledThreadPoolExecutor`. That path is actually *more* silent than today's: `ScheduledThreadPoolExecutor.schedule()` wraps the runnable in an internal future-backed task whose `run()` catches `Throwable` and stores it via `setException()`, with zero stderr output, since the returned future was never queried. Today's `dispatcher.execute(task)` runs `PipelineCheckpointTask` as a plain `Runnable` (not future-wrapped), so an uncaught exception now propagates to the worker thread's default uncaught-exception handler, which prints a stack trace to stderr — strictly more visible than before, even though it's still not routed through SeaTunnel's own logger or any alert. This is a real, pre-existing gap — not something this PR needs to fix, and not something it makes worse — but I think it's worth a dedicated follow-up: wrap the scheduled body (or `PipelineCheckpointTask.run()`) with a catch that logs at `WARN`/`ERROR` through SeaTunnel's logger, since checkpoint triggering going silent is exactly the kind of failure an operator needs visibility into. I would not treat it as a blocker for this PR specifically, since it predates it and isn't made worse by it, but it's a legitimate item for a follow-up issue. ########## seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/scheduler/SharedCheckpointScheduler.java: ########## @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.engine.server.checkpoint.scheduler; + +import org.apache.seatunnel.shade.com.google.common.util.concurrent.ThreadFactoryBuilder; + +import lombok.extern.slf4j.Slf4j; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * Node-wide scheduler shared by every checkpoint coordinator on this member. + * + * <p>Before this existed, each {@code CheckpointCoordinator} built its own two-thread scheduled + * pool, so timer threads grew with the number of active pipelines. The coordinators only ever used + * that pool as a timer: to re-arm the periodic checkpoint trigger and to arm the checkpoint-timeout + * watchdog. This class keeps a single small timer pool for the whole member and hands each + * coordinator a {@link PipelineCheckpointScheduler} lease. + * + * <p>Timing and execution are deliberately split across two pools. A timer thread does nothing but + * hand the task to the dispatch pool, so a checkpoint body that blocks on an RPC delays neither its + * own pipeline's next timer nor any other pipeline's. The dispatch pool is elastic and reaps idle + * threads, so it is not a fixed cost per pipeline either. + * + * <p>Checkpoint work is intentionally not dispatched onto the coordinator executor: that pool is + * bounded over a {@code SynchronousQueue} and rejects work when saturated, which would silently + * drop a checkpoint trigger or a timeout watchdog. + * + * <p>One instance is held per member by {@code SeaTunnelEngineContext}. It is not a singleton + * because several members run in one JVM during tests and E2E runs. + */ +@Slf4j +public class SharedCheckpointScheduler implements AutoCloseable { + + /** + * Timer threads only dispatch, so a small fixed pool is enough regardless of pipeline count. + * Two threads keep one slow dispatch hand-off from stalling the other timers. + */ + private static final int TIMER_THREAD_NUM = 2; + + private final ScheduledThreadPoolExecutor timer; + private final ExecutorService dispatcher; + private volatile boolean closed = false; + + public SharedCheckpointScheduler() { + this.timer = + new ScheduledThreadPoolExecutor( + TIMER_THREAD_NUM, + new ThreadFactoryBuilder() + .setNameFormat("checkpoint-timer-%d") + .setDaemon(true) + .build()); + // Coordinators cancel the timeout watchdog on every acknowledged checkpoint. Without this, + // cancelled entries would sit in the shared queue until their delay elapsed. + this.timer.setRemoveOnCancelPolicy(true); + this.dispatcher = + Executors.newCachedThreadPool( + new ThreadFactoryBuilder() + .setNameFormat("checkpoint-dispatcher-%d") + .setDaemon(true) + .build()); + } + + /** + * Creates a timer lease for one pipeline. + * + * <p>The lease is the cancellation unit: it is identified by {@code (jobId, pipelineId)} and + * holds only its own outstanding tasks. Nothing is registered on this object, so dropping a + * lease when its job ends leaks nothing. + * + * @param jobId the owning job + * @param pipelineId the owning pipeline + * @return a lease that borrows this member's shared timer and dispatch threads + */ + public PipelineCheckpointScheduler lease(long jobId, int pipelineId) { + return new PipelineCheckpointScheduler(this, jobId, pipelineId); + } + + /** Review Comment: Good catch on the ambiguity — `SharedCheckpointScheduler` itself keeps no registry at all, so there is no code-level assertion that only one `PipelineCheckpointScheduler` can exist per `(jobId, pipelineId)`. `lease(jobId, pipelineId)` (lines 95-97) just returns a brand-new, unregistered object on every call; uniqueness is not enforced by this class. What actually makes this safe in practice is the calling discipline one layer up, which I traced end-to-end: - `CheckpointCoordinator.scheduler` is a `final` field, assigned exactly once in the constructor (`CheckpointCoordinator.java:155,229`) via `manager.leaseCheckpointScheduler(pipelineId)`. - `CheckpointManager` builds exactly one `CheckpointCoordinator` per `pipelineId` in its `coordinatorMap` (`CheckpointManager.java:132-181`, keyed by `CheckpointCoordinator::getPipelineId`), so exactly one lease is created per pipeline for that `CheckpointManager`'s lifetime. - Critically, on master-failover reset the SAME `CheckpointCoordinator` instance is reused: `cleanPendingCheckpoint(CHECKPOINT_COORDINATOR_RESET)` calls `scheduler.cancelAll()` (`CheckpointCoordinator.java:1196`) on the existing lease and flips `shutdown = false` — it does not call `leaseCheckpointScheduler()` again. So restart/failover doesn't create a second lease for the same `(jobId, pipelineId)`; it reuses the first one. So there's no "replace vs. leak vs. exception" branch to worry about, because re-registration never happens on the paths I found. A second lease for the same `(jobId, pipelineId)` would only appear if a brand-new `CheckpointManager` were constructed for the same pipeline while an old one was still alive — which would itself be a bug in the job-restart lifecycle outside this PR's scope, not something `SharedCheckpointScheduler` needs to guard against. And because leases hold no shared state (each just has its own local `outstanding` set, `PipelineCheckpointScheduler.java:45`), even a hypothetical duplicate lease wouldn't corrupt the shared pools — it would just be two independent cancellation domains, a correctness smell rather than a resource leak, since nothing is centrally registered to go stale. If you'd like a hard guarantee here (a defensive check-and-throw on duplicate lease, or an explicit registry with `dispose()`), I'd treat that as a legitimate hardening ask — but I don't see a live code path today that would actually trigger the failure mode you're describing. -- 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]
