[camel] branch camel-3.20.x updated: CAMEL-19650: Camel Kafka doesn't honor 'workerPool' configuration (#10827) (#10839)

[nfilotto]

This is an automated email from the ASF dual-hosted git repository.

nfilotto pushed a commit to branch camel-3.20.x
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/camel-3.20.x by this push:
     new 9e802133715 CAMEL-19650: Camel Kafka doesn't honor 'workerPool' 
configuration (#10827) (#10839)
9e802133715 is described below

commit 9e8021337159980e5a76c282eeaff6a878e7c0ff
Author: Nicolas Filotto <essob...@users.noreply.github.com>
AuthorDate: Wed Jul 26 17:26:46 2023 +0200

    CAMEL-19650: Camel Kafka doesn't honor 'workerPool' configuration (#10827) 
(#10839)
    
    * CAMEL-19650: Camel Kafka doesn't honor 'workerPool' configuration
    
    Camel kafka component when provided with custom worker pool is ignored and 
still creates a new pool, this is fixed with check added to use custom worker 
pool if provided.
    
    * CAMEL-19650: Camel Kafka doesn't honor 'workerPool' configuration
    
    Fixing checkstyle issue
    
    Co-authored-by: Kartik  kalaghatgi <karthikkalaghatgi...@gmail.com>
---
 .../java/org/apache/camel/component/kafka/KafkaProducer.java | 12 +++++++++---
 1 file changed, 9 insertions(+), 3 deletions(-)

diff --git 
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
 
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index 95afbde2960..d1dcaa30bbe 100755
--- 
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++ 
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -169,9 +169,15 @@ public class KafkaProducer extends DefaultAsyncProducer {
 
         // if we are in asynchronous mode we need a worker pool
         if (!configuration.isSynchronous() && workerPool == null) {
-            workerPool = endpoint.createProducerExecutor();
-            // we create a thread pool so we should also shut it down
-            shutdownWorkerPool = true;
+            // If custom worker pool is provided, then use it, else create a 
new one.
+            if (configuration.getWorkerPool() != null) {
+                workerPool = configuration.getWorkerPool();
+                shutdownWorkerPool = false;
+            } else {
+                workerPool = endpoint.createProducerExecutor();
+                // we create a thread pool so we should also shut it down
+                shutdownWorkerPool = true;
+            }
         }
 
         // init client id which we may need to get from the kafka producer via 
reflection