Skip to content

[Fix][Zeta] Fix losing checkpoint scheduling in extreme cases #9246

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
corgy-w merged 9 commits into from
May 13, 2025
Merged

[Fix][Zeta] Fix losing checkpoint scheduling in extreme cases #9246

Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
7c96b26
[Fix][Zeta] Fix losing checkpoint scheduling in extreme cases
Hisoka-X Apr 28, 2025
65d93f4
Update seatunnel-engine/seatunnel-engine-server/src/main/java/org/apa…
Hisoka-X Apr 29, 2025
419687f
update
Hisoka-X Apr 29, 2025
a2a9d6d
update
Hisoka-X Apr 30, 2025
b8cc850
update
Hisoka-X Apr 30, 2025
9306290
update
Hisoka-X May 6, 2025
ca3658b
update
Hisoka-X May 6, 2025
283797e
Merge branch 'refs/heads/dev' into fix-lose-checkpoint
Hisoka-X May 7, 2025
3e5723b
Merge branch 'refs/heads/dev' into fix-lose-checkpoint
Hisoka-X May 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 25 additions & 12 deletions ...er/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -470,7 +470,7 @@ protected void completedCloseIdleTask(TaskLocation taskLocation) {
}

protected void restoreCoordinator(boolean alreadyStarted) {
LOG.info("received restore CheckpointCoordinator with alreadyStarted= " + alreadyStarted);
LOG.info("received restore CheckpointCoordinator with alreadyStarted = {}", alreadyStarted);
errorByPhysicalVertex = new AtomicReference<>();
checkpointCoordinatorFuture = new CompletableFuture<>();
updateStatus(CheckpointCoordinatorStatus.RUNNING);
Expand All @@ -492,15 +492,24 @@ protected void tryTriggerPendingCheckpoint(CheckpointType checkpointType) {
}
final long currentTimestamp = Instant.now().toEpochMilli();
if (checkpointType.notFinalCheckpoint() && checkpointType.notSchemaChangeCheckpoint()) {
long diffFromLastTimestamp = currentTimestamp - latestTriggerTimestamp.get();
if (diffFromLastTimestamp <= 0) {
if (!isAllTaskReady) {
LOG.info("The all task not ready, skip trigger checkpoint");
return;
}
long interval = currentTimestamp - latestTriggerTimestamp.get();
if (interval <= 0) {
LOG.error(
"The time on your server may not be incremental which can lead checkpoint to stop. The latestTriggerTimestamp: ({}), but the currentTimestamp: ({})",
latestTriggerTimestamp.get(),
currentTimestamp);
}
if (diffFromLastTimestamp < coordinatorConfig.getCheckpointInterval()
|| !isAllTaskReady) {
if (interval < coordinatorConfig.getCheckpointInterval()) {
LOG.info(
"skip trigger checkpoint because the last trigger timestamp is {} and current timestamp is {}, the interval is less than config.",
latestTriggerTimestamp.get(),
currentTimestamp);
scheduleTriggerPendingCheckpoint(
checkpointType, coordinatorConfig.getCheckpointInterval() - interval);
return;
}
}
Expand Down Expand Up @@ -534,6 +543,10 @@ protected void tryTriggerPendingCheckpoint(CheckpointType checkpointType) {
// if checkpoint type are final type, we don't need to trigger next checkpoint
if (checkpointType.notFinalCheckpoint() && checkpointType.notSchemaChangeCheckpoint()) {
scheduleTriggerPendingCheckpoint(coordinatorConfig.getCheckpointInterval());
} else {
LOG.info(
"skip schedule trigger checkpoint because checkpoint type is {}",
checkpointType);
}
}
}
Expand Down Expand Up @@ -596,7 +609,7 @@ private void startTriggerPendingCheckpoint(
CompletableFuture<PendingCheckpoint> pendingCompletableFuture) {
pendingCompletableFuture.thenAccept(
pendingCheckpoint -> {
LOG.info("wait checkpoint completed: " + pendingCheckpoint.getCheckpointId());
LOG.info("wait checkpoint completed: {}", pendingCheckpoint.getCheckpointId());
PassiveCompletableFuture<CompletedCheckpoint> completableFuture =
pendingCheckpoint.getCompletableFuture();
completableFuture.whenCompleteAsync(
Expand Down Expand Up @@ -647,8 +660,8 @@ private void startTriggerPendingCheckpoint(
}
if (coordinatorConfig.isCheckpointEnable()) {
LOG.debug(
"Start a scheduled task to prevent checkpoint timeouts for barrier "
+ pendingCheckpoint.getInfo());
"Start a scheduled task to prevent checkpoint timeouts for barrier {}",
pendingCheckpoint.getInfo());
long checkpointTimeout = coordinatorConfig.getCheckpointTimeout();
if (pendingCheckpoint.getCheckpointType().isSchemaChangeAfterCheckpoint()) {
checkpointTimeout =
Expand All @@ -665,8 +678,8 @@ private void startTriggerPendingCheckpoint(
!= null
&& !pendingCheckpoint.isFullyAcknowledged()) {
LOG.info(
"timeout checkpoint: "
+ pendingCheckpoint.getInfo());
"timeout checkpoint: {}",
pendingCheckpoint.getInfo());
handleCoordinatorError(
CheckpointCloseReason.CHECKPOINT_EXPIRED,
null);
Expand Down Expand Up @@ -817,7 +830,7 @@ protected void acknowledgeTask(TaskAcknowledgeOperation ackOperation) {
final long checkpointId = ackOperation.getBarrier().getId();
final PendingCheckpoint pendingCheckpoint = pendingCheckpoints.get(checkpointId);
if (pendingCheckpoint == null) {
LOG.info("skip already ack checkpoint " + checkpointId);
LOG.info("skip already ack checkpoint {}", checkpointId);
return;
}
TaskLocation location = ackOperation.getTaskLocation();
Expand Down Expand Up @@ -995,7 +1008,7 @@ private synchronized void updateStatus(@NonNull CheckpointCoordinatorStatus targ
new RetryUtils.RetryMaterial(
Constant.OPERATION_RETRY_TIME,
true,
exception -> ExceptionUtil.isOperationNeedRetryException(exception),
ExceptionUtil::isOperationNeedRetryException,
Constant.OPERATION_RETRY_SLEEP));
} catch (Exception e) {
LOG.warn(
Expand Down