@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 testUploadTimeout() throws Exception {
        AtomicReference<List<SequenceNumber>> failed = new AtomicReference<>();
        UploadTask upload =
                new UploadTask(getChanges(4), unused -> {}, (sqn, error) -> failed.set(sqn));
        ManuallyTriggeredScheduledExecutorService executorService =
                new ManuallyTriggeredScheduledExecutorService();
        try (BatchingStateChangeUploadScheduler store =
                scheduler(1, executorService, new BlockingUploader(), 1)) {
            store.upload(upload);
            Deadline deadline = Deadline.fromNow(Duration.ofMinutes(5));
            while (!upload.finished.get() && deadline.hasTimeLeft()) {
                executorService.triggerScheduledTasks();
                executorService.triggerAll();
                Thread.sleep(10);
            }
        }

        assertThat(upload.finished.get()).isTrue();
        assertThat(
                        upload.changeSets.stream()
                                .map(StateChangeSet::getSequenceNumber)
                                .collect(Collectors.toSet()))
                .isEqualTo(new HashSet<>(failed.get()));
    }