[Bug] Memory was leaked by query when close sender before send WriteBatch.
- Dominant language
- Java
- Stars
- 2.1k
- Forks
- 625
- Avg merge
- 3d 14h
- Merged PRs (30d)
- 97
Description
### Search before asking
- [X] I searched in the [issues](https://github.com/alibaba/fluss/issues) and found nothing similar.
### Fluss version
0.5
### Minimal reproduce step
Add test to SenderTest
```java
@Test
void testRecycle() throws Exception {
long offset = 0;
CompletableFuture 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?
- [X] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with SenderTest's testRecycle reproduction, then trace Sender.forceClose through Sender.initiateClose and RecordAccumulator.close. Inspect ArrowLogWriteBatch ownership alongside MemoryLogRecordsArrowBuilder#trySerialize and arrowWriterPool#close; done means closing before send recycles the held arrowWriter without the memory-leak exception.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, data-engineering
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100