[CAMEL-10069] Update to use ClassResolver to help search for the part…

…itioner and serializers

Also pull search for Partitioner out into separate try block to allow for use with 0.8 kafka client (which doesn't have partitioner)

components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConsumer.java

@@ -16,7 +16,6 @@
  */
 package org.apache.camel.component.kafka;
 
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -30,7 +29,6 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
-import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.InterruptException;
 import org.slf4j.Logger;

components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java

@@ -16,6 +16,7 @@
  */
 package org.apache.camel.component.kafka;
 
+import java.lang.reflect.Field;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.LinkedList;
@@ -30,15 +31,20 @@
 import org.apache.camel.CamelExchangeException;
 import org.apache.camel.Exchange;
 import org.apache.camel.impl.DefaultAsyncProducer;
+import org.apache.camel.spi.ClassResolver;
+import org.apache.camel.util.CastUtils;
 import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.serialization.Serializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class KafkaProducer extends DefaultAsyncProducer {
-
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaProducer.class);
+
     private org.apache.kafka.clients.producer.KafkaProducer kafkaProducer;
     private final KafkaEndpoint endpoint;
     private ExecutorService workerPool;
@@ -49,55 +55,63 @@ public KafkaProducer(KafkaEndpoint endpoint) {
         this.endpoint = endpoint;
     }
 
-
-    Class<?> loadClass(Object o, ClassLoader loader) {
+    <T> Class<T> loadClass(Object o, ClassResolver resolver, Class<T> type) {
         if (o == null || o instanceof Class) {
-            return (Class<?>)o;
+            return CastUtils.cast((Class<?>)o);
         }
         String name = o.toString();
-        Class<?> c;
-        try {
-            c = Class.forName(name, true, loader);
-        } catch (ClassNotFoundException e) {
-            c = null;
-        }
+        Class<T> c = resolver.resolveClass(name, type);
         if (c == null) {
-            try {
-                c = Class.forName(name, true, getClass().getClassLoader());
-            } catch (ClassNotFoundException e) {
-                c = null;
-            }
+            c = resolver.resolveClass(name, type, getClass().getClassLoader());
         }
         if (c == null) {
-            try {
-                c = Class.forName(name, true, org.apache.kafka.clients.producer.KafkaProducer.class.getClassLoader());
-            } catch (ClassNotFoundException e) {
-                c = null;
-            }
+            c = resolver.resolveClass(name, type, org.apache.kafka.clients.producer.KafkaProducer.class.getClassLoader());
         }
         return c;
     }
-    void replaceWithClass(Properties props, String key,  ClassLoader loader, Class<?> type) {
-        Class<?> c = loadClass(props.get(key), loader);
+    void replaceWithClass(Properties props, String key,  ClassResolver resolver, Class<?> type) {
+        Class<?> c = loadClass(props.get(key), resolver, type);
         if (c != null) {
             props.put(key, c);
         }
     }
 
     Properties getProps() {
         Properties props = endpoint.getConfiguration().createProducerProperties();
-        if (endpoint.getCamelContext() != null) {
-            ClassLoader loader = endpoint.getCamelContext().getApplicationContextClassLoader();
-            replaceWithClass(props, "key.serializer", loader, Serializer.class);
-            replaceWithClass(props, "value.serializer", loader, Serializer.class);
-            replaceWithClass(props, "partitioner.class", loader, Partitioner.class);
+        try {
+            if (endpoint.getCamelContext() != null) {
+                ClassResolver resolver = endpoint.getCamelContext().getClassResolver();
+                replaceWithClass(props, "key.serializer", resolver, Serializer.class);
+                replaceWithClass(props, "value.serializer", resolver, Serializer.class);
+
+                try {
+                    //doesn't exist in old version of Kafka client so detect and only call the method if
+                    //the field/config actually exists
+                    Field f = ProducerConfig.class.getDeclaredField("PARTITIONER_CLASS_CONFIG");
+                    if (f != null) {
+                        loadParitionerClass(resolver, props);
+                    }
+                } catch (NoSuchFieldException e) {
+                    //ignore
+                } catch (SecurityException e) {
+                    //ignore
+                }
+            }
+        } catch (Throwable t) {
+            //can ignore and Kafka itself might be able to handle it, if not, it will throw an exception
+            LOG.debug("Problem loading classes for Serializers", t);
         }
         if (endpoint.getBrokers() != null) {
             props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, endpoint.getBrokers());
         }
         return props;
     }
 
+    private void loadParitionerClass(ClassResolver resolver, Properties props) {
+        replaceWithClass(props, "partitioner.class", resolver, Partitioner.class);
+    }
+
+
     public org.apache.kafka.clients.producer.KafkaProducer getKafkaProducer() {
         return kafkaProducer;
     }
@@ -184,6 +198,11 @@ public ProducerRecord next() {
                     }
                     return new ProducerRecord(msgTopic, msgList.next());
                 }
+
+                @Override
+                public void remove() {
+                    msgList.remove();
+                }
             };
         }
         ProducerRecord record;

components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java

@@ -99,7 +99,7 @@ public static void after() {
     }
 
     @Override
-    protected RoutesBuilder createRouteBuilder() throws Exception {
+    protected RouteBuilder createRouteBuilder() throws Exception {
         return new RouteBuilder() {
             @Override
             public void configure() throws Exception {