diff --git a/src/main/java/net/greghaines/jesque/worker/DefaultFailQueueStrategy.java b/src/main/java/net/greghaines/jesque/worker/DefaultFailQueueStrategy.java index 2c8166ac..c71a038c 100644 --- a/src/main/java/net/greghaines/jesque/worker/DefaultFailQueueStrategy.java +++ b/src/main/java/net/greghaines/jesque/worker/DefaultFailQueueStrategy.java @@ -21,7 +21,8 @@ import net.greghaines.jesque.utils.JesqueUtils; /** - * DefaultFailQueueStrategy puts all jobs in the standard Redis failure queue. + * DefaultFailQueueStrategy puts all jobs in the standard Redis failure queue + * and imposes no limits on the max number of failed items to keep. */ public class DefaultFailQueueStrategy implements FailQueueStrategy { @@ -42,4 +43,12 @@ public DefaultFailQueueStrategy(final String namespace) { public String getFailQueueKey(final Throwable thrwbl, final Job job, final String curQueue) { return JesqueUtils.createKey(this.namespace, FAILED); } + + /** + * {@inheritDoc} + */ + @Override + public int getFailQueueMaxItems(String curQueue) { + return 0; + } } diff --git a/src/main/java/net/greghaines/jesque/worker/FailQueueStrategy.java b/src/main/java/net/greghaines/jesque/worker/FailQueueStrategy.java index df7fee6b..a3f8ea6c 100644 --- a/src/main/java/net/greghaines/jesque/worker/FailQueueStrategy.java +++ b/src/main/java/net/greghaines/jesque/worker/FailQueueStrategy.java @@ -32,4 +32,12 @@ public interface FailQueueStrategy { * @return the key of the failure queue to put the job into or null */ String getFailQueueKey(Throwable thrwbl, Job job, String curQueue); + + /** + * Determine the max number of items to keep in the failure queue. + * Returning a value <1 means that there is no limit. + * @param curQueue the queue the Job came from + * @return the max number of items to keep for the failure queue + */ + int getFailQueueMaxItems(String curQueue); } diff --git a/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java b/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java index f09c86d3..3ba046c2 100644 --- a/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java +++ b/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java @@ -51,6 +51,7 @@ import net.greghaines.jesque.utils.ScriptUtils; import net.greghaines.jesque.utils.VersionUtils; import redis.clients.jedis.Jedis; +import redis.clients.jedis.Transaction; import redis.clients.jedis.exceptions.JedisException; /** @@ -677,9 +678,21 @@ protected void failure(final Throwable thrwbl, final Job job, final String curQu try { this.jedis.incr(key(STAT, FAILED)); this.jedis.incr(key(STAT, FAILED, this.name)); - final String failQueueKey = this.failQueueStrategyRef.get().getFailQueueKey(thrwbl, job, curQueue); + final FailQueueStrategy strategy = this.failQueueStrategyRef.get(); + final String failQueueKey = strategy.getFailQueueKey(thrwbl, job, curQueue); if (failQueueKey != null) { - this.jedis.rpush(failQueueKey, failMsg(thrwbl, curQueue, job)); + final int failQueueMaxItems = strategy.getFailQueueMaxItems(curQueue); + if (failQueueMaxItems > 0) { + Long currentItems = this.jedis.llen(failQueueKey); + if (currentItems >= failQueueMaxItems) { + Transaction tx = this.jedis.multi(); + tx.ltrim(failQueueKey, 1, -1); + tx.rpush(failQueueKey, failMsg(thrwbl, curQueue, job)); + tx.exec(); + } + } else { + this.jedis.rpush(failQueueKey, failMsg(thrwbl, curQueue, job)); + } } } catch (JedisException je) { LOG.warn("Error updating failure stats for throwable=" + thrwbl + " job=" + job, je); diff --git a/src/main/java/net/greghaines/jesque/worker/WorkerPoolImpl.java b/src/main/java/net/greghaines/jesque/worker/WorkerPoolImpl.java index 3611c11a..6fb108bc 100644 --- a/src/main/java/net/greghaines/jesque/worker/WorkerPoolImpl.java +++ b/src/main/java/net/greghaines/jesque/worker/WorkerPoolImpl.java @@ -49,6 +49,7 @@ import org.slf4j.LoggerFactory; import redis.clients.jedis.Jedis; +import redis.clients.jedis.Transaction; import redis.clients.jedis.exceptions.JedisException; import redis.clients.util.Pool; @@ -675,9 +676,21 @@ protected void failure(final Throwable thrwbl, final Job job, final String curQu public Void doWork(final Jedis jedis) throws IOException { jedis.incr(key(STAT, FAILED)); jedis.incr(key(STAT, FAILED, name)); - final String failQueueKey = failQueueStrategyRef.get().getFailQueueKey(thrwbl, job, curQueue); + final FailQueueStrategy strategy = failQueueStrategyRef.get(); + final String failQueueKey = strategy.getFailQueueKey(thrwbl, job, curQueue); if (failQueueKey != null) { - jedis.rpush(failQueueKey, failMsg(thrwbl, curQueue, job)); + final int failQueueMaxItems = strategy.getFailQueueMaxItems(curQueue); + if (failQueueMaxItems > 0) { + Long currentItems = jedis.llen(failQueueKey); + if (currentItems >= failQueueMaxItems) { + Transaction tx = jedis.multi(); + tx.ltrim(failQueueKey, 1, -1); + tx.rpush(failQueueKey, failMsg(thrwbl, curQueue, job)); + tx.exec(); + } + } else { + jedis.rpush(failQueueKey, failMsg(thrwbl, curQueue, job)); + } } return null; }