[client] Fix the IllegalStateException error that ArrowBuf refCount <= 0 (#1068)

Author: swuferhong

fluss-client/src/main/java/com/alibaba/fluss/client/write/ArrowLogWriteBatch.java

@@ -70,7 +70,7 @@ public boolean tryAppend(WriteRecord writeRecord, WriteCallback callback) throws
         checkArgument(writeRecord.getKey() == null, "key must be null for log record");
         checkNotNull(row != null, "row must not be null for log record");
         checkNotNull(callback, "write callback must be not null");
-        if (recordsBuilder.isFull() || recordsBuilder.isClosed()) {
+        if (recordsBuilder.isClosed() || recordsBuilder.isFull()) {
             return false;
         } else {
             recordsBuilder.append(ChangeType.APPEND_ONLY, row);

fluss-client/src/main/java/com/alibaba/fluss/client/write/RecordAccumulator.java

@@ -639,6 +639,11 @@ private RecordAppendResult tryAppend(
         if (last != null) {
             boolean success = last.tryAppend(writeRecord, callback);
             if (!success) {
+                // TODO For ArrowLogWriteBatch, close here is a heavy operation (including build
+                // logic), we need to avoid do that in an lock which locked dq. However, why we not
+                // remove build logic out of close for ArrowLogWriteBatch is that we want to release
+                // non-heap memory hold by arrowWriter as soon as possible to avoid OOM. Maybe we
+                // need to introduce a more reasonable way to solve these two problems.
                 last.close();
             } else {
                 return new RecordAppendResult(deque.size() > 1 || last.isClosed(), false, false);

fluss-common/src/main/java/com/alibaba/fluss/record/KvRecordBatchBuilder.java

@@ -53,7 +53,7 @@ public class KvRecordBatchBuilder implements AutoCloseable {
     private int batchSequence;
     private int currentRecordNumber;
     private int sizeInBytes;
-    private boolean isClosed;
+    private volatile boolean isClosed;
     private final KvFormat kvFormat;
     private boolean aborted = false;
 

fluss-common/src/main/java/com/alibaba/fluss/record/MemoryLogRecordsArrowBuilder.java

@@ -54,12 +54,13 @@ public class MemoryLogRecordsArrowBuilder implements AutoCloseable {
     private final AbstractPagedOutputView pagedOutputView;
     private final boolean appendOnly;
 
-    private MultiBytesView bytesView = null;
+    private volatile MultiBytesView bytesView = null;
+
     private long writerId;
     private int batchSequence;
     private int estimatedSizeInBytes;
     private int recordCount;
-    private boolean isClosed;
+    private volatile boolean isClosed;
     private boolean reCalculateSizeInBytes = false;
     private boolean resetBatchHeader = false;
     private boolean aborted = false;
@@ -216,10 +217,10 @@ public void close() throws Exception {
             return;
         }
 
+        isClosed = true;
+
         // Build arrowBatch when batch close to recycle arrow writer.
         build();
-
-        isClosed = true;
     }
 
     public void recycleArrowWriter() {

fluss-common/src/main/java/com/alibaba/fluss/record/MemoryLogRecordsIndexedBuilder.java

@@ -59,7 +59,7 @@ public class MemoryLogRecordsIndexedBuilder implements AutoCloseable {
     private int batchSequence;
     private int currentRecordNumber;
     private int sizeInBytes;
-    private boolean isClosed;
+    private volatile boolean isClosed;
     private boolean aborted = false;
 
     private MemoryLogRecordsIndexedBuilder(

fluss-common/src/main/java/com/alibaba/fluss/row/arrow/ArrowWriter.java

@@ -284,6 +284,7 @@ public void increaseEpoch() {
     public void recycle(long epoch) {
         if (this.epoch == epoch) {
             root.clear();
+            recordsCount = 0;
             provider.recycleWriter(this);
         }
     }