[Fix](batch-stream-load) handle types that do not have line delimiters (#619)

Author: AlexRiedler

flink-doris-connector/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java

@@ -380,7 +380,7 @@ private boolean merge(BatchRecordBuffer mergeBuffer, BatchRecordBuffer buffer) {
         if (buffer.getBuffer().isEmpty()) {
             return false;
         }
-        if (!mergeBuffer.getBuffer().isEmpty()) {
+        if (!mergeBuffer.getBuffer().isEmpty() && mergeBuffer.getLineDelimiter() != null) {
             mergeBuffer.getBuffer().add(mergeBuffer.getLineDelimiter());
             mergeBuffer.setBufferSizeBytes(
                     mergeBuffer.getBufferSizeBytes() + mergeBuffer.getLineDelimiter().length);

flink-doris-connector/src/test/java/org/apache/doris/flink/sink/batch/TestDorisBatchStreamLoad.java

@@ -26,6 +26,7 @@
 import org.apache.doris.flink.sink.HttpTestUtil;
 import org.apache.doris.flink.sink.TestUtil;
 import org.apache.doris.flink.sink.writer.LabelGenerator;
+import org.apache.doris.flink.sink.writer.LoadConstants;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClientBuilder;
@@ -45,6 +46,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Properties;
 
 import static org.apache.doris.flink.sink.batch.TestBatchBufferStream.mergeByteArrays;
 import static org.mockito.ArgumentMatchers.any;
@@ -226,4 +228,53 @@ public void mergeBufferTest() {
         flag = loader.mergeBuffer(bufferList, buffer);
         Assert.assertEquals(false, flag);
     }
+
+    @Test
+    public void mergeBufferNullDelimiterTest() {
+        DorisReadOptions readOptions = DorisReadOptions.builder().build();
+        Properties streamProperties = new Properties();
+        streamProperties.setProperty(
+                LoadConstants.FORMAT_KEY, LoadConstants.ARROW); // this makes lineDelimiter null
+        DorisExecutionOptions executionOptions =
+                DorisExecutionOptions.builder().setStreamLoadProp(streamProperties).build();
+        DorisOptions options =
+                DorisOptions.builder()
+                        .setFenodes("127.0.0.1:8030")
+                        .setBenodes("127.0.0.1:9030")
+                        .setTableIdentifier("db.tbl")
+                        .build();
+
+        DorisBatchStreamLoad loader =
+                new DorisBatchStreamLoad(
+                        options, readOptions, executionOptions, new LabelGenerator("xx", false), 0);
+
+        List<BatchRecordBuffer> bufferList = new ArrayList<>();
+        BatchRecordBuffer recordBuffer = new BatchRecordBuffer("db", "tbl", null, 0);
+        recordBuffer.insert("111".getBytes(StandardCharsets.UTF_8));
+        recordBuffer.setLabelName("label2");
+        BatchRecordBuffer buffer = new BatchRecordBuffer("db", "tbl", null, 0);
+        buffer.insert("222".getBytes(StandardCharsets.UTF_8));
+        buffer.setLabelName("label1");
+
+        boolean flag = loader.mergeBuffer(bufferList, buffer);
+        Assert.assertEquals(false, flag);
+
+        bufferList.add(buffer);
+        bufferList.add(recordBuffer);
+        flag = loader.mergeBuffer(bufferList, buffer);
+        Assert.assertEquals(true, flag);
+        byte[] bytes = mergeByteArrays(buffer.getBuffer());
+        Assert.assertArrayEquals(bytes, "222111".getBytes(StandardCharsets.UTF_8));
+
+        // multi table
+        bufferList.clear();
+        bufferList.add(buffer);
+        BatchRecordBuffer recordBuffer2 =
+                new BatchRecordBuffer("db", "tbl2", "\n".getBytes(StandardCharsets.UTF_8), 0);
+        recordBuffer2.insert("333".getBytes(StandardCharsets.UTF_8));
+        recordBuffer2.setLabelName("label3");
+        bufferList.add(recordBuffer2);
+        flag = loader.mergeBuffer(bufferList, buffer);
+        Assert.assertEquals(false, flag);
+    }
 }