@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 testInterruptedWhenBackPressured() throws Exception {
        int limit = MAX_BYTES_IN_FLIGHT;
        TestScenario test =
                (uploader, probe) -> {
                    List<StateChangeSet> changes = getChanges(limit + 1);
                    upload(uploader, changes);
                    assertSaved(probe, changes); // only sent for upload
                    probe.reset(); // don't complete the upload - so capacity isn't released
                    try {
                        upload(uploader, getChanges(1)); // should block
                        fail("upload shouldn't succeed after exceeding the limit");
                    } catch (IOException e) {
                        if (findThrowable(e, InterruptedException.class).isPresent()) {
                            assertThat(probe.getUploaded()).isEmpty();
                        } else {
                            rethrow(e);
                        }
                    }
                };

        Tuple2<Thread, CompletableFuture<Void>> threadAndFuture = uploadAsync(limit, test);
        Thread.sleep(500); // allow to upload (i.e. fail)
        threadAndFuture.f0.interrupt();
        threadAndFuture.f1.join();
    }