[Bug] Memory was leaked by query when close sender before send WriteBatch.

[loserwang1024]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

0.5

Minimal reproduce step

Add test to SenderTest

    @Test
    void testRecycle() throws Exception {
        long offset = 0;
        CompletableFuture<Exception> future = new CompletableFuture<>();
        appendToAccumulator(tb1, row(DATA1_ROW_TYPE, new Object[] {1, "a"}), future::complete);
        sender.forceClose();
       // sender.runOnce();
}

What doesn't meet your expectations?

Throw exception:

java.lang.IllegalStateException: Memory was leaked by query. Memory leaked: (24576)
Allocator(ROOT) 0/24576/24576/9223372036854775807 (res/actual/peak/limit)


	at com.alibaba.fluss.shaded.arrow.org.apache.arrow.memory.BaseAllocator.close(BaseAllocator.java:477)
	at com.alibaba.fluss.shaded.arrow.org.apache.arrow.memory.RootAllocator.close(RootAllocator.java:29)
	at com.alibaba.fluss.client.write.RecordAccumulator.close(RecordAccumulator.java:908)
	at com.alibaba.fluss.client.write.Sender.initiateClose(Sender.java:539)
	at com.alibaba.fluss.client.write.Sender.forceClose(Sender.java:532)
	at com.alibaba.fluss.client.write.SenderTest.testSimple(SenderTest.java:99)

Anything else?

Currently, the arrowWriter is recycle in the following situations:

1. when Sender#sendWriteData, invokes com.alibaba.fluss.record.MemoryLogRecordsArrowBuilder#trySerialize , the ArrowLogWriteBatch will arrowWriter.recycle(writerEpoch) of ArrowLogWriteBatch.
2. when Sender is closed, sender#initiateClose -> accumulator#close ->arrowWriterPool#close will recycle the free arrowWriter.

It means that if close before Sender#sendWriteData, the non-free( hold by ArrowLogWriteBatch) arrowWriter won't be recycled.

Are you willing to submit a PR?

• I'm willing to submit a PR!