@Override
    public void upload(UploadTask uploadTask) throws IOException {
        Throwable error = getErrorSafe();
        if (error != null) {
            LOG.debug("don't persist {} changesets, already failed", uploadTask.changeSets.size());
            uploadTask.fail(error);
            return;
        }
        LOG.debug("persist {} changeSets", uploadTask.changeSets.size());
        try {
            long size = uploadTask.getSize();
            synchronized (lock) {
                while (!uploadThrottle.hasCapacity()) {
                    lock.wait();
                }
                uploadThrottle.seizeCapacity(size);
                if (!uploadThrottle.hasCapacity()) {
                    availabilityHelper.resetUnavailable();
                }
                scheduledBytesCounter += size;
                scheduled.add(wrapWithSizeUpdate(uploadTask, size));
                scheduleUploadIfNeeded();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            uploadTask.fail(e);
            throw new IOException(e);
        } catch (Exception e) {
            uploadTask.fail(e);
            throw e;
        }
    }
@Test
    void testRetry() throws Exception {
        final int maxAttempts = 5;
        ChangelogStorageMetricGroup metrics = createUnregisteredChangelogStorageMetricGroup();

        try (BatchingStateChangeUploadScheduler store =
                new BatchingStateChangeUploadScheduler(
                        0,
                        0,
                        MAX_BYTES_IN_FLIGHT,
                        RetryPolicy.fixed(maxAttempts, 0, 0),
                        new TestingStateChangeUploader() {
                            final AtomicInteger currentAttempt = new AtomicInteger(0);

                            @Override
                            public UploadTasksResult upload(Collection<UploadTask> tasks)
                                    throws IOException {
                                for (UploadTask uploadTask : tasks) {
                                    if (currentAttempt.getAndIncrement() < maxAttempts - 1) {
                                        throw new IOException();
                                    } else {
                                        uploadTask.complete(emptyList());
                                    }
                                }
                                return null;
                            }
                        },
                        new DirectScheduledExecutorService(),
                        new RetryingExecutor(
                                new DirectScheduledExecutorService(),
                                metrics.getAttemptsPerUpload(),
                                metrics.getTotalAttemptsPerUpload()),
                        metrics)) {
            CompletableFuture<List<UploadResult>> completionFuture = new CompletableFuture<>();
            store.upload(
                    new UploadTask(
                            getChanges(4),
                            completionFuture::complete,
                            (unused, throwable) ->
                                    completionFuture.completeExceptionally(throwable)));
            completionFuture.get();
        }
    }