update

flink-doris-connector/src/main/java/org/apache/doris/flink/cfg/DorisExecutionOptions.java

@@ -40,8 +40,8 @@ public class DorisExecutionOptions implements Serializable {
     private static final int DEFAULT_BUFFER_COUNT = 3;
     // batch flush
     private static final int DEFAULT_FLUSH_QUEUE_SIZE = 2;
-    private static final int DEFAULT_BUFFER_FLUSH_MAX_ROWS = 50000;
-    private static final int DEFAULT_BUFFER_FLUSH_MAX_BYTES = 10 * 1024 * 1024;
+    private static final int DEFAULT_BUFFER_FLUSH_MAX_ROWS = 500000;
+    private static final int DEFAULT_BUFFER_FLUSH_MAX_BYTES = 100 * 1024 * 1024;
     private static final long DEFAULT_BUFFER_FLUSH_INTERVAL_MS = 10 * 1000;
     private final int checkInterval;
     private final int maxRetries;

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

@@ -0,0 +1,77 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.flink.sink.batch;
+
+import org.apache.http.entity.AbstractHttpEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.List;
+
+public class BatchBufferHttpEntity extends AbstractHttpEntity {
+
+    private static final Logger LOG = LoggerFactory.getLogger(BatchBufferHttpEntity.class);
+
+    protected static final int OUTPUT_BUFFER_SIZE = 4096;
+    private final List<byte[]> buffer;
+    private final long contentLength;
+
+    public BatchBufferHttpEntity(BatchRecordBuffer recordBuffer) {
+        this.buffer = recordBuffer.getBuffer();
+        this.contentLength = recordBuffer.getBufferSizeBytes();
+    }
+
+    @Override
+    public boolean isRepeatable() {
+        return true;
+    }
+
+    @Override
+    public boolean isChunked() {
+        return false;
+    }
+
+    @Override
+    public long getContentLength() {
+        return contentLength;
+    }
+
+    @Override
+    public InputStream getContent() {
+        return new BatchBufferStream(buffer);
+    }
+
+    @Override
+    public void writeTo(OutputStream outStream) throws IOException {
+        try (InputStream inStream = new BatchBufferStream(buffer)) {
+            final byte[] buffer = new byte[OUTPUT_BUFFER_SIZE];
+            int readLen;
+            while ((readLen = inStream.read(buffer)) != -1) {
+                outStream.write(buffer, 0, readLen);
+            }
+        }
+    }
+
+    @Override
+    public boolean isStreaming() {
+        return false;
+    }
+}

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

@@ -0,0 +1,74 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.flink.sink.batch;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Iterator;
+import java.util.List;
+
+public class BatchBufferStream extends InputStream {
+    private final Iterator<byte[]> iterator;
+    private byte[] currentItem;
+    private int currentPos;
+
+    public BatchBufferStream(List<byte[]> buffer) {
+        this.iterator = buffer.iterator();
+    }
+
+    @Override
+    public int read() throws IOException {
+        return 0;
+    }
+
+    @Override
+    public int read(byte[] buf) throws IOException {
+        return read(buf, 0, buf.length);
+    }
+
+    @Override
+    public int read(byte[] buf, int off, int len) throws IOException {
+        if (!iterator.hasNext() && currentItem == null) {
+            return -1;
+        }
+
+        byte[] item = currentItem;
+        int pos = currentPos;
+        int readBytes = 0;
+        while (readBytes < len && (item != null || iterator.hasNext())) {
+            if (item == null) {
+                item = iterator.next();
+                pos = 0;
+            }
+
+            int size = Math.min(len - readBytes, item.length - pos);
+            System.arraycopy(item, pos, buf, off + readBytes, size);
+            readBytes += size;
+            pos += size;
+
+            if (pos == item.length) {
+                item = null;
+                pos = 0;
+            }
+        }
+
+        currentItem = item;
+        currentPos = pos;
+        return readBytes;
+    }
+}

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

@@ -17,19 +17,17 @@
 
 package org.apache.doris.flink.sink.batch;
 
-import org.apache.flink.annotation.VisibleForTesting;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.nio.ByteBuffer;
+import java.util.LinkedList;
 
 /** buffer to queue. */
 public class BatchRecordBuffer {
     private static final Logger LOG = LoggerFactory.getLogger(BatchRecordBuffer.class);
     public static final String LINE_SEPARATOR = "\n";
     private String labelName;
-    private ByteBuffer buffer;
+    private LinkedList<byte[]> buffer;
     private byte[] lineDelimiter;
     private int numOfRecords = 0;
     private int bufferSizeBytes = 0;
@@ -39,64 +37,30 @@ public class BatchRecordBuffer {
     private final long createTime = System.currentTimeMillis();
     private long retainTime = 0;
 
-    public BatchRecordBuffer() {}
-
-    public BatchRecordBuffer(byte[] lineDelimiter, int bufferSize) {
-        super();
-        this.lineDelimiter = lineDelimiter;
-        this.buffer = ByteBuffer.allocate(bufferSize);
+    public BatchRecordBuffer() {
+        this.buffer = new LinkedList<>();
     }
 
-    public BatchRecordBuffer(
-            String database, String table, byte[] lineDelimiter, int bufferSize, long retainTime) {
+    public BatchRecordBuffer(String database, String table, byte[] lineDelimiter, long retainTime) {
         super();
         this.database = database;
         this.table = table;
         this.lineDelimiter = lineDelimiter;
-        this.buffer = ByteBuffer.allocate(bufferSize);
+        this.buffer = new LinkedList<>();
         this.retainTime = retainTime;
     }
 
     public void insert(byte[] record) {
-        ensureCapacity(record.length);
         if (loadBatchFirstRecord) {
             loadBatchFirstRecord = false;
         } else if (lineDelimiter != null) {
-            this.buffer.put(this.lineDelimiter);
+            this.buffer.add(this.lineDelimiter);
         }
-        this.buffer.put(record);
+        this.buffer.add(record);
         setNumOfRecords(getNumOfRecords() + 1);
         setBufferSizeBytes(getBufferSizeBytes() + record.length);
     }
 
-    @VisibleForTesting
-    public void ensureCapacity(int length) {
-        int lineDelimiterSize = this.lineDelimiter == null ? 0 : this.lineDelimiter.length;
-        if (buffer.remaining() - lineDelimiterSize >= length) {
-            return;
-        }
-        int currentRemain = buffer.remaining();
-        int currentCapacity = buffer.capacity();
-        // add lineDelimiter length
-        int needed = length - buffer.remaining() + lineDelimiterSize;
-        // grow at least 1MB
-        long grow = Math.max(needed, 1024 * 1024);
-        // grow at least 50% of the current size
-        grow = Math.max(buffer.capacity() / 2, grow);
-        int newCapacity = (int) Math.min(Integer.MAX_VALUE, buffer.capacity() + grow);
-        ByteBuffer tmp = ByteBuffer.allocate(newCapacity);
-        buffer.flip();
-        tmp.put(buffer);
-        buffer.clear();
-        buffer = tmp;
-        LOG.info(
-                "record length {},buffer remain {} ,grow capacity {} to {}",
-                length,
-                currentRemain,
-                currentCapacity,
-                newCapacity);
-    }
-
     public String getLabelName() {
         return labelName;
     }
@@ -110,13 +74,6 @@ public boolean isEmpty() {
         return numOfRecords == 0;
     }
 
-    public ByteBuffer getData() {
-        // change mode
-        buffer.flip();
-        LOG.debug("flush buffer: {} records, {} bytes", getNumOfRecords(), getBufferSizeBytes());
-        return buffer;
-    }
-
     public void clear() {
         this.buffer.clear();
         this.numOfRecords = 0;
@@ -125,7 +82,7 @@ public void clear() {
         this.loadBatchFirstRecord = true;
     }
 
-    public ByteBuffer getBuffer() {
+    public LinkedList<byte[]> getBuffer() {
         return buffer;
     }
 
@@ -165,6 +122,13 @@ public void setTable(String table) {
         this.table = table;
     }
 
+    public String getTableIdentifier() {
+        if (database != null && table != null) {
+            return database + "." + table;
+        }
+        return null;
+    }
+
     public boolean shouldFlush() {
         return System.currentTimeMillis() - createTime > retainTime;
     }