public synchronized void close(long timeoutMs) throws Exception {
        final long startTime = System.currentTimeMillis();
        closed = true;
        fetchers.values().forEach(SplitFetcher::shutdown);
        // Actively drain the element queue in case there are previously shutting down
        // fetcher threads blocking on putting batches into the element queue.
        executors.submit(
                () -> {
                    while (fetchersToShutDown.get() > 0
                            && System.currentTimeMillis() - startTime < timeoutMs) {
                        elementsQueue
                                .getAvailabilityFuture()
                                .thenRun(() -> elementsQueue.poll().recycle());
                    }
                });
        executors.shutdown();
        if (!executors.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
            LOG.warn(
                    "Failed to close the split fetchers in {} ms. There are still {} split fetchers running",
                    timeoutMs,
                    fetchersToShutDown);
        }
    }
@Test
    public void testCloseFetcherWithException() throws Exception {
        TestingSplitReader<Object, TestingSourceSplit> reader = new TestingSplitReader<>();
        reader.setCloseWithException();
        SplitFetcherManager<Object, TestingSourceSplit> fetcherManager =
                createFetcher("test-split", reader, new Configuration());
        fetcherManager.close(1000L);
        assertThatThrownBy(fetcherManager::checkErrors)
                .hasRootCauseMessage("Artificial exception on closing the split reader.");
    }