SAMZA-1795 table: add common retry for IO functions #618
Conversation
Add common retry functionality to table IO functions for data stores that do not have native retry support. We use failsafe as the retry library.
|
@prateekm please help review when you have a chance. |
weisong44
left a comment
weisong44
left a comment
There was a problem hiding this comment.
Thanks for the contribution, overall LGTM. A few minor comments.
|
|
||
| import org.apache.samza.SamzaException; | ||
|
|
||
| import net.jodah.failsafe.RetryPolicy; |
There was a problem hiding this comment.
Ideally the retry policy is just a POJO that captures retry parameters. This class should be independent on the implementation (failsafe) we chose. And it should not take a dependency on failsafe.
| * Currently, the policy object can be translated into {@link RetryPolicy} of failsafe library. | ||
| */ | ||
| public class TableRetryPolicy implements Serializable { | ||
| enum BackoffType { NONE, FIXED, RANDOM, EXPONENTIAL } |
There was a problem hiding this comment.
Can we describe the behavior and related parameters of each?
| * Wrapper of retry-related metrics common to both {@link RetriableReadFunction} and | ||
| * {@link RetriableWriteFunction}. | ||
| */ | ||
| class RetryMetrics { |
There was a problem hiding this comment.
It it possible to get counts for failed request after retry?
|
|
||
| private AsyncFailsafe<?> failsafe() { | ||
| long startMs = System.currentTimeMillis(); | ||
| return Failsafe.with(retryPolicy).with(retryExecutor) |
There was a problem hiding this comment.
Is this part reusable or do we have to create a new instance for every request?
There was a problem hiding this comment.
We need the instance because the timestamp is unique per request and each failsafe object contains its own context as retry is per-request not shared across all requests.
| TableRetryPolicy writeRetryPolicy = null; | ||
|
|
||
| if (readRetryPolicy != null || writeRetryPolicy != null) { | ||
| retryExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> { |
There was a problem hiding this comment.
Wondering if it's better to share this across all tasks and tables?
There was a problem hiding this comment.
Makes sense. I'll make this a singleton
- extract failsafe logic into an adapter class from TableRetryPolicy - added a unit test class for TableRetryPolicy + FailsafeAdapter - use retry executor service as a singleton - added a permanent-failure metric
This allows the application to have a say on which exception types can be retried. Exception will be retried if either table function or the custom predicate say so.
| */ | ||
| public TableRetryPolicy withRetryOn(RetryPredicate isRetriable) { | ||
| Preconditions.checkNotNull(isRetriable); | ||
| this.isRetriable = isRetriable; |
There was a problem hiding this comment.
Shall we simply call it retryPredicate?
| * @param isRetriable predicate for retriable exception identification | ||
| * @return this policy instance | ||
| */ | ||
| public TableRetryPolicy withRetryOn(RetryPredicate isRetriable) { |
There was a problem hiding this comment.
Shall we just call it withRetryPredicate?
xinyuiscool
left a comment
xinyuiscool
left a comment
There was a problem hiding this comment.
LGTM. Have one minor comment.
A question for table provider: is it created per task? I want to make sure the readFn and writeFn will be created a new instance for each task. thx
samza-core/src/main/java/org/apache/samza/table/retry/RetriableReadFunction.java
Outdated
Show resolved
Hide resolved
|
@xinyuiscool yes, the table and the associated read/write fns are created per task instance. |
|
Thanks for answering my question. Please address all @weisong44 's feedback and I will commit it once it's finalized. |
3 participants
Add common retry functionality to table IO functions for data stores
that do not have native retry support. We use failsafe as the retry
library.