Couldn't we use a `WithinThreadExecutor` here? This constructor creates a new single-threaded executor, which would mean one thread is allocated for each cluster listenre.