public CompletableFuture<Acknowledge> triggerSavepoint(
            AsynchronousJobOperationKey operationKey,
            String targetDirectory,
            SavepointFormatType formatType,
            TriggerSavepointMode savepointMode,
            Time timeout) {
        return registerOperationIdempotently(
                operationKey,
                () ->
                        triggerSavepointFunction.apply(
                                operationKey.getJobId(),
                                targetDirectory,
                                formatType,
                                savepointMode,
                                timeout));
    }
@Test
    public void retryingCompletedOperationDoesNotMarkCacheEntryAsAccessed()
            throws ExecutionException, InterruptedException {
        handler.triggerSavepoint(
                        operationKey,
                        targetDirectory,
                        SavepointFormatType.CANONICAL,
                        TriggerSavepointMode.SAVEPOINT,
                        TIMEOUT)
                .get();
        savepointLocationFuture.complete("");

        handler.triggerSavepoint(
                        operationKey,
                        targetDirectory,
                        SavepointFormatType.CANONICAL,
                        TriggerSavepointMode.SAVEPOINT,
                        TIMEOUT)
                .get();

        // should not complete because we wait for the result to be accessed
        assertThat(
                savepointTriggerCache.closeAsync(),
                FlinkMatchers.willNotComplete(Duration.ofMillis(10)));
    }