add a retry interval config

apache · Sep 15, 2023 · f709ba6 · f709ba6
1 parent 787de06
commit f709ba6
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 5 deletions.
diff --git a/spark-doris-connector/src/main/java/org/apache/doris/spark/cfg/ConfigurationOptions.java b/spark-doris-connector/src/main/java/org/apache/doris/spark/cfg/ConfigurationOptions.java
@@ -90,6 +90,10 @@ public interface ConfigurationOptions {
 
     int DORIS_SINK_BATCH_INTERVAL_MS_DEFAULT = 50;
 
+    String DORIS_SINK_RETRY_INTERVAL_MS = "doris.sink.retry.interval.ms";
+
+    int DORIS_SINK_RETRY_INTERVAL_MS_DEFAULT = 50;
+
     /**
      * set types to ignore, split by comma
      * e.g.

diff --git a/spark-doris-connector/src/main/scala/org/apache/doris/spark/writer/DorisWriter.scala b/spark-doris-connector/src/main/scala/org/apache/doris/spark/writer/DorisWriter.scala
@@ -31,6 +31,8 @@ import java.io.IOException
 import java.time.Duration
 import java.util
 import java.util.Objects
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.locks.LockSupport
 import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.util.{Failure, Success}
@@ -46,8 +48,10 @@ class DorisWriter(settings: SparkSettings) extends Serializable {
   private val sinkTaskPartitionSize: Integer = settings.getIntegerProperty(ConfigurationOptions.DORIS_SINK_TASK_PARTITION_SIZE)
   private val sinkTaskUseRepartition: Boolean = settings.getProperty(ConfigurationOptions.DORIS_SINK_TASK_USE_REPARTITION,
     ConfigurationOptions.DORIS_SINK_TASK_USE_REPARTITION_DEFAULT.toString).toBoolean
-  private val batchInterValMs: Integer = settings.getIntegerProperty(ConfigurationOptions.DORIS_SINK_BATCH_INTERVAL_MS,
+  private val batchIntervalMs: Integer = settings.getIntegerProperty(ConfigurationOptions.DORIS_SINK_BATCH_INTERVAL_MS,
     ConfigurationOptions.DORIS_SINK_BATCH_INTERVAL_MS_DEFAULT)
+  private val retryIntervalMs: Integer = settings.getIntegerProperty(ConfigurationOptions.DORIS_SINK_RETRY_INTERVAL_MS,
+    ConfigurationOptions.DORIS_SINK_RETRY_INTERVAL_MS_DEFAULT)
 
   private val enable2PC: Boolean = settings.getBooleanProperty(ConfigurationOptions.DORIS_SINK_ENABLE_2PC,
     ConfigurationOptions.DORIS_SINK_ENABLE_2PC_DEFAULT);
@@ -80,10 +84,12 @@ class DorisWriter(settings: SparkSettings) extends Serializable {
      *
      */
     def flush(batch: Seq[util.List[Object]], dfColumns: Array[String]): Unit = {
-      Utils.retry[util.List[Integer], Exception](maxRetryTimes, Duration.ofMillis(batchInterValMs.toLong), logger) {
+      Utils.retry[util.List[Integer], Exception](maxRetryTimes, Duration.ofMillis(retryIntervalMs.toLong), logger) {
         dorisStreamLoader.loadV2(batch.asJava, dfColumns, enable2PC)
       } match {
-        case Success(txnIds) => if (enable2PC) handleLoadSuccess(txnIds.asScala, preCommittedTxnAcc)
+        case Success(txnIds) =>
+          if (enable2PC) handleLoadSuccess(txnIds.asScala, preCommittedTxnAcc)
+          LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(batchIntervalMs.toLong))
         case Failure(e) =>
           if (enable2PC) handleLoadFailure(preCommittedTxnAcc)
           throw new IOException(
@@ -120,10 +126,12 @@ class DorisWriter(settings: SparkSettings) extends Serializable {
      *
      */
     def flush(batch: Seq[InternalRow], dfColumns: Array[String]): Unit = {
-      Utils.retry[util.List[Integer], Exception](maxRetryTimes, Duration.ofMillis(batchInterValMs.toLong), logger) {
+      Utils.retry[util.List[Integer], Exception](maxRetryTimes, Duration.ofMillis(retryIntervalMs.toLong), logger) {
         dorisStreamLoader.loadStream(convertToObjectList(batch, schema), dfColumns, enable2PC)
       } match {
-        case Success(txnIds) => if (enable2PC) handleLoadSuccess(txnIds.asScala, preCommittedTxnAcc)
+        case Success(txnIds) =>
+          if (enable2PC) handleLoadSuccess(txnIds.asScala, preCommittedTxnAcc)
+          LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(batchIntervalMs.toLong))
         case Failure(e) =>
           if (enable2PC) handleLoadFailure(preCommittedTxnAcc)
           throw new IOException(