From f60bab86270bca337148da1b25cde2757490f972 Mon Sep 17 00:00:00 2001 From: Alexey Kuzin Date: Tue, 6 Feb 2024 23:44:24 +0100 Subject: [PATCH] Add an option for retrying conflict errors Make it possible to specify retrying of "Transaction aborted by conflict" errors. --- README.md | 6 +++--- .../connector/config/TarantoolConfig.scala | 4 ++-- .../connection/TarantoolConnection.scala | 18 ++++++++++++++++-- .../connector/util/ScalaToJavaHelper.scala | 8 ++++++++ 4 files changed, 29 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index eec58db..73ee431 100644 --- a/README.md +++ b/README.md @@ -49,9 +49,9 @@ libraryDependencies += "io.tarantool" %% "spark-tarantool-connector" % "0.7.0" | tarantool.requestTimeout | request completion timeout, in milliseconds | 2000 | | tarantool.connections | number of connections established with each host | 1 | | tarantool.cursorBatchSize | default limit for prefetching tuples in RDD iterator | 1000 | -| tarantool.retries.errorType | configures automatic retry of requests to Tarantool cluster. Possible values: "network", "none" | none | -| tarantool.retries.maxAttempts | maximum number of retry attempts for each request. Mandatory if errorType is set to "network" | | -| tarantool.retries.delay | delay between subsequent retries of each request (in milliseconds). Mandatory if errorType is set to "network" | | +| tarantool.retries.errorType | configures automatic retry of requests to Tarantool cluster. Possible values: "network", "conflict", "all", "none" | none | +| tarantool.retries.maxAttempts | maximum number of retry attempts for each request. Mandatory if errorType is not "none" | | +| tarantool.retries.delay | delay between subsequent retries of each request (in milliseconds). Mandatory if errorType is not "none" | | ### Dataset API request options diff --git a/src/main/scala/io/tarantool/spark/connector/config/TarantoolConfig.scala b/src/main/scala/io/tarantool/spark/connector/config/TarantoolConfig.scala index 1f0203f..8514a5b 100644 --- a/src/main/scala/io/tarantool/spark/connector/config/TarantoolConfig.scala +++ b/src/main/scala/io/tarantool/spark/connector/config/TarantoolConfig.scala @@ -11,7 +11,7 @@ case class Timeouts(connect: Option[Int], read: Option[Int], request: Option[Int object ErrorTypes extends Enumeration { type ErrorType = Value - val NONE, NETWORK = Value + val NONE, NETWORK, CONFLICT, ALL = Value } case class Retries(errorType: ErrorType, retryAttempts: Option[Int], delay: Option[Int]) extends Serializable @@ -110,7 +110,7 @@ object TarantoolConfig { if (strErrorType.isDefined) { val errorType = ErrorTypes.withName(strErrorType.get.toUpperCase) - if (errorType == ErrorTypes.NETWORK) { + if (errorType != ErrorTypes.NONE) { if (retryAttempts.isEmpty) { throw new IllegalArgumentException("Number of retry attempts must be specified") } diff --git a/src/main/scala/io/tarantool/spark/connector/connection/TarantoolConnection.scala b/src/main/scala/io/tarantool/spark/connector/connection/TarantoolConnection.scala index f473c88..54c1272 100644 --- a/src/main/scala/io/tarantool/spark/connector/connection/TarantoolConnection.scala +++ b/src/main/scala/io/tarantool/spark/connector/connection/TarantoolConnection.scala @@ -5,10 +5,12 @@ import io.tarantool.driver.api.{TarantoolClient, TarantoolClientConfig, Tarantoo import io.tarantool.driver.auth.SimpleTarantoolCredentials import io.tarantool.driver.api.TarantoolClientFactory import io.tarantool.driver.api.retry.TarantoolRequestRetryPolicies.AttemptsBoundRetryPolicyFactory +import io.tarantool.driver.api.retry.TarantoolRequestRetryPolicies.retryNetworkErrors +import io.tarantool.driver.exceptions.TarantoolInternalException import io.tarantool.driver.protocol.Packable import io.tarantool.spark.connector.Logging import io.tarantool.spark.connector.config.{ErrorTypes, TarantoolConfig} -import io.tarantool.spark.connector.util.ScalaToJavaHelper.toJavaUnaryOperator +import io.tarantool.spark.connector.util.ScalaToJavaHelper.{toJavaPredicate, toJavaUnaryOperator} import java.io.{Closeable, Serializable} import java.util @@ -24,6 +26,12 @@ object TarantoolConnection { def apply(): TarantoolConnection[TarantoolTuple, TarantoolResult[TarantoolTuple]] = TarantoolConnection(defaultClient) + private def isConflictError(e: Throwable): Boolean = + e.isInstanceOf[TarantoolInternalException] && + e.getMessage.indexOf("Transaction has been aborted by conflict") > 0 + + private def retryConflictErrors(): Predicate[Throwable] = toJavaPredicate(isConflictError) + private def defaultClient( clientConfig: TarantoolConfig ): TarantoolClient[TarantoolTuple, TarantoolResult[TarantoolTuple]] = { @@ -36,9 +44,15 @@ object TarantoolConnection { if (clientConfig.retries.isDefined) { val retries = clientConfig.retries.get - if (retries.errorType == ErrorTypes.NETWORK) { + if (retries.errorType != ErrorTypes.NONE) { + val predicate = retries.errorType match { + case ErrorTypes.NETWORK => retryNetworkErrors + case ErrorTypes.CONFLICT => retryConflictErrors + case _ => retryNetworkErrors.and(retryConflictErrors) + } clientFactory = clientFactory.withRetryingByNumberOfAttempts( retries.retryAttempts.get, + predicate, toJavaUnaryOperator { policyBuilder: AttemptsBoundRetryPolicyFactory.Builder[Predicate[Throwable]] => policyBuilder.withDelay(retries.delay.get) } diff --git a/src/main/scala/io/tarantool/spark/connector/util/ScalaToJavaHelper.scala b/src/main/scala/io/tarantool/spark/connector/util/ScalaToJavaHelper.scala index 14caf2a..ed1dce5 100644 --- a/src/main/scala/io/tarantool/spark/connector/util/ScalaToJavaHelper.scala +++ b/src/main/scala/io/tarantool/spark/connector/util/ScalaToJavaHelper.scala @@ -4,6 +4,7 @@ import java.util.function.{ BiFunction => JBiFunction, Consumer => JConsumer, Function => JFunction, + Predicate => JPredicate, Supplier => JSupplier, UnaryOperator => JUnaryOperator } @@ -33,6 +34,13 @@ object ScalaToJavaHelper { override def apply(t: T1): R = f.apply(t) } + /** + * Converts a Scala {@link Function1} to a Java {@link java.util.function.Predicate} + */ + def toJavaPredicate[T1](f: T1 => Boolean): JPredicate[T1] = new JPredicate[T1] { + override def test(t: T1): Boolean = f.apply(t) + } + /** * Converts a Scala {@link Function1} to a Java {@link java.util.function.UnaryOperator} */