Vivek1106-04 commented on code in PR #12165: URL: https://github.com/apache/seatunnel/pull/12165#discussion_r4071701160
########## 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; Review Comment: @nzw921rx Late reply from me on this one, and the code has moved since, so here is where it stands. The dispatcher is no longer unbounded. Since `b0bc24603` it is a `ThreadPoolExecutor` with core equal to maximum at `max(8, availableProcessors * 2)`, an unbounded `LinkedBlockingQueue`, and `allowCoreThreadTimeOut(true)` - so it grows on demand, reaps idle threads, is bounded in thread count, and never rejects. The reason it cannot use a bounded queue is that a rejection here is a dropped checkpoint trigger or timeout watchdog, which is a correctness problem rather than a resource one, so the bound is paid for in delay instead. On making it configurable, which was your actual ask: I measured it rather than argued it. #12418 adds a scheduling-delay benchmark, and the comparison is on #12339. The short version is that scheduling delay is flat at about 10 us from 1 to 500 pipelines here against a knee between 100 and 500 for the per-pipeline model, and flat through 31% dispatch utilization - but past saturation it collapses to a p50 of 461 ms, so the pool size does matter in a regime the default interval does not reach. Both @DanielLeens and @SEZ9 read that as: no `checkpoint.scheduler-dispatch-thread-num` option in this STIP, and scheduling-delay observability ahead of any tuning knob, since the metric is what would tell an operator they are approaching the regime where the option would help. I have taken that as the decision for this PR. If you want the option anyway, I would rather add it on the back of that metric than on its own, and I am happy to do both as a follow-up. -- 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]
