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 triggerSavepointRepeatedly() throws ExecutionException, InterruptedException {
        CompletableFuture<Acknowledge> firstAcknowledge =
                handler.triggerSavepoint(
                        operationKey,
                        targetDirectory,
                        SavepointFormatType.CANONICAL,
                        TriggerSavepointMode.SAVEPOINT,
                        TIMEOUT);
        CompletableFuture<Acknowledge> secondAcknowledge =
                handler.triggerSavepoint(
                        operationKey,
                        targetDirectory,
                        SavepointFormatType.CANONICAL,
                        TriggerSavepointMode.SAVEPOINT,
                        TIMEOUT);

        assertThat(triggerSavepointFunction.getNumberOfInvocations(), is(1));
        assertThat(
                triggerSavepointFunction.getInvocationParameters().get(0),
                is(
                        new Tuple4<>(
                                jobID,
                                targetDirectory,
                                SavepointFormatType.CANONICAL,
                                TriggerSavepointMode.SAVEPOINT)));

        assertThat(firstAcknowledge.get(), is(Acknowledge.get()));
        assertThat(secondAcknowledge.get(), is(Acknowledge.get()));
    }