PartitioningChannel: Allow to customize the Executor used for each Partition

[joshiste]

Expected Behavior

I want to block on work being dispatched while the thread for the partition is busy.

Current Behavior

The executors used for each partition have an unbounded queue, and thus, it doesn't block.

Context

• Possible OOM
• Don't want to pull elements from the persisted message Queue into memory too early.

[artembilan]

I think the behavior you expect should be a default one.
I had doubts about losing data when developed PartitionedChannel, but somehow missed the fact about unbounded queue.

Please, share with us an idea how to make those executors blocking.
I'm not sure, though, if this has to be configurable since it is really better to avoid data loss.
And at the same time I wonder if there are any other options on those executors which need to be adjusted from the target application.

Perhaps making them blocking would be enough as a fix for this issue which I treat as a bug to be back-ported to all the supported version?

Thanks

[artembilan]

Looks like instead of Executors.newSingleThreadExecutor(this.threadFactory), we should use ThreadPoolExecutor constructor directly.
Something like this might work:

new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(1),
                                    threadFactory,
                                    new CallerBlocksPolicy(Long.MAX_VALUE))

Doesn't looks like there are any other options which could be customized on this ThreadPoolExecutor.

[joshiste]

@artembilan I think the queue should have zero length and you'd need to use a blocking rejection handler (e.g. RejectedExecutionHandlers.block())

But I suggest that users should provide an executor factory instead of a thread factory. ... Then you have great power ... and great responsibility.

[artembilan]

That queue cannot have a zero length:

    public LinkedBlockingQueue(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);
    }

Do you mean any other queue impl?
For example, SynchronousQueue ?

I'm not sure what is RejectedExecutionHandlers.block().
I see only one from Netty, but that does not have block() factory.

I don't think the goal of the PartitioningChannel is to let end-user to have the whole power over task executor.
That way we just going easy to lose the partition ordering goal of this channel implementation.
For full control we have a generic ExecutorChannel.

[joshiste]

That queue cannot have a zero length:

    public LinkedBlockingQueue(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);
    }

Do you mean any other queue impl? For example, SynchronousQueue ?

Ah yes. Wasn't digging deep enough into our code, where we use the org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor. That one uses a SynchronousQueue when setting the queue size to zero. So yes I believe it should be a SynchronousQueue.

I'm not sure what is RejectedExecutionHandlers.block(). I see only one from Netty, but that does not have block() factory.

Ah sorry! Something from our own code. It basically does the same as your CallerBlocksPolicy