@Override
    public void release() {
        try {
            ioExecutor.shutdown();
            if (!ioExecutor.awaitTermination(5L, TimeUnit.MINUTES)) {
                throw new TimeoutException("Timeout to shutdown the flush thread.");
            }
            dataFileChannel.close();
        } catch (Exception e) {
            ExceptionUtils.rethrow(e);
        }
        partitionFileIndex.release();
    }
@Test
    void testRelease() {
        AtomicBoolean isReleased = new AtomicBoolean(false);
        TestingProducerMergedPartitionFileIndex partitionFileIndex =
                new TestingProducerMergedPartitionFileIndex.Builder()
                        .setIndexFilePath(new File(tempFolder.toFile(), "testIndex").toPath())
                        .setReleaseRunnable(() -> isReleased.set(true))
                        .build();
        ProducerMergedPartitionFileWriter partitionFileWriter =
                new ProducerMergedPartitionFileWriter(
                        new File(tempFolder.toFile(), "testFile1").toPath(), partitionFileIndex);
        partitionFileWriter.release();
        assertThat(isReleased).isTrue();
    }