mateczagany commented on code in PR #860:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/860#discussion_r1712595662
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SnapshotObserver.java:
##########
@@ -220,63 +235,179 @@ private void
observeTriggeredCheckpoint(FlinkResourceContext<CR> ctx, String job
/** Clean up and dispose savepoints according to the configured max
size/age. */
@VisibleForTesting
- void cleanupSavepointHistory(FlinkResourceContext<CR> ctx, SavepointInfo
currentSavepointInfo) {
+ void cleanupSavepointHistory(FlinkResourceContext<CR> ctx) {
+ if (!FlinkStateSnapshotUtils.isSnapshotResourceEnabled(
+ ctx.getOperatorConfig(), ctx.getObserveConfig())) {
+ cleanupSavepointHistoryLegacy(ctx);
+ return;
+ }
+
+ var snapshots =
ctx.getJosdkContext().getSecondaryResources(FlinkStateSnapshot.class);
+ if (CollectionUtil.isNullOrEmpty(snapshots)) {
+ return;
+ }
+ if (ctx.getObserveConfig().get(OPERATOR_SAVEPOINT_CLEANUP_ENABLED)) {
+ var savepointsToDelete =
+ getFlinkStateSnapshotsToCleanUp(
+ snapshots, ctx.getObserveConfig(),
ctx.getOperatorConfig(), SAVEPOINT);
+ var checkpointsToDelete =
+ getFlinkStateSnapshotsToCleanUp(
+ snapshots, ctx.getObserveConfig(),
ctx.getOperatorConfig(), CHECKPOINT);
+ Stream.concat(savepointsToDelete.stream(),
checkpointsToDelete.stream())
+ .forEach(
+ snapshot ->
+ ctx.getKubernetesClient()
+ .resource(snapshot)
+ .withTimeoutInMillis(0L)
+ .delete());
+ }
+ }
+
+ @VisibleForTesting
+ Set<FlinkStateSnapshot> getFlinkStateSnapshotsToCleanUp(
+ Collection<FlinkStateSnapshot> snapshots,
+ Configuration observeConfig,
+ FlinkOperatorConfiguration operatorConfig,
+ SnapshotType snapshotType) {
+ var snapshotList =
+ snapshots.stream()
+ .filter(s -> s.getStatus() != null)
+ .filter(s ->
COMPLETED.equals(s.getStatus().getState()))
+ .filter(
+ s ->
+ SnapshotTriggerType.PERIODIC.equals(
+
FlinkStateSnapshotUtils.getSnapshotTriggerType(s)))
Review Comment:
You are right. I will add a check to make sure that during cleanup the
latest COMPLETED snapshot won't be deleted. Then it should be safe to add
upgrade savepoints as well.
--
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]
