BlockingScheduledRetryableTaskQueue.submit() blocks forever when a task's payload size exceeds maxDataSize #53

BlockingScheduledRetryableTaskQueue.submit() waits until the task fits in the remaining data capacity:

while (true) {
    long capacityLeft = maxDataSize - dataSize;
    if (capacityLeft >= task.getPayloadSize() && taskSet.size() < maxTaskCount) {
        dataSize += task.getPayloadSize();
        addTask(task);
        break;
    } else {
        removal.awaitUninterruptibly();
    } 
}

If task.getPayloadSize() > maxDataSize, the condition can never be true, even when the queue is empty (dataSize == 0). The calling thread then:

• waits forever (awaitUninterruptibly, so it can't be interrupted, not even on shutdown);
• logs nothing and throws nothing, so the caller can't tell it's stuck.

The same thing can happen to tasks that were accepted earlier if setMaxDataSize() is later called with a value smaller than a pending task's payload size.

Version: 0.1.4

Steps to reproduce

var queue = new BlockingScheduledRetryableTaskQueue(10, 100, null);
queue.submit(new ScheduledRetryableTask<>(System.nanoTime(), 1, 101, new Object())); // never returns

Expected behavior

An oversized task must not block the caller forever. Possible options:

1. Fail fast: throw IllegalArgumentException from submit() when payloadSize > maxDataSize. The caller can then reject or log the item.
2. Accept an oversized task when the queue is empty (dataSize == 0). It then uses all the capacity until it completes. Processing continues, and the memory limit is still respected as far as it can be.

Option 1 is the most predictable. Option 2 is friendlier for callers that can't reject data. Either way:

• Use removal.await() instead of awaitUninterruptibly(), so callers can unblock a waiting submit() during shutdown.
• Document the behavior in the submit() Javadoc.

Impact

th2-estore:5.10.0-dev calls submit() from the RabbitMQ consumer thread (EventPersistor.persist()). One event larger than maxTaskDataSize blocks the consumer for good. The message is never acked, so the broker closes the
channel when its consumer ack timeout runs out (PRECONDITION_FAILED - delivery acknowledgement ... timed out), then redelivers the message and the hang starts again.

See the linked th2-net/th2-estore#164 issue