[GH-53] Fixed BlockingScheduledRetryableTaskQueue.submit() blocking forever on oversized task (#53)

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<V> 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:

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

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;
@@ -24,7 +27,8 @@
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
 public class BlockingScheduledRetryableTaskQueue<V> {
+    private static final Logger LOGGER = LoggerFactory.getLogger(BlockingScheduledRetryableTaskQueue.class);
 
     /**
      * GuardedBy {@link #lock}
@@ -82,31 +86,48 @@
     /**
      * Submits task to the queue.<BR>
      * 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.
+     * <P>
+     * 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.
+     * </P>
      *
      * @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<V> task) {
+    public void submit(ScheduledRetryableTask<V> 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 {
             lock.unlock();
         }
     }
 
+    /**
+     * GuardedBy {@link #lock}
+     */
+    private boolean fitsDataCapacity(long payloadSize) {
+        return dataSize == 0 || maxDataSize - dataSize >= payloadSize;
+    }
+
 
     /**
      * Submits task for retry to the queue.<BR>
@@ -207,6 +228,7 @@
                     ScheduledRetryableTask<V> job = taskQueue.peek();
                     long now = System.nanoTime();
                     if (now < job.getScheduledTime())
+                        //noinspection ResultOfMethodCallIgnored
                         addition.awaitNanos(job.getScheduledTime() - now);
                     else
                         return taskQueue.poll();
@@ -224,14 +246,14 @@
     }
 
     /**
      * @return Maximum number of uncompleted tasks that can be submitted before blocking
      */
     public int getMaxTaskCount() {
         return maxTaskCount;
     }
 
     /**
      * @return Current number of tasks in the queue
      */
     public int getTaskCount() {
         lock.lock();
@@ -243,7 +265,7 @@
     }
 
     /**
      * @return Current number of tasks in the queue that are not extracted yet
      */
     public int getQueuedTaskCount() {
         lock.lock();
@@ -255,7 +277,7 @@
     }
 
     /**
      * @return Maximum cumulative data size for all tasks that can be submitted before blocking
      */
     public long getMaxDataSize() {
         lock.lock();
@@ -267,20 +289,23 @@
     }
 
     /**
-     * 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();
         }
     }
 
     /**
      * @return Current cumulative data size of submitted tasks
      */
     public long getUsedDataSize() {
         lock.lock();

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<Object> task1 = new ScheduledRetryableTask<>(START_TIME, 1, 100, null);
         ScheduledRetryableTask<Object> 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<Object> queue = new BlockingScheduledRetryableTaskQueue<>(10, 100, null);
+        ScheduledRetryableTask<Object> 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<Object> queue = new BlockingScheduledRetryableTaskQueue<>(10, 100, null);
+        ScheduledRetryableTask<Object> smallTask = new ScheduledRetryableTask<>(System.nanoTime(), 1, 10, null);
+        ScheduledRetryableTask<Object> oversizedTask = new ScheduledRetryableTask<>(System.nanoTime(), 1, 150, null);
+        queue.submit(smallTask);
+
+        CompletableFuture<Void> 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<Object> queue = new BlockingScheduledRetryableTaskQueue<>(10, 100, null);
+        ScheduledRetryableTask<Object> oversizedTask = new ScheduledRetryableTask<>(System.nanoTime(), 1, 150, null);
+        ScheduledRetryableTask<Object> smallTask = new ScheduledRetryableTask<>(System.nanoTime(), 1, 10, null);
+        assertTimeoutPreemptively(TIMEOUT, () -> queue.submit(oversizedTask), "Oversized task submission to empty queue");
+
+        CompletableFuture<Void> 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<Object> queue = new BlockingScheduledRetryableTaskQueue<>(10, 100, null);
+        ScheduledRetryableTask<Object> firstTask = new ScheduledRetryableTask<>(System.nanoTime(), 1, 80, null);
+        ScheduledRetryableTask<Object> secondTask = new ScheduledRetryableTask<>(System.nanoTime(), 1, 80, null);
+        queue.submit(firstTask);
+
+        CompletableFuture<Void> 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<Object> queue = new BlockingScheduledRetryableTaskQueue<>(10, 100, null);
+        queue.submit(new ScheduledRetryableTask<>(System.nanoTime(), 1, 80, null));
+
+        CompletableFuture<Void> 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<Object> queue = new BlockingScheduledRetryableTaskQueue<>(1, 100, null);
+        queue.submit(new ScheduledRetryableTask<>(System.nanoTime(), 1, 10, null));
+
+        CompletableFuture<Throwable> 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<Void> submitAsync(BlockingScheduledRetryableTaskQueue<Object> queue,
+                                                       ScheduledRetryableTask<Object> task) {
+        CompletableFuture<Void> 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<Object> queue,
+                                        ScheduledRetryableTask<Object> 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) {
 
         }
     }