From 414094c36a6358f24748e94327fa910917fbaa03 Mon Sep 17 00:00:00 2001 From: "nikita.smirnov" Date: Thu, 15 Oct 2026 10:24:24 +0000 Subject: [PATCH] [GH-53] Fixed BlockingScheduledRetryableTaskQueue.submit() blocking forever on oversized task (#53) * Oversized task is accepted when the queue holds no data * submit() throws InterruptedException (breaking change) * setMaxDataSize() wakes up blocked submit() calls * Bumped version to 0.1.5, updated README --- README.md | 18 +- gradle.properties | 2 +- .../BlockingScheduledRetryableTaskQueue.java | 39 ++++- ...stBlockingScheduledRetryableTaskQueue.java | 165 +++++++++++++++++- 4 files changed, 207 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index f4425ec..b0073c1 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# th2-tasks-utils (0.1.4) +# th2-tasks-utils (0.1.5) This library includes classes for managing tasks. @@ -14,8 +14,24 @@ Task extraction order is determined by their schedule time, that is task with ea Task extraction does not free up resources in queue. They are reserved for future retries. Resources are only released on task completion. +A task whose payload size exceeds the maximum data size is accepted only when the queue holds no data. It then occupies the whole data capacity until it is completed, and other submissions wait for it. + ## Release notes +### 0.1.5 + +* The release includes BROKEN changes related to `BlockingScheduledRetryableTaskQueue` class: + * The `void submit(ScheduledRetryableTask task)` method can throw InterruptedException. A blocked submission can now be interrupted, for example on shutdown. + +* Fixed: `BlockingScheduledRetryableTaskQueue.submit()` blocked forever when task's payload size exceeded max data size ([#53](https://github.com/th2-net/th2-task-utils/issues/53)). + Such task is now accepted when the queue holds no data and occupies the whole data capacity until it is completed. +* `BlockingScheduledRetryableTaskQueue.setMaxDataSize()` wakes up blocked `submit()` calls so that they re-check capacity against the new value. + +* Updated: + * th2 gradle plugin: `0.3.15` (from `0.2.4`) + * junit-bom: `5.14.4` (from `5.12.1`) + * assertj-core: `3.27.7` (from `3.27.3`) + ### 0.1.4 * The release includes BROKEN changes related to `FutureTracker` class: diff --git a/gradle.properties b/gradle.properties index 1f9aecd..37e1d18 100644 --- a/gradle.properties +++ b/gradle.properties @@ -1,3 +1,3 @@ -release_version = 0.1.4 +release_version = 0.1.5 description = "Task management utility classes" vcs_url = https://github.com/th2-net/th2-task-utils \ No newline at end of file diff --git a/src/main/java/com/exactpro/th2/taskutils/BlockingScheduledRetryableTaskQueue.java b/src/main/java/com/exactpro/th2/taskutils/BlockingScheduledRetryableTaskQueue.java index 7df4cc6..db261d4 100644 --- a/src/main/java/com/exactpro/th2/taskutils/BlockingScheduledRetryableTaskQueue.java +++ b/src/main/java/com/exactpro/th2/taskutils/BlockingScheduledRetryableTaskQueue.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2025 Exactpro (Exactpro Systems Limited) + * Copyright 2020-2026 Exactpro (Exactpro Systems Limited) * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,6 +16,9 @@ package com.exactpro.th2.taskutils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.util.HashSet; import java.util.PriorityQueue; import java.util.Queue; @@ -25,6 +28,7 @@ import java.util.concurrent.locks.ReentrantLock; public class BlockingScheduledRetryableTaskQueue { + private static final Logger LOGGER = LoggerFactory.getLogger(BlockingScheduledRetryableTaskQueue.class); /** * GuardedBy {@link #lock} @@ -82,24 +86,34 @@ public BlockingScheduledRetryableTaskQueue(int maxTaskCount, long maxDataSize, R /** * Submits task to the queue.
* Blocks until available capacity requirements are met, that is queue has space for this task by - * the count and cumulative data size of submitted task + * the count and cumulative data size of submitted task. + *

+ * A task whose payload size exceeds the maximum data size (see {@link #getMaxDataSize()}) can never fit + * into the remaining capacity. Such task is accepted only when the queue holds no data (used data size is 0), + * so it occupies the whole data capacity until it is completed. Until then all other submissions are blocked. + *

* * @param task Task to be added to the queue + * @throws InterruptedException if the current thread was interrupted while waiting for capacity. + * The task is not added to the queue in this case. */ - public void submit(ScheduledRetryableTask task) { + public void submit(ScheduledRetryableTask task) throws InterruptedException { lock.lock(); try { if (taskSet.contains(task)) throw new IllegalStateException("Task has been already submitted"); while (true) { - long capacityLeft = maxDataSize - dataSize; - if (capacityLeft >= task.getPayloadSize() && taskSet.size() < maxTaskCount) { + if (taskSet.size() < maxTaskCount && fitsDataCapacity(task.getPayloadSize())) { + if (task.getPayloadSize() > maxDataSize) { + LOGGER.warn("Task payload size {} exceeds max data size {}, task is accepted because queue is empty", + task.getPayloadSize(), maxDataSize); + } dataSize += task.getPayloadSize(); addTask(task); break; } else { - removal.awaitUninterruptibly(); + removal.await(); } } } finally { @@ -107,6 +121,13 @@ public void submit(ScheduledRetryableTask task) { } } + /** + * GuardedBy {@link #lock} + */ + private boolean fitsDataCapacity(long payloadSize) { + return dataSize == 0 || maxDataSize - dataSize >= payloadSize; + } + /** * Submits task for retry to the queue.
@@ -207,6 +228,7 @@ public ScheduledRetryableTask awaitScheduled() throws InterruptedException { ScheduledRetryableTask job = taskQueue.peek(); long now = System.nanoTime(); if (now < job.getScheduledTime()) + //noinspection ResultOfMethodCallIgnored addition.awaitNanos(job.getScheduledTime() - now); else return taskQueue.poll(); @@ -267,13 +289,16 @@ public long getMaxDataSize() { } /** - * Sets cumulative data size for all tasks that can be submitted before blocking + * Sets cumulative data size for all tasks that can be submitted before blocking. + * Blocked submit() calls re-check capacity against the new value. + * Tasks already in the queue are not affected even if their payload size exceeds the new value. * @param value New data size to be set */ public void setMaxDataSize(long value) { lock.lock(); try { this.maxDataSize = value; + removal.signalAll(); } finally { lock.unlock(); } diff --git a/src/test/java/com/exactpro/th2/taskutils/TestBlockingScheduledRetryableTaskQueue.java b/src/test/java/com/exactpro/th2/taskutils/TestBlockingScheduledRetryableTaskQueue.java index 841d906..d09fa17 100644 --- a/src/test/java/com/exactpro/th2/taskutils/TestBlockingScheduledRetryableTaskQueue.java +++ b/src/test/java/com/exactpro/th2/taskutils/TestBlockingScheduledRetryableTaskQueue.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2022 Exactpro (Exactpro Systems Limited) + * Copyright 2020-2026 Exactpro (Exactpro Systems Limited) * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,14 +18,23 @@ import org.junit.jupiter.api.Test; +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; public class TestBlockingScheduledRetryableTaskQueue { public static final long START_TIME = 1278417364718L; @Test - public void testBasicLogic() { + public void testBasicLogic() throws InterruptedException { final int maxTaskCount = 35; final long maxDataSize = 23748987; @@ -69,7 +78,7 @@ public void testBasicLogic() { } @Test - public void testJobExtractionOrder() { + public void testJobExtractionOrder() throws InterruptedException { ScheduledRetryableTask task1 = new ScheduledRetryableTask<>(START_TIME, 1, 100, null); ScheduledRetryableTask task2 = new ScheduledRetryableTask<>(START_TIME + 1234, 2, 100, null); @@ -91,7 +100,7 @@ public void testBlockingByCount() throws InterruptedException { final int taskCount = 10; StartableRunnable fastProducer = StartableRunnable.of(() -> { for (int i = 0; i < taskCount; i++) { - queue.submit(new ScheduledRetryableTask<>(System.nanoTime(), 0, i * 10, null)); + submitUnchecked(queue, new ScheduledRetryableTask<>(System.nanoTime(), 0, i * 10, null)); pause(1); } }); @@ -103,7 +112,7 @@ public void testBlockingByCount() throws InterruptedException { assertEquals(i * 10, task.getPayloadSize(), "Incorrect task"); pause(20); queue.complete(task); - } catch (InterruptedException e) { + } catch (InterruptedException ignored) { } } }); @@ -136,7 +145,7 @@ public void testBlockingBySize() throws InterruptedException { final int taskCount = 10; StartableRunnable fastProducer = StartableRunnable.of(() -> { for (int i = 0; i < taskCount; i++) { - queue.submit(new ScheduledRetryableTask<>(System.nanoTime(), i, 80, null)); + submitUnchecked(queue, new ScheduledRetryableTask<>(System.nanoTime(), i, 80, null)); pause(1); } }); @@ -148,7 +157,7 @@ public void testBlockingBySize() throws InterruptedException { assertEquals(i, task.getRetriesLeft(), "Incorrect task"); pause(20); queue.complete(task); - } catch (InterruptedException e) { + } catch (InterruptedException ignored) { } } }); @@ -173,10 +182,150 @@ public void testBlockingBySize() throws InterruptedException { assertEquals(0, queue.getUsedDataSize(), "Data size after execution"); } + @Test + public void testOversizedTaskIsAcceptedByEmptyQueue() { + BlockingScheduledRetryableTaskQueue queue = new BlockingScheduledRetryableTaskQueue<>(10, 100, null); + ScheduledRetryableTask task = new ScheduledRetryableTask<>(System.nanoTime(), 1, 101, new Object()); + + assertTimeoutPreemptively(TIMEOUT, () -> queue.submit(task), "Oversized task submission to empty queue"); + assertEquals(1, queue.getTaskCount(), "Task count"); + assertEquals(101, queue.getUsedDataSize(), "Used data size"); + + assertEquals(task, queue.take(), "Extracted task"); + queue.complete(task); + assertEquals(0, queue.getTaskCount(), "Task count"); + assertEquals(0, queue.getUsedDataSize(), "Used data size"); + } + + @Test + public void testOversizedTaskWaitsForEmptyQueue() throws Exception { + BlockingScheduledRetryableTaskQueue queue = new BlockingScheduledRetryableTaskQueue<>(10, 100, null); + ScheduledRetryableTask smallTask = new ScheduledRetryableTask<>(System.nanoTime(), 1, 10, null); + ScheduledRetryableTask oversizedTask = new ScheduledRetryableTask<>(System.nanoTime(), 1, 150, null); + queue.submit(smallTask); + + CompletableFuture submission = submitAsync(queue, oversizedTask); + assertBlocked(submission, "Oversized task submission to non-empty queue"); + assertEquals(1, queue.getTaskCount(), "Task count"); + + queue.complete(queue.take()); + submission.get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + assertEquals(1, queue.getTaskCount(), "Task count"); + assertEquals(150, queue.getUsedDataSize(), "Used data size"); + } + + @Test + public void testOversizedTaskBlocksFollowingSubmissions() throws Exception { + BlockingScheduledRetryableTaskQueue queue = new BlockingScheduledRetryableTaskQueue<>(10, 100, null); + ScheduledRetryableTask oversizedTask = new ScheduledRetryableTask<>(System.nanoTime(), 1, 150, null); + ScheduledRetryableTask smallTask = new ScheduledRetryableTask<>(System.nanoTime(), 1, 10, null); + assertTimeoutPreemptively(TIMEOUT, () -> queue.submit(oversizedTask), "Oversized task submission to empty queue"); + + CompletableFuture submission = submitAsync(queue, smallTask); + assertBlocked(submission, "Task submission while oversized task occupies queue"); + + queue.complete(queue.take()); + submission.get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + assertEquals(1, queue.getTaskCount(), "Task count"); + assertEquals(10, queue.getUsedDataSize(), "Used data size"); + } + + @Test + public void testSubmitAfterMaxDataSizeDecreasedBelowTaskSize() throws Exception { + BlockingScheduledRetryableTaskQueue queue = new BlockingScheduledRetryableTaskQueue<>(10, 100, null); + ScheduledRetryableTask firstTask = new ScheduledRetryableTask<>(System.nanoTime(), 1, 80, null); + ScheduledRetryableTask secondTask = new ScheduledRetryableTask<>(System.nanoTime(), 1, 80, null); + queue.submit(firstTask); + + CompletableFuture submission = submitAsync(queue, secondTask); + assertBlocked(submission, "Task submission to full queue"); + + queue.setMaxDataSize(50); + queue.complete(queue.take()); + submission.get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + assertEquals(1, queue.getTaskCount(), "Task count"); + assertEquals(80, queue.getUsedDataSize(), "Used data size"); + } + + @Test + public void testIncreasingMaxDataSizeUnblocksSubmit() throws Exception { + BlockingScheduledRetryableTaskQueue queue = new BlockingScheduledRetryableTaskQueue<>(10, 100, null); + queue.submit(new ScheduledRetryableTask<>(System.nanoTime(), 1, 80, null)); + + CompletableFuture submission = submitAsync(queue, new ScheduledRetryableTask<>(System.nanoTime(), 1, 80, null)); + assertBlocked(submission, "Task submission to full queue"); + + queue.setMaxDataSize(200); + submission.get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + assertEquals(2, queue.getTaskCount(), "Task count"); + assertEquals(160, queue.getUsedDataSize(), "Used data size"); + } + + @Test + public void testBlockedSubmitCanBeInterrupted() throws Exception { + BlockingScheduledRetryableTaskQueue queue = new BlockingScheduledRetryableTaskQueue<>(1, 100, null); + queue.submit(new ScheduledRetryableTask<>(System.nanoTime(), 1, 10, null)); + + CompletableFuture result = new CompletableFuture<>(); + Thread thread = new Thread(() -> { + try { + queue.submit(new ScheduledRetryableTask<>(System.nanoTime(), 1, 10, null)); + result.complete(null); + } catch (Throwable e) { + result.complete(e); + } + }); + thread.setDaemon(true); + thread.start(); + + assertBlocked(result, "Task submission to full queue"); + thread.interrupt(); + assertInstanceOf(InterruptedException.class, result.get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS), + "Blocked submit must be interrupted"); + assertEquals(1, queue.getTaskCount(), "Task count"); + assertEquals(10, queue.getUsedDataSize(), "Used data size"); + } + + private static final Duration TIMEOUT = Duration.ofSeconds(5); + private static final long BLOCKING_CHECK_MILLIS = 200; + + private static CompletableFuture submitAsync(BlockingScheduledRetryableTaskQueue queue, + ScheduledRetryableTask task) { + CompletableFuture future = new CompletableFuture<>(); + Thread thread = new Thread(() -> { + try { + queue.submit(task); + future.complete(null); + } catch (Throwable e) { + future.completeExceptionally(e); + } + }); + thread.setDaemon(true); + thread.start(); + return future; + } + + private static void assertBlocked(CompletableFuture future, String message) { + assertThrows(TimeoutException.class, + () -> future.get(BLOCKING_CHECK_MILLIS, TimeUnit.MILLISECONDS), + message + " must block"); + assertFalse(future.isDone(), message + " must block"); + } + + private static void submitUnchecked(BlockingScheduledRetryableTaskQueue queue, + ScheduledRetryableTask task) { + try { + queue.submit(task); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + } + private static void pause(long millis) { try { Thread.sleep(millis); - } catch (InterruptedException e) { + } catch (InterruptedException ignored) { } }