@Override
    public void close() throws Exception {
        LOG.debug("close");
        scheduler.shutdownNow();
        if (!scheduler.awaitTermination(5, SECONDS)) {
            LOG.warn("Unable to cleanly shutdown scheduler in 5s");
        }
        ArrayList<UploadTask> drained;
        synchronized (lock) {
            drained = new ArrayList<>(scheduled);
            scheduled.clear();
            scheduledBytesCounter = 0;
        }
        CancellationException ce = new CancellationException();
        drained.forEach(task -> task.fail(ce));
        retryingExecutor.close();
        delegate.close();
    }
@Test
    void testClose() throws Exception {
        TestingStateChangeUploader probe = new TestingStateChangeUploader();
        DirectScheduledExecutorService scheduler = new DirectScheduledExecutorService();
        DirectScheduledExecutorService retryScheduler = new DirectScheduledExecutorService();
        ChangelogStorageMetricGroup metrics = createUnregisteredChangelogStorageMetricGroup();
        new BatchingStateChangeUploadScheduler(
                        0,
                        0,
                        MAX_BYTES_IN_FLIGHT,
                        RetryPolicy.NONE,
                        probe,
                        scheduler,
                        new RetryingExecutor(
                                retryScheduler,
                                metrics.getAttemptsPerUpload(),
                                metrics.getTotalAttemptsPerUpload()),
                        metrics)
                .close();
        assertThat(probe.isClosed()).isTrue();
        assertThat(scheduler.isShutdown()).isTrue();
        assertThat(retryScheduler.isShutdown()).isTrue();
    }