cdkrot commented on code in PR #43757:
URL: https://github.com/apache/spark/pull/43757#discussion_r1392473855
##########
connector/connect/common/src/main/scala/org/apache/spark/sql/connect/client/GrpcRetryHandler.scala:
##########
@@ -254,22 +267,75 @@ private[sql] object GrpcRetryHandler extends Logging {
* Function that determines whether a retry is to be performed in the
event of an error.
*/
case class RetryPolicy(
- // Please synchronize changes here with Python side:
- // pyspark/sql/connect/client/core.py
- //
- // Note: these constants are selected so that the maximum tolerated wait
is guaranteed
- // to be at least 10 minutes
- maxRetries: Int = 15,
- initialBackoff: FiniteDuration = FiniteDuration(50, "ms"),
- maxBackoff: FiniteDuration = FiniteDuration(1, "min"),
- backoffMultiplier: Double = 4.0,
- jitter: FiniteDuration = FiniteDuration(500, "ms"),
- minJitterThreshold: FiniteDuration = FiniteDuration(2, "s"),
- canRetry: Throwable => Boolean = retryException) {}
+ maxRetries: Option[Int] = None,
+ initialBackoff: FiniteDuration = FiniteDuration(1000, "ms"),
+ maxBackoff: Option[FiniteDuration] = None,
+ backoffMultiplier: Double = 1.0,
+ jitter: FiniteDuration = FiniteDuration(0, "s"),
+ minJitterThreshold: FiniteDuration = FiniteDuration(0, "s"),
+ canRetry: Throwable => Boolean = (_ => false),
+ name: String = this.getClass.getName) {
+
+ def getName: String = name
+ def toState: RetryPolicyState = new RetryPolicyState(this)
+ }
+
+ def defaultPolicy(): RetryPolicy = RetryPolicy(
+ name = "DefaultPolicy",
+ // Please synchronize changes here with Python side:
+ // pyspark/sql/connect/client/core.py
+ //
+ // Note: these constants are selected so that the maximum tolerated wait
is guaranteed
+ // to be at least 10 minutes
+ maxRetries = Some(15),
+ initialBackoff = FiniteDuration(50, "ms"),
+ maxBackoff = Some(FiniteDuration(1, "min")),
+ backoffMultiplier = 4.0,
+ jitter = FiniteDuration(500, "ms"),
+ minJitterThreshold = FiniteDuration(2, "s"),
+ canRetry = retryException)
+
+ // list of policies to be used by this client
+ def defaultPolicies(): Seq[RetryPolicy] = List(defaultPolicy())
+
+ // represents a state of the specific policy
+ // (how many retries have happened and how much to wait until next one)
+ class RetryPolicyState(val policy: RetryPolicy) {
Review Comment:
I think we can keep it open, that would allow someone to extend it and
override `to_state` method if they want so.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
