<T> void execute(RetryPolicy retryPolicy, RetriableAction<T> action) {
        LOG.debug("execute with retryPolicy: {}", retryPolicy);
        RetriableActionAttempt<T> task =
                RetriableActionAttempt.initialize(
                        action,
                        retryPolicy,
                        blockingExecutor,
                        attemptsPerTaskHistogram,
                        totalAttemptsPerTaskHistogram,
                        timer);
        blockingExecutor.submit(task);
    }
@Test
    void testDiscardOnTimeout() throws Exception {
        int timeoutMs = 50; // should be long enough for the last attempt to succeed
        int numAttempts = 7;
        int successfulAttempt = numAttempts - 1;
        List<Integer> completed = new CopyOnWriteArrayList<>();
        List<Integer> discarded = new CopyOnWriteArrayList<>();
        AtomicBoolean executionBlocked = new AtomicBoolean(true);
        AtomicReference<Throwable> unexpectedException = new AtomicReference<>();
        Deadline deadline = Deadline.fromNow(Duration.ofMinutes(5));
        ChangelogStorageMetricGroup metrics = createUnregisteredChangelogStorageMetricGroup();
        try (RetryingExecutor executor =
                new RetryingExecutor(
                        numAttempts,
                        metrics.getAttemptsPerUpload(),
                        metrics.getTotalAttemptsPerUpload())) {
            executor.execute(
                    RetryPolicy.fixed(numAttempts, timeoutMs, 0),
                    new RetriableAction<Integer>() {
                        private final AtomicInteger attemptsCounter = new AtomicInteger(0);

                        @Override
                        public Integer tryExecute() throws Exception {
                            int attempt = attemptsCounter.getAndIncrement();
                            if (attempt < successfulAttempt) {
                                while (executionBlocked.get()) {
                                    Thread.sleep(10);
                                }
                            }
                            return attempt;
                        }

                        @Override
                        public void completeWithResult(Integer result) {
                            completed.add(result);
                        }

                        @Override
                        public void discardResult(Integer result) {
                            discarded.add(result);
                        }

                        @Override
                        public void handleFailure(Throwable throwable) {
                            executionBlocked.set(false);
                            unexpectedException.set(throwable);
                        }
                    });
            while (completed.isEmpty() && deadline.hasTimeLeft()) {
                Thread.sleep(10);
            }
            executionBlocked.set(false);
            while (discarded.size() < successfulAttempt && deadline.hasTimeLeft()) {
                Thread.sleep(10);
            }
        }
        if (unexpectedException.get() != null) {
            // the last attempt might still timeout if the worker node is overloaded
            // and the thread is unscheduled for more than timeoutMs
            assertThat(unexpectedException).isInstanceOf(TimeoutException.class);
        }
        assertThat(singletonList(successfulAttempt)).isEqualTo(completed);
        assertThat(IntStream.range(0, successfulAttempt).boxed().collect(toList()))
                .isEqualTo(discarded.stream().sorted().collect(toList()));
    }