[FIX_KAFKA_EXECUTOR] fix executor service

[elguardian]

The current approach behaves like monothread.

[sonarqubecloud]

Quality Gate passed

Issues
3 New issues
0 Accepted issues

Measures
0 Security Hotspots
100.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarCloud

[martinweiler]

This works properly, thanks @elguardian

[porcelli]

LGTM

[fjtirado]

I do not think is a good idea to create a fixed number of thread.
According to ThreadPoolExecutor documentation, new threads will be created only if needed. And removed once they are idle for a while.
With this change, we might have a number of threads forever idle when there is not event burst (also the name of the property will be inaccurate)
Lets discuss it a bit to find out why the pool is not working as expected right now, I need to understand that.

[fjtirado]

Ok, I misundertood the doc (bold is mine)
Unbounded queues. Using an unbounded queue (for example a LinkedBlockingQueue without a predefined capacity) will cause new tasks to wait in the queue when all corePoolSize threads are busy. Thus, no more than corePoolSize threads will ever be created. (And the value of the maximumPoolSize therefore doesn't have any effect.) This may be appropriate when each task is completely independent of others, so tasks cannot affect each others execution; for example, in a web page server. While this style of queuing can be useful in smoothing out transient bursts of requests, it admits the possibility of unbounded work queue growth when commands continue to arrive on average faster than they can be processed.
I believed it will create threads till maximum is reached and then use the unbounded queue (when the maximun numbers of thread), but its unfortunately the other way around (and that is not configurable, apparently). Fixed number of core threads are a indeed the easiest solution (the new code is equivalent to

           new ThreadPoolExecutorInteger.getInteger(KAFKA_EXTENSION_PREFIX + "maxNotifyThreads", 10), Integer.getInteger(KAFKA_EXTENSION_PREFIX + "maxNotifyThreads", 10),
                                60L,
                                TimeUnit.SECONDS, new LinkedBlockingQueue<>()

),
so lets go for it

And together with
By default, even core threads are initially created and started only when new tasks arrive
makes the property name still accurate (the only caveat if that, once created, the thread will remain there forever, but its a good tradeoff and definetely fixes a bug provoked by my misreading of the doc, my apologies for that)