[improvement] batch load retry (#148)

Co-authored-by: gnehil <adamlee489@gamil.com>

spark-doris-connector/src/main/java/org/apache/doris/spark/cfg/ConfigurationOptions.java

@@ -70,7 +70,7 @@ public interface ConfigurationOptions {
     int SINK_BATCH_SIZE_DEFAULT = 100000;
 
     String DORIS_SINK_MAX_RETRIES = "doris.sink.max-retries";
-    int SINK_MAX_RETRIES_DEFAULT = 1;
+    int SINK_MAX_RETRIES_DEFAULT = 0;
 
     String DORIS_MAX_FILTER_RATIO = "doris.max.filter.ratio";
 

spark-doris-connector/src/main/java/org/apache/doris/spark/load/DorisStreamLoad.java

@@ -98,7 +98,6 @@ public class DorisStreamLoad implements Serializable {
     private String FIELD_DELIMITER;
     private final String LINE_DELIMITER;
     private boolean streamingPassthrough = false;
-    private final Integer batchSize;
     private final boolean enable2PC;
     private final Integer txnRetries;
     private final Integer txnIntervalMs;
@@ -128,8 +127,6 @@ public DorisStreamLoad(SparkSettings settings) {
         LINE_DELIMITER = escapeString(streamLoadProp.getOrDefault("line_delimiter", "\n"));
         this.streamingPassthrough = settings.getBooleanProperty(ConfigurationOptions.DORIS_SINK_STREAMING_PASSTHROUGH,
                 ConfigurationOptions.DORIS_SINK_STREAMING_PASSTHROUGH_DEFAULT);
-        this.batchSize = settings.getIntegerProperty(ConfigurationOptions.DORIS_SINK_BATCH_SIZE,
-                ConfigurationOptions.SINK_BATCH_SIZE_DEFAULT);
         this.enable2PC = settings.getBooleanProperty(ConfigurationOptions.DORIS_SINK_ENABLE_2PC,
                 ConfigurationOptions.DORIS_SINK_ENABLE_2PC_DEFAULT);
         this.txnRetries = settings.getIntegerProperty(ConfigurationOptions.DORIS_SINK_TXN_RETRIES,
@@ -200,7 +197,6 @@ public int load(Iterator<InternalRow> rows, StructType schema)
             this.loadUrlStr = loadUrlStr;
             HttpPut httpPut = getHttpPut(label, loadUrlStr, enable2PC);
             RecordBatchInputStream recodeBatchInputStream = new RecordBatchInputStream(RecordBatch.newBuilder(rows)
-                    .batchSize(batchSize)
                     .format(fileType)
                     .sep(FIELD_DELIMITER)
                     .delim(LINE_DELIMITER)

spark-doris-connector/src/main/java/org/apache/doris/spark/load/RecordBatch.java

@@ -36,11 +36,6 @@ public class RecordBatch {
      */
     private final Iterator<InternalRow> iterator;
 
-    /**
-     * batch size for single load
-     */
-    private final int batchSize;
-
     /**
      * stream load format
      */
@@ -63,10 +58,9 @@ public class RecordBatch {
 
     private final boolean addDoubleQuotes;
 
-    private RecordBatch(Iterator<InternalRow> iterator, int batchSize, String format, String sep, byte[] delim,
+    private RecordBatch(Iterator<InternalRow> iterator, String format, String sep, byte[] delim,
                         StructType schema, boolean addDoubleQuotes) {
         this.iterator = iterator;
-        this.batchSize = batchSize;
         this.format = format;
         this.sep = sep;
         this.delim = delim;
@@ -78,10 +72,6 @@ public Iterator<InternalRow> getIterator() {
         return iterator;
     }
 
-    public int getBatchSize() {
-        return batchSize;
-    }
-
     public String getFormat() {
         return format;
     }
@@ -112,8 +102,6 @@ public static class Builder {
 
         private final Iterator<InternalRow> iterator;
 
-        private int batchSize;
-
         private String format;
 
         private String sep;
@@ -128,11 +116,6 @@ public Builder(Iterator<InternalRow> iterator) {
             this.iterator = iterator;
         }
 
-        public Builder batchSize(int batchSize) {
-            this.batchSize = batchSize;
-            return this;
-        }
-
         public Builder format(String format) {
             this.format = format;
             return this;
@@ -159,7 +142,7 @@ public Builder addDoubleQuotes(boolean addDoubleQuotes) {
         }
 
         public RecordBatch build() {
-            return new RecordBatch(iterator, batchSize, format, sep, delim, schema, addDoubleQuotes);
+            return new RecordBatch(iterator, format, sep, delim, schema, addDoubleQuotes);
         }
 
     }

spark-doris-connector/src/main/java/org/apache/doris/spark/load/RecordBatchInputStream.java

@@ -60,11 +60,6 @@ public class RecordBatchInputStream extends InputStream {
 
     private final byte[] delim;
 
-    /**
-     * record count has been read
-     */
-    private int readCount = 0;
-
     /**
      * streaming mode pass through data without process
      */
@@ -122,12 +117,12 @@ public int read(byte[] b, int off, int len) throws IOException {
      */
     public boolean endOfBatch() throws DorisException {
         Iterator<InternalRow> iterator = recordBatch.getIterator();
-        if (readCount >= recordBatch.getBatchSize() || !iterator.hasNext()) {
-            delimBuf = null;
-            return true;
+        if (iterator.hasNext()) {
+            readNext(iterator);
+            return false;
         }
-        readNext(iterator);
-        return false;
+        delimBuf = null;
+        return true;
     }
 
     /**
@@ -149,7 +144,6 @@ private void readNext(Iterator<InternalRow> iterator) throws DorisException {
             delimBuf =  ByteBuffer.wrap(delim);
             lineBuf = ByteBuffer.wrap(rowBytes);
         }
-        readCount++;
     }
 
     /**

spark-doris-connector/src/main/scala/org/apache/doris/spark/listener/DorisTransactionListener.scala

@@ -47,8 +47,8 @@ class DorisTransactionListener(preCommittedTxnAcc: CollectionAccumulator[Int], d
         txnIds.foreach(txnId =>
           Utils.retry(sinkTxnRetries, Duration.ofMillis(sinkTnxIntervalMs), logger) {
             dorisStreamLoad.commit(txnId)
-          } match {
-            case Success(_) =>
+          } () match {
+            case Success(_) => // do nothing
             case Failure(_) => failedTxnIds += txnId
           }
         )
@@ -68,8 +68,8 @@ class DorisTransactionListener(preCommittedTxnAcc: CollectionAccumulator[Int], d
         txnIds.foreach(txnId =>
           Utils.retry(sinkTxnRetries, Duration.ofMillis(sinkTnxIntervalMs), logger) {
             dorisStreamLoad.abortById(txnId)
-          } match {
-            case Success(_) =>
+          } () match {
+            case Success(_) => // do nothing
             case Failure(_) => failedTxnIds += txnId
           })
         if (failedTxnIds.nonEmpty) {

spark-doris-connector/src/main/scala/org/apache/doris/spark/sql/Utils.scala

@@ -34,15 +34,17 @@ import scala.util.{Failure, Success, Try}
 private[spark] object Utils {
   /**
    * quote column name
+   *
    * @param colName column name
    * @return quoted column name
    */
   def quote(colName: String): String = s"`$colName`"
 
   /**
    * compile a filter to Doris FE filter format.
-   * @param filter filter to be compile
-   * @param dialect jdbc dialect to translate value to sql format
+   *
+   * @param filter             filter to be compile
+   * @param dialect            jdbc dialect to translate value to sql format
    * @param inValueLengthLimit max length of in value array
    * @return if Doris FE can handle this filter, return None if Doris FE can not handled it.
    */
@@ -87,6 +89,7 @@ private[spark] object Utils {
 
   /**
    * Escape special characters in SQL string literals.
+   *
    * @param value The string to be escaped.
    * @return Escaped string.
    */
@@ -95,6 +98,7 @@ private[spark] object Utils {
 
   /**
    * Converts value to SQL expression.
+   *
    * @param value The value to be converted.
    * @return Converted value.
    */
@@ -108,16 +112,17 @@ private[spark] object Utils {
 
   /**
    * check parameters validation and process it.
+   *
    * @param parameters parameters from rdd and spark conf
-   * @param logger slf4j logger
+   * @param logger     slf4j logger
    * @return processed parameters
    */
   def params(parameters: Map[String, String], logger: Logger) = {
     // '.' seems to be problematic when specifying the options
     val dottedParams = parameters.map { case (k, v) =>
-      if (k.startsWith("sink.properties.") || k.startsWith("doris.sink.properties.")){
-        (k,v)
-      }else {
+      if (k.startsWith("sink.properties.") || k.startsWith("doris.sink.properties.")) {
+        (k, v)
+      } else {
         (k.replace('_', '.'), v)
       }
     }
@@ -141,7 +146,7 @@ private[spark] object Utils {
       case (k, v) =>
         if (k.startsWith("doris.")) (k, v)
         else ("doris." + k, v)
-    }.map{
+    }.map {
       case (ConfigurationOptions.DORIS_REQUEST_AUTH_PASSWORD, _) =>
         logger.error(s"${ConfigurationOptions.DORIS_REQUEST_AUTH_PASSWORD} cannot use in Doris Datasource.")
         throw new DorisException(s"${ConfigurationOptions.DORIS_REQUEST_AUTH_PASSWORD} cannot use in" +
@@ -165,13 +170,14 @@ private[spark] object Utils {
 
     // validate path is available
     finalParams.getOrElse(ConfigurationOptions.DORIS_TABLE_IDENTIFIER,
-        throw new DorisException("table identifier must be specified for doris table identifier."))
+      throw new DorisException("table identifier must be specified for doris table identifier."))
 
     finalParams
   }
 
   @tailrec
-  def retry[R, T <: Throwable : ClassTag](retryTimes: Int, interval: Duration, logger: Logger)(f: => R): Try[R] = {
+  def retry[R, T <: Throwable : ClassTag](retryTimes: Int, interval: Duration, logger: Logger)
+                                         (f: => R)(h: => Unit): Try[R] = {
     assert(retryTimes >= 0)
     val result = Try(f)
     result match {
@@ -182,7 +188,8 @@ private[spark] object Utils {
         logger.warn(s"Execution failed caused by: ", exception)
         logger.warn(s"$retryTimes times retry remaining, the next attempt will be in ${interval.toMillis} ms")
         LockSupport.parkNanos(interval.toNanos)
-        retry(retryTimes - 1, interval, logger)(f)
+        h
+        retry(retryTimes - 1, interval, logger)(f)(h)
       case Failure(exception) => Failure(exception)
     }
   }