flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tillrohrmann <...@git.apache.org>
Subject [GitHub] flink pull request #1954: [FLINK-3190] failure rate restart strategy
Date Fri, 08 Jul 2016 15:37:26 GMT
Github user tillrohrmann commented on a diff in the pull request:

    https://github.com/apache/flink/pull/1954#discussion_r70094058
  
    --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/FailureRateRestartStrategy.java
---
    @@ -36,19 +36,19 @@
     public class FailureRateRestartStrategy implements RestartStrategy {
     	private final Duration failuresInterval;
     	private final Duration delayInterval;
    -	private EvictingQueue<Long> restartTimestampsQueue;
    +	private FixedSizeFifoQueue<Long> restartTimestampsQueue;
     	private boolean disabled = false;
     
     	public FailureRateRestartStrategy(int maxFailuresPerInterval, Duration failuresInterval,
Duration delayInterval) {
    -		Preconditions.checkArgument(maxFailuresPerInterval > 0, "Maximum number of restart
attempts per time unit must be greater than 0.");
     		Preconditions.checkNotNull(failuresInterval, "Failures interval cannot be null.");
    -		Preconditions.checkNotNull(failuresInterval.length() > 0, "Failures interval must
be greater than 0 ms.");
     		Preconditions.checkNotNull(delayInterval, "Delay interval cannot be null.");
    -		Preconditions.checkNotNull(delayInterval.length() >= 0, "Delay interval must be
at least 0 ms.");
    +		Preconditions.checkArgument(maxFailuresPerInterval > 0, "Maximum number of restart
attempts per time unit must be greater than 0.");
    +		Preconditions.checkArgument(failuresInterval.length() > 0, "Failures interval must
be greater than 0 ms.");
    +		Preconditions.checkArgument(delayInterval.length() >= 0, "Delay interval must be
at least 0 ms.");
     
     		this.failuresInterval = failuresInterval;
     		this.delayInterval = delayInterval;
    -		this.restartTimestampsQueue = EvictingQueue.create(maxFailuresPerInterval);
    +		this.restartTimestampsQueue = new FixedSizeFifoQueue<>(maxFailuresPerInterval);
    --- End diff --
    
    Can't we simply use `new ArrayDeque(maxFailuresPerInterval)`? Of course, we would then
allocate 2^(ceil(log(maxFailuresPerInterval)/log(2)) elements, but this should be ok. We could
then check in the `restart` method via the `size` method whether the queue is full or not.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

Mime
View raw message