public void close() {
        try {
            ioExecutor.shutdown();
            if (!ioExecutor.awaitTermination(5L, TimeUnit.MINUTES)) {
                throw new TimeoutException("Shutdown spilling thread timeout.");
            }
            dataFileChannel.close();
        } catch (Exception e) {
            ExceptionUtils.rethrow(e);
        }
    }
@Test
    void testClose() throws Exception {
        memoryDataSpiller = createMemoryDataSpiller(dataFilePath);
        List<BufferWithIdentity> bufferWithIdentityList =
                new ArrayList<>(
                        createBufferWithIdentityList(
                                false,
                                0,
                                Arrays.asList(Tuple2.of(0, 0), Tuple2.of(1, 1), Tuple2.of(2, 2))));
        memoryDataSpiller.spillAsync(bufferWithIdentityList);
        // blocked until spill finished.
        memoryDataSpiller.close();
        checkData(false, Arrays.asList(Tuple2.of(0, 0), Tuple2.of(1, 1), Tuple2.of(2, 2)));
        assertThatThrownBy(() -> memoryDataSpiller.spillAsync(bufferWithIdentityList))
                .isInstanceOf(RejectedExecutionException.class);
    }