Skip to content
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

[Bugfix][zeta] Fix memory leak issues related to checkpoints #5539

Merged
merged 4 commits into from
Sep 25, 2023
Merged

[Bugfix][zeta] Fix memory leak issues related to checkpoints #5539

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8272a98
[bugfix][zeta] Fix the issue of generating a large number of pending …
ic4y Sep 21, 2023
b5e12c7
[bugfix][zeta] Avoid memory leak in ScheduledThreadPoolExecutor due t…
ic4y Sep 21, 2023
476d8ba
fix
ic4y Sep 21, 2023
7456f00
[Improve][zeta] Optimize memory usage of completedCheckpoints
ic4y Sep 22, 2023
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
60 changes: 33 additions & 27 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 @@ -64,6 +64,7 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
Expand Down Expand Up @@ -111,7 +112,7 @@ public class CheckpointCoordinator {
private final Set<TaskLocation> readyToCloseStartingTask;
private final ConcurrentHashMap<Long, PendingCheckpoint> pendingCheckpoints;

private final ArrayDeque<CompletedCheckpoint> completedCheckpoints;
private final ArrayDeque<String> completedCheckpointIds;

private volatile CompletedCheckpoint latestCompletedCheckpoint = null;

Expand Down Expand Up @@ -165,7 +166,7 @@ public CheckpointCoordinator(
this.plan = plan;
this.coordinatorConfig = checkpointConfig;
this.pendingCheckpoints = new ConcurrentHashMap<>();
this.completedCheckpoints =
this.completedCheckpointIds =
new ArrayDeque<>(coordinatorConfig.getStorage().getMaxRetainedCheckpoints() + 1);
this.scheduler =
Executors.newScheduledThreadPool(
Expand All @@ -178,6 +179,7 @@ public CheckpointCoordinator(
"checkpoint-coordinator-%s/%s", pipelineId, jobId));
return thread;
});
((ScheduledThreadPoolExecutor) this.scheduler).setRemoveOnCancelPolicy(true);
this.serializer = new ProtoStuffSerializer();
this.pipelineTasks = getPipelineTasks(plan.getPipelineSubtasks());
this.pipelineTaskStatus = new ConcurrentHashMap<>();
Expand Down Expand Up @@ -406,18 +408,18 @@ protected void tryTriggerPendingCheckpoint(CheckpointType checkpointType) {
shutdown));
return;
}
if (checkpointType.isFinalCheckpoint() || checkpointType.isSchemaChangeCheckpoint()) {
if (pendingCounter.get() > 0) {
scheduleTriggerPendingCheckpoint(checkpointType, 500L);
return;
}
}

if (schemaChanging.get() && checkpointType.isGeneralCheckpoint()) {
LOG.info("skip trigger generic-checkpoint because schema change in progress");
return;
}

if (pendingCounter.get() > 0) {
scheduleTriggerPendingCheckpoint(checkpointType, 500L);
LOG.info("skip trigger checkpoint because there is already a pending checkpoint.");
return;
}

CompletableFuture<PendingCheckpoint> pendingCheckpoint =
createPendingCheckpoint(currentTimestamp, checkpointType);
startTriggerPendingCheckpoint(pendingCheckpoint);
Expand Down Expand Up @@ -522,19 +524,24 @@ private void startTriggerPendingCheckpoint(
checkpointTimeout = coordinatorConfig.getSchemaChangeCheckpointTimeout();
}
// TODO Need change to polling check until max timeout fails
scheduler.schedule(
() -> {
// If any task is not acked within the checkpoint timeout
if (pendingCheckpoints.get(pendingCheckpoint.getCheckpointId())
!= null
&& !pendingCheckpoint.isFullyAcknowledged()) {
LOG.info("timeout checkpoint: " + pendingCheckpoint.getInfo());
handleCoordinatorError(
CheckpointCloseReason.CHECKPOINT_EXPIRED, null);
}
},
checkpointTimeout,
TimeUnit.MILLISECONDS);
pendingCheckpoint.setCheckpointTimeOutFuture(
scheduler.schedule(
() -> {
// If any task is not acked within the checkpoint
// timeout
if (pendingCheckpoints.get(
pendingCheckpoint.getCheckpointId())
!= null
&& !pendingCheckpoint.isFullyAcknowledged()) {
LOG.info(
"timeout checkpoint: "
+ pendingCheckpoint.getInfo());
handleCoordinatorError(
CheckpointCloseReason.CHECKPOINT_EXPIRED, null);
}
},
checkpointTimeout,
TimeUnit.MILLISECONDS));
});
}

Expand Down Expand Up @@ -695,7 +702,7 @@ public synchronized void completePendingCheckpoint(CompletedCheckpoint completed
completedCheckpoint.getCheckpointTimestamp(),
completedCheckpoint.getCompletedTimestamp());
final long checkpointId = completedCheckpoint.getCheckpointId();
completedCheckpoints.addLast(completedCheckpoint);
completedCheckpointIds.addLast(String.valueOf(completedCheckpoint.getCheckpointId()));
try {
byte[] states = serializer.serialize(completedCheckpoint);
checkpointStorage.storeCheckPoint(
Expand All @@ -705,18 +712,17 @@ public synchronized void completePendingCheckpoint(CompletedCheckpoint completed
.pipelineId(pipelineId)
.states(states)
.build());
if (completedCheckpoints.size()
if (completedCheckpointIds.size()
% coordinatorConfig.getStorage().getMaxRetainedCheckpoints()
== 0
&& completedCheckpoints.size()
&& completedCheckpointIds.size()
/ coordinatorConfig.getStorage().getMaxRetainedCheckpoints()
> 1) {
List<String> needDeleteCheckpointId = new ArrayList<>();
for (int i = 0;
i < coordinatorConfig.getStorage().getMaxRetainedCheckpoints();
i++) {
needDeleteCheckpointId.add(
completedCheckpoints.removeFirst().getCheckpointId() + "");
needDeleteCheckpointId.add(completedCheckpointIds.removeFirst());
}
checkpointStorage.deleteCheckpoint(
String.valueOf(completedCheckpoint.getJobId()),
Expand All @@ -734,7 +740,7 @@ public synchronized void completePendingCheckpoint(CompletedCheckpoint completed
completedCheckpoint.getJobId());
latestCompletedCheckpoint = completedCheckpoint;
notifyCompleted(completedCheckpoint);
pendingCheckpoints.remove(checkpointId);
pendingCheckpoints.remove(checkpointId).abortCheckpointTimeoutFutureWhenIsCompleted();
pendingCounter.decrementAndGet();
if (isCompleted()) {
cleanPendingCheckpoint(CheckpointCloseReason.CHECKPOINT_COORDINATOR_COMPLETED);
Expand Down
13 changes: 13 additions & 0 deletions ...server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/PendingCheckpoint.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,13 +27,15 @@

import com.beust.jcommander.internal.Nullable;
import lombok.Getter;
import lombok.Setter;

import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledFuture;

public class PendingCheckpoint implements Checkpoint {
private static final Logger LOG = LoggerFactory.getLogger(PendingCheckpoint.class);
Expand All @@ -57,6 +59,8 @@ public class PendingCheckpoint implements Checkpoint {

@Getter private CheckpointException failureCause;

@Setter ScheduledFuture<?> checkpointTimeOutFuture;

public PendingCheckpoint(
long jobId,
int pipelineId,
Expand Down Expand Up @@ -174,6 +178,15 @@ public void abortCheckpoint(CheckpointCloseReason closedReason, @Nullable Throwa
}
}

// Avoid memory leak in ScheduledThreadPoolExecutor due to overly long timeout settings causing
// numerous completed checkpoints to remain
public void abortCheckpointTimeoutFutureWhenIsCompleted() {
if (checkpointTimeOutFuture == null) {
return;
}
checkpointTimeOutFuture.cancel(false);
}

public String getInfo() {
return String.format(
"%s/%s/%s, %s",
Expand Down