From ab4447fc12a1b29b86132ce702846a70a549603b Mon Sep 17 00:00:00 2001 From: Ali Alsatrawi <7576983+ali-jawad@users.noreply.github.com> Date: Fri, 23 Oct 2026 16:35:31 +0200 Subject: [PATCH 1/2] Close the batch enqueue pipeline to return its pooled connection UnifiedJedis.pipelined() borrows a connection from the pool that is only returned when the pipeline is closed, so every ClientPoolImpl.batchEnqueue() call leaked one connection until the pool was exhausted. --- .../jesque/client/AbstractClient.java | 12 +++-- .../java/net/greghaines/jesque/TestUtils.java | 17 ++++++ .../jesque/client/TestClientPoolImpl.java | 54 +++++++++++++++++++ 3 files changed, 78 insertions(+), 5 deletions(-) create mode 100644 src/test/java/net/greghaines/jesque/client/TestClientPoolImpl.java diff --git a/src/main/java/net/greghaines/jesque/client/AbstractClient.java b/src/main/java/net/greghaines/jesque/client/AbstractClient.java index 04c4db81..e933249d 100644 --- a/src/main/java/net/greghaines/jesque/client/AbstractClient.java +++ b/src/main/java/net/greghaines/jesque/client/AbstractClient.java @@ -197,6 +197,7 @@ public static void doEnqueue( * Helper method that encapsulates the minimum logic for adding jobs to a queue. * * @param jedis the connection to Redis + * @param makePipeline creates the pipeline to send the commands with, which is closed afterwards * @param namespace the Resque namespace * @param queue the Resque queue name * @param jobJsons a list of jobs serialized as JSON @@ -207,12 +208,13 @@ public static void doBatchEnqueue( final String namespace, final String queue, final List jobJsons) { - AbstractPipeline pipelined = makePipeline.get(); // jedis.pipelined(); - pipelined.sadd(JesqueUtils.createKey(namespace, QUEUES), queue); - for (String jobJson : jobJsons) { - pipelined.rpush(JesqueUtils.createKey(namespace, QUEUE, queue), jobJson); + try (AbstractPipeline pipelined = makePipeline.get()) { // jedis.pipelined(); + pipelined.sadd(JesqueUtils.createKey(namespace, QUEUES), queue); + for (String jobJson : jobJsons) { + pipelined.rpush(JesqueUtils.createKey(namespace, QUEUE, queue), jobJson); + } + pipelined.sync(); } - pipelined.sync(); } /** diff --git a/src/test/java/net/greghaines/jesque/TestUtils.java b/src/test/java/net/greghaines/jesque/TestUtils.java index 677e7435..0b017ab7 100644 --- a/src/test/java/net/greghaines/jesque/TestUtils.java +++ b/src/test/java/net/greghaines/jesque/TestUtils.java @@ -15,17 +15,21 @@ import static com.google.common.truth.Truth.assertThat; +import java.time.Duration; import java.time.Instant; import java.util.List; import java.util.Map; import net.greghaines.jesque.client.Client; import net.greghaines.jesque.client.ClientImpl; +import net.greghaines.jesque.utils.PoolUtils; import net.greghaines.jesque.worker.JobExecutor; import net.greghaines.jesque.worker.JobFactory; import net.greghaines.jesque.worker.MapBasedJobFactory; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import redis.clients.jedis.ConnectionPoolConfig; import redis.clients.jedis.Jedis; +import redis.clients.jedis.UnifiedJedis; /** * Test helpers. @@ -65,6 +69,19 @@ public static Jedis createJedis(final Config config) { return jedis; } + /** + * @param config the location of the Redis server + * @param maxTotal the maximum number of connections the pool hands out at once + * @return a new pool that fails to hand out a connection after waiting for one second + */ + public static UnifiedJedis createBoundedJedisPool(final Config config, final int maxTotal) { + final ConnectionPoolConfig poolConfig = new ConnectionPoolConfig(); + poolConfig.setMaxTotal(maxTotal); + poolConfig.setBlockWhenExhausted(true); + poolConfig.setMaxWait(Duration.ofSeconds(1)); + return PoolUtils.createJedisPool(config, poolConfig); + } + public static void enqueueJobs(final String queue, final List jobs, final Config config) { final Client client = new ClientImpl(config); try { diff --git a/src/test/java/net/greghaines/jesque/client/TestClientPoolImpl.java b/src/test/java/net/greghaines/jesque/client/TestClientPoolImpl.java new file mode 100644 index 00000000..110d0de0 --- /dev/null +++ b/src/test/java/net/greghaines/jesque/client/TestClientPoolImpl.java @@ -0,0 +1,54 @@ +package net.greghaines.jesque.client; + +import static com.google.common.truth.Truth.assertThat; +import static net.greghaines.jesque.TestUtils.createBoundedJedisPool; +import static net.greghaines.jesque.TestUtils.createJedis; +import static net.greghaines.jesque.utils.JesqueUtils.createKey; +import static net.greghaines.jesque.utils.ResqueConstants.QUEUE; + +import java.util.List; +import net.greghaines.jesque.Config; +import net.greghaines.jesque.Job; +import net.greghaines.jesque.TestAction; +import net.greghaines.jesque.TestUtils; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.UnifiedJedis; + +public class TestClientPoolImpl { + + private static final Config CONFIG = Config.getDefaultConfig(); + private static final String TEST_QUEUE = "foo"; + private static final String TEST_QUEUE_KEY = createKey(CONFIG.getNamespace(), QUEUE, TEST_QUEUE); + private static final int POOL_SIZE = 3; + + private UnifiedJedis jedisPool; + private Client client; + + @Before + public void setUp() { + TestUtils.resetRedis(CONFIG); + this.jedisPool = createBoundedJedisPool(CONFIG, POOL_SIZE); + this.client = new ClientPoolImpl(CONFIG, this.jedisPool); + } + + @After + public void tearDown() { + this.client.end(); + this.jedisPool.close(); + } + + @Test + public void testBatchEnqueue_ReturnsConnectionToPool() { + final int batchCount = POOL_SIZE + 1; + final Job job = new Job(TestAction.class.getSimpleName()); + for (int i = 0; i < batchCount; i++) { + this.client.batchEnqueue(TEST_QUEUE, List.of(job)); + } + try (Jedis jedis = createJedis(CONFIG)) { + assertThat(jedis.llen(TEST_QUEUE_KEY)).isEqualTo(batchCount); + } + } +} From 339b5762a56198964efd50d4eaa1443d7f62fdb6 Mon Sep 17 00:00:00 2001 From: Ali Alsatrawi <7576983+ali-jawad@users.noreply.github.com> Date: Fri, 23 Oct 2026 16:37:09 +0200 Subject: [PATCH 2/2] Close transactions to return their pooled connections UnifiedJedis.multi() also borrows a connection from the pool that is only returned when the transaction is closed, so ClientPoolImpl's recurring enqueue/removal and WorkerPoolImpl's fail queue trimming (when a FailQueueStrategy limits the max items) leaked one connection per call. --- .../jesque/client/AbstractClient.java | 22 ++++--- .../jesque/worker/AbstractWorker.java | 9 +-- .../jesque/client/TestClientPoolImpl.java | 42 +++++++++++++ .../jesque/worker/TestWorkerPoolImpl.java | 63 +++++++++++++++++++ 4 files changed, 122 insertions(+), 14 deletions(-) create mode 100644 src/test/java/net/greghaines/jesque/worker/TestWorkerPoolImpl.java diff --git a/src/main/java/net/greghaines/jesque/client/AbstractClient.java b/src/main/java/net/greghaines/jesque/client/AbstractClient.java index e933249d..da28aa86 100644 --- a/src/main/java/net/greghaines/jesque/client/AbstractClient.java +++ b/src/main/java/net/greghaines/jesque/client/AbstractClient.java @@ -374,11 +374,12 @@ public static void doRecurringEnqueue( final String hashKey = JesqueUtils.createRecurringHashKey(queueKey); if (JedisUtils.canUseAsRecurringQueue(jedis, queueKey, hashKey)) { - AbstractTransaction transaction = makeTransaction.get(); // jedis.multi(); - transaction.zadd(queueKey, future.toEpochMilli(), jobJson); - transaction.hset(hashKey, jobJson, Long.toString(frequency.toMillis())); - if (transaction.exec() == null) { - throw new RuntimeException("cannot add " + jobJson + " to recurring queue " + queue); + try (AbstractTransaction transaction = makeTransaction.get()) { // jedis.multi(); + transaction.zadd(queueKey, future.toEpochMilli(), jobJson); + transaction.hset(hashKey, jobJson, Long.toString(frequency.toMillis())); + if (transaction.exec() == null) { + throw new RuntimeException("cannot add " + jobJson + " to recurring queue " + queue); + } } } else { throw new IllegalArgumentException(queue + " cannot be used as a recurring queue"); @@ -412,11 +413,12 @@ public static void doRemoveRecurringEnqueue( final String hashKey = JesqueUtils.createRecurringHashKey(queueKey); if (JedisUtils.canUseAsRecurringQueue(jedis, queueKey, hashKey)) { - AbstractTransaction transaction = makeTransaction.get(); // jedis.multi(); - transaction.hdel(hashKey, jobJson); - transaction.zrem(queueKey, jobJson); - if (transaction.exec() == null) { - throw new RuntimeException("cannot remove " + jobJson + " from recurring queue " + queue); + try (AbstractTransaction transaction = makeTransaction.get()) { // jedis.multi(); + transaction.hdel(hashKey, jobJson); + transaction.zrem(queueKey, jobJson); + if (transaction.exec() == null) { + throw new RuntimeException("cannot remove " + jobJson + " from recurring queue " + queue); + } } } else { throw new IllegalArgumentException(queue + " cannot be used as a recurring queue"); diff --git a/src/main/java/net/greghaines/jesque/worker/AbstractWorker.java b/src/main/java/net/greghaines/jesque/worker/AbstractWorker.java index 5c780496..b9e4e942 100644 --- a/src/main/java/net/greghaines/jesque/worker/AbstractWorker.java +++ b/src/main/java/net/greghaines/jesque/worker/AbstractWorker.java @@ -567,10 +567,11 @@ protected void failure(final Throwable thrwbl, final Job job, final String curQu if (failQueueKey != null) { final int failQueueMaxItems = strategy.getFailQueueMaxItems(curQueue); if (failQueueMaxItems > 0) { - AbstractTransaction tx = createTransaction(); - tx.rpush(failQueueKey, failMsg(thrwbl, curQueue, job)); - tx.ltrim(failQueueKey, -failQueueMaxItems, -1); - tx.exec(); + try (AbstractTransaction tx = createTransaction()) { + tx.rpush(failQueueKey, failMsg(thrwbl, curQueue, job)); + tx.ltrim(failQueueKey, -failQueueMaxItems, -1); + tx.exec(); + } } else { getJedis().rpush(failQueueKey, failMsg(thrwbl, curQueue, job)); } diff --git a/src/test/java/net/greghaines/jesque/client/TestClientPoolImpl.java b/src/test/java/net/greghaines/jesque/client/TestClientPoolImpl.java index 110d0de0..bd450312 100644 --- a/src/test/java/net/greghaines/jesque/client/TestClientPoolImpl.java +++ b/src/test/java/net/greghaines/jesque/client/TestClientPoolImpl.java @@ -6,7 +6,10 @@ import static net.greghaines.jesque.utils.JesqueUtils.createKey; import static net.greghaines.jesque.utils.ResqueConstants.QUEUE; +import java.time.Duration; +import java.time.Instant; import java.util.List; +import java.util.stream.IntStream; import net.greghaines.jesque.Config; import net.greghaines.jesque.Job; import net.greghaines.jesque.TestAction; @@ -22,6 +25,7 @@ public class TestClientPoolImpl { private static final Config CONFIG = Config.getDefaultConfig(); private static final String TEST_QUEUE = "foo"; private static final String TEST_QUEUE_KEY = createKey(CONFIG.getNamespace(), QUEUE, TEST_QUEUE); + private static final Duration RECURRING_FREQUENCY = Duration.ofMinutes(1); private static final int POOL_SIZE = 3; private UnifiedJedis jedisPool; @@ -51,4 +55,42 @@ public void testBatchEnqueue_ReturnsConnectionToPool() { assertThat(jedis.llen(TEST_QUEUE_KEY)).isEqualTo(batchCount); } } + + @Test + public void testRecurringEnqueue_ReturnsConnectionToPool() { + final List jobs = createDistinctJobs(POOL_SIZE + 1); + final Instant future = Instant.now().plus(RECURRING_FREQUENCY); + for (Job job : jobs) { + this.client.recurringEnqueue(TEST_QUEUE, job, future, RECURRING_FREQUENCY); + } + try (Jedis jedis = createJedis(CONFIG)) { + assertThat(jedis.zcard(TEST_QUEUE_KEY)).isEqualTo(jobs.size()); + } + } + + @Test + public void testRemoveRecurringEnqueue_ReturnsConnectionToPool() { + final List jobs = createDistinctJobs(POOL_SIZE + 1); + final Instant future = Instant.now().plus(RECURRING_FREQUENCY); + final Client seedingClient = new ClientImpl(CONFIG); + try { + for (Job job : jobs) { + seedingClient.recurringEnqueue(TEST_QUEUE, job, future, RECURRING_FREQUENCY); + } + } finally { + seedingClient.end(); + } + for (Job job : jobs) { + this.client.removeRecurringEnqueue(TEST_QUEUE, job); + } + try (Jedis jedis = createJedis(CONFIG)) { + assertThat(jedis.zcard(TEST_QUEUE_KEY)).isEqualTo(0L); + } + } + + private static List createDistinctJobs(final int count) { + return IntStream.range(0, count) + .mapToObj(i -> new Job(TestAction.class.getSimpleName(), i)) + .toList(); + } } diff --git a/src/test/java/net/greghaines/jesque/worker/TestWorkerPoolImpl.java b/src/test/java/net/greghaines/jesque/worker/TestWorkerPoolImpl.java new file mode 100644 index 00000000..1f230089 --- /dev/null +++ b/src/test/java/net/greghaines/jesque/worker/TestWorkerPoolImpl.java @@ -0,0 +1,63 @@ +package net.greghaines.jesque.worker; + +import static com.google.common.truth.Truth.assertThat; +import static net.greghaines.jesque.TestUtils.createBoundedJedisPool; +import static net.greghaines.jesque.TestUtils.createJedis; +import static net.greghaines.jesque.TestUtils.createTestActionJobFactory; +import static net.greghaines.jesque.utils.JesqueUtils.createKey; +import static net.greghaines.jesque.utils.ResqueConstants.FAILED; + +import java.util.List; +import net.greghaines.jesque.Config; +import net.greghaines.jesque.Job; +import net.greghaines.jesque.TestAction; +import net.greghaines.jesque.TestUtils; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.UnifiedJedis; + +public class TestWorkerPoolImpl { + + private static final Config CONFIG = Config.getDefaultConfig(); + private static final String TEST_QUEUE = "foo"; + private static final String FAIL_QUEUE_KEY = createKey(CONFIG.getNamespace(), FAILED); + private static final int POOL_SIZE = 3; + + private UnifiedJedis jedisPool; + private WorkerPoolImpl worker; + + @Before + public void setUp() { + TestUtils.resetRedis(CONFIG); + this.jedisPool = createBoundedJedisPool(CONFIG, POOL_SIZE); + this.worker = + new WorkerPoolImpl( + CONFIG, List.of(TEST_QUEUE), createTestActionJobFactory(), this.jedisPool); + } + + @After + public void tearDown() { + this.jedisPool.close(); + } + + @Test + public void testFailure_FailQueueMaxItems_ReturnsConnectionToPool() { + final int failureCount = POOL_SIZE + 1; + this.worker.setFailQueueStrategy( + new DefaultFailQueueStrategy(CONFIG.getNamespace()) { + @Override + public int getFailQueueMaxItems(final String curQueue) { + return failureCount; + } + }); + final Job job = new Job(TestAction.class.getSimpleName()); + for (int i = 0; i < failureCount; i++) { + this.worker.failure(new RuntimeException("BOOM!"), job, TEST_QUEUE); + } + try (Jedis jedis = createJedis(CONFIG)) { + assertThat(jedis.llen(FAIL_QUEUE_KEY)).isEqualTo(failureCount); + } + } +}