[client] release the resource of record accumulator only when sender thread is stopped.

[loserwang1024]

Purpose

Linked issue: close #3101

Brief change log

Tests

API and Format

Documentation

[utafrali]

The core fix is correct: splitting close() (reject appends) from destroyResources() (free Arrow resources) and calling the latter only after all drain loops have finished cleanly resolves the 'Accounted size went negative' race. The main issues are an unused public getter that leaks internal state, missing lifecycle documentation on Sender.destroyResources() that hides a latent double-close risk, and a regression test with no assertion.

[loserwang1024]

@platinumhamburg ,CC

[platinumhamburg]

volatile guarantees visibility but does not guarantee atomicity of the check + set. Two threads can both read false, pass the guard, and both enter the destruction path. Arrow's BufferAllocator.close() is not idempotent and will throw IllegalStateException on a second invocation.

The Javadoc claims "This method is idempotent: subsequent calls after the first are no-ops", but the implementation cannot honor this guarantee under concurrent access.

Suggested fix:

private final AtomicBoolean resourcesDestroyed = new AtomicBoolean(false);

public void destroyResources() {
    if (!resourcesDestroyed.compareAndSet(false, true)) {
        return;
    }
    writerBufferPool.close();
    arrowWriterPool.close();
    bufferAllocator.close();
    chunkedFactory.close();
}

[platinumhamburg]

Thanks @loserwang1024, I left some comments.

[platinumhamburg]

Missing @VisibleForTesting