Revise RetryTopicConfigurationSupport TaskExecutor Strategy

[tomazfernandes]

Currently we add a ThreadPoolTaskExecutor @Bean in RetryTopicConfigurationSupport. That might have undesirable side-effects such as making Spring Boot's auto configuration back off from providing a default TaskExecutor for users.

We should investigate this further and if necessary look for alternatives.

[tomazfernandes]

I started looking into this, and TBH I think it's a slightly worse scenario than what I'd have thought.

Turns out when there is a TaskExecutor bean in the context, Boot's auto configured task executor does back off, as we supposed it would. That might already be bad enough, since auto configuration enables many task executor configurations that users may be relying on.

But what I think is the worst part is that there's logic in the WebMvcAutoConfiguration#237 class that tries to fetch the auto configured task executor by name - since it's not there when we provide our own, it falls back to a SimpleAsyncTaskExecutor for the RequestMappingHandlerAdapter, which should be quite less performant than the default ThreadPoolTaskExecutor since it doesn't reuse threads.

I think this is really unexpected and weird behavior, but at least in my test application with Boot 2.6.6 that's how it goes.

I'm not sure what'd be a good move here - the less impactful thing I can think of would be if we'd go back to letting the WakingTimingAdjuster manage the TaskExecutor lifecycle, and maybe even having it instantiate it lazily on the first timing adjustment request.

Any thoughts?

Thanks.

[artembilan]

You know I think the SimpleAsyncTaskExecutor would be a good compromise for a default settings.
I see we use it in many-many place in all the Spring projects.
Using a SimpleAsyncTaskExecutor we won't need to worry about its lifecycle.
The end-user would have a choice for other TaskExecutor impls to inject otherwise.

[tomazfernandes]

Oh, that sounds great really. TBH, I have no clue of how frequently this timing adjuster kicks in in real life scenarios - it depends on too many factors. But as an overall default dependency for the feature, a full fledged ThreadPoolTaskExecutor indeed seems like too much.

I'll do some testing to see if it has any impact on actual precision and submit a PR with this change.

Please let me know if you think of anything else.

Thanks!

[garyrussell]

Actually, a SATE is probably not good here; especially with long delays; we could end up with many threads sleeping.

On the other hand too small a pool could cause some short delay to be queued behind longer delays for other partitions.

Maybe in 3.0, we should look at changing the logic to use a TaskScheduler instead, rather than sleeping the executor thread?

[tomazfernandes]

Hmm, I didn't really know this TaskScheduler abstraction, looks interesting! I'll take a better look into it, but it does seem like a more proper approach than the current sleeping solution.

For clarification, the amount of time sleeping isn't actually related to the delay size. Every time we receive a PartitionIdleEvent for a backed off partition, we check if we're within 2 x pollTimeOut away from the dueTimestamp. If we are, we calculate how off we are from arriving at the due time in increments of pollTimeouts (long adjustmentAmount = timeUntilDue % pollTimeout), and then we sleep for this difference.

Let's say pollTimeout is 5000, and we're 5500 ms away from the due time, we're going to sleep for 500 ms only. Of course, the larger the pollTimeout, the bigger the adjustment can get.

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/WakingKafkaConsumerTimingAdjuster.java

Lines 114 to 127 in e1160fa

	public long adjustTiming(Consumer<?, ?> consumerToAdjust, TopicPartition topicPartition,
	long pollTimeout, long timeUntilDue) {
	boolean isInAdjustmentWindow = timeUntilDue > pollTimeout && timeUntilDue <=
	pollTimeout * this.pollTimeoutsForAdjustmentWindow;
	long adjustmentAmount = timeUntilDue % pollTimeout;
	if (isInAdjustmentWindow && adjustmentAmount > this.timingAdjustmentThreshold.toMillis()) {
	this.taskExecutor.execute(() ->
	doApplyTimingAdjustment(consumerToAdjust, topicPartition, adjustmentAmount));
	return adjustmentAmount;
	}
	return 0L;
	}

A different solution might be having a single thread that runs through a list and wakes up the proper consumers at the proper time, but it seems the TaskScheduler would be simpler.

[tomazfernandes]

Also, there's a minimal threshold of 100ms - adjustments below that won't be triggered. All these numbers are configurable, but I think most users should not really toy with these, that's why I didn't add it to the docs.

[tomazfernandes]

About this, considering all that's been said, I think it's worth it using the SimpleAsyncTaskExecutor for 2.9, since with that we'd no longer keep Spring Boot's Executor from auto configuring correctly. Also, if I understand correctly, we'd not have the need to manage the Executor lifecycle.

Another option would be registering the WakingTimingAdjuster as a bean instead of the Task Executor and have it create and manage the Executor's lifecycle directly.

For 3.0 we can change the implementation to rely on the TaskScheduler instead.

WDYT?

Thanks

[tomazfernandes]

About this, considering all that's been said, I think it's worth it using the SimpleAsyncTaskExecutor for 2.9, since with that we'd no longer keep Spring Boot's Executor from auto configuring correctly. Also, if I understand correctly, we'd not have the need to manage the Executor lifecycle.

Another option would be registering the WakingTimingAdjuster as a bean instead of the Task Executor and have it create and manage the Executor's lifecycle directly.

For 3.0 we can change the implementation to rely on the TaskScheduler instead.

WDYT?

Thanks

[artembilan]

Even if a TaskScheduler solution is better, we still have a problem with abusing auto-configuration. It is fully similar to what we have discussed above about a TaskExecutor.

I guess I need to investigate more in Spring Boot what and how they do with many TaskExecutor/TaskScheduler beans.

I recently changes Spring Integration to rely on the auto-configured TaskScheduler, but I cannot say yet what is going with the AbstractMessageBrokerConfiguration and its TaskExecutor/TaskScheduler beans...

[tomazfernandes]

How about the idea of turning the WakingTimingAdjuster to a bean and having it manage the TaskExecutor directly? This way we leave auto configuration issues out, and can change the implementation later.

The issue I see with that is that currently we have an option of not having a TimingAdjuster - not sure if we can still have that with a @Bean declaration without using some @Conditional.

[artembilan]

Well, the bean method can return null. Therefore a dependency on this bean can be marked with @Nullable.

[tomazfernandes]

Nice, I wasn't sure if that was a proper solution, thanks. I think since we have this auto configuration problems and with some not-so-great consequences, it's worth it shielding the solution from that. We can always revise it later for 3.0 if we need to.

If that sounds good, I can open a PR for us to look at, probably with the TimingAdjuster implementing InitializingBean and DisposableBean, and being registered as a @Bean in the RetryTopicConfigurationSupport class.