[client] Recycle ArrowWriter when MemoryLogRecordsArrowBuilder closed instead of built (#973)

Author: swuferhong

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

@@ -119,7 +119,7 @@ public void setWriterState(long writerId, int batchSequence) {
     @Override
     public void resetWriterState(long writerId, int batchSequence) {
         super.resetWriterState(writerId, batchSequence);
-        recordsBuilder.resetWriterState(writerId, batchSequence);
+        recordsBuilder.setWriterState(writerId, batchSequence);
     }
 
     @Override

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

@@ -190,15 +190,10 @@ public int batchSequence() {
     }
 
     public void setWriterState(long writerId, int batchBaseSequence) {
-        this.writerId = writerId;
-        this.batchSequence = batchBaseSequence;
-    }
-
-    public void resetWriterState(long writerId, int batchSequence) {
         // trigger to rewrite batch header when next build.
         this.resetBatchHeader = true;
         this.writerId = writerId;
-        this.batchSequence = batchSequence;
+        this.batchSequence = batchBaseSequence;
     }
 
     public void abort() {
@@ -221,6 +216,9 @@ public void close() throws Exception {
             return;
         }
 
+        // Build arrowBatch when batch close to recycle arrow writer.
+        build();
+
         isClosed = true;
     }
 

fluss-common/src/test/java/com/alibaba/fluss/record/MemoryLogRecordsArrowBuilderTest.java

@@ -327,7 +327,7 @@ void testResetWriterState() throws Exception {
 
         // test reset writer state and build (This situation will happen when the produceLog request
         // failed and the batch is re-enqueue to send with different write state).
-        builder.resetWriterState(1L, 1);
+        builder.setWriterState(1L, 1);
         records = MemoryLogRecords.pointToBytesView(builder.build());
         assertLogRecordsEquals(DATA1_ROW_TYPE, records, expectedResult);
         recordBatch = records.batches().iterator().next();