++ I don't think we should have this. If we run into a Exception in the listener.onResponse we should fix the listener to handle that exception? Otherwise we're adding another situation where a listener behaves unexpectedly and we won't even see a log line for it?

I wonder if we should somehow assert that this never blocks? (since AsyncIOProcessor might be using blocking ArrayBlockingQueue#put on the item). It seems to me logically that's not an option with the current code, but maybe something we should make sure of to avoid (admittedly super unlikely) deadlocks in the future?

The current implementation is perfectly fine as we control the number of possible items in the queue. We can assert the remaining capacity before putting an item but this option subjects to a race condition. Another option is to use "offer" instead of "put" in AsyncIOProcessor if the assertion is enabled and blocking is not allowed. I will think about it more.

Yea, I didn't have a good idea on how to do this now either. Just figured I'd raise it :) Either way, once we start using this in tests that use DeterministicTaskQueue we'll probably be guarded against deadlocks/blocking anyway :)