RedisQueueMessageDrivenEndpoint stuck in waiting thread state

[fred84]

Preconditions

• RedisQueueMessageDrivenEndpoint
• DirectChannel as output
• SimpleAsyncTaskExecutor with low concurrency level as executor (e.g. 1)

Scenario

• submit invalid message to queue

Current behavior

• exception will be not propagated to error handler
• new messages will be not consumed from queue

Example

https://github.com/fred84/integration_redis_stuck_thread

Proposal

• clarify in documentation that using SimpleAsyncTaskExecutor with low concurrency level is bad idea. Especially with concurrency == 1
• emit warning when RedisQueueMessageDrivenEndpoint configured with SimpleAsyncTaskExecutor and low concurrency level

[artembilan]

I see what you mean and thank you for the sample to play with.

You know, that's definitely not a RedisQueueMessageDrivenEndpoint problem. You introduce such a dead-lock yourself using SimpleAsyncTaskExecutor with a concurrencyLimit = 1.
See the source code of the ConcurrencyThrottleSupport:

while (this.concurrencyCount >= this.concurrencyLimit) {
	if (interrupted) {
		throw new IllegalStateException("Thread was interrupted while waiting for invocation access, " +
						"but concurrency limit still does not allow for entering");
	}
	if (debug) {
		logger.debug("Concurrency count " + this.concurrencyCount +
				" has reached limit " + this.concurrencyLimit + " - blocking");
	}
	try {
		this.monitor.wait();
	}
	catch (InterruptedException ex) {
			// Re-interrupt current thread, to allow other threads to react.
			Thread.currentThread().interrupt();
			interrupted = true;
	}
}

Since we are restarting the task in the used thread:

try {
	while (RedisQueueMessageDrivenEndpoint.this.active) {
		RedisQueueMessageDrivenEndpoint.this.listening = true;
		RedisQueueMessageDrivenEndpoint.this.popMessageAndSend();
	}
}
finally {
	if (RedisQueueMessageDrivenEndpoint.this.active) {
		RedisQueueMessageDrivenEndpoint.this.restart();
	}
	else if (RedisQueueMessageDrivenEndpoint.this.stopCallback != null) {
		RedisQueueMessageDrivenEndpoint.this.stopCallback.run();
		RedisQueueMessageDrivenEndpoint.this.stopCallback = null;
	}
}

It is indeed a dead-lock on that this.monitor.wait();

You can overcome issue adding an errorChannel to the RedisQueueMessageDrivenEndpoint or as you know already with a TaskExecutor with concurrency more than 1.
And that is definitely mentioned in the appropriate documentation.

Of course we can add some documentation WARN on the matter, but that's definitely feels much better to request Spring Framework to detect such a situation (we try to execute one more time at the same thread) in the mentioned ConcurrencyThrottleSupport rather then pollute all the users with extra checks and logs.