Use correct ClassLoader for Kafka. (#297)

Signed-off-by: Greg Allen <[email protected]>

Author: ggallen

plugin/src/main/java/com/redhat/jenkins/plugins/ci/messaging/ActiveMqMessagingWorker.java

@@ -445,8 +445,8 @@ public SendResult sendMessage(Run<?, ?> build, TaskListener listener, ProviderDa
                 mesgId = message.getJMSMessageID();
                 mesgContent = message.getText();
 
-                log.info("Sent message for job '" + build.getParent().getName() + "' to " + kind + " '" + ltopic + "':\n"
-                        + formatMessage(message));
+                log.info("Sent message for job '" + build.getParent().getName() + "' to " + kind + " '" + ltopic
+                        + "':\n" + formatMessage(message));
             } else {
                 log.severe("One or more of the following is invalid (null): user, password, topic, broker.");
                 return new SendResult(false, mesgId, mesgContent);

plugin/src/main/java/com/redhat/jenkins/plugins/ci/messaging/KafkaMessagingProvider.java

@@ -181,12 +181,12 @@ public FormValidation doTestConnection(@QueryParameter("topic") String topic,
             Properties cprops = getMergedConsumerProperties(consumerProperties);
 
             ClassLoader original = Thread.currentThread().getContextClassLoader();
-            Thread.currentThread().setContextClassLoader(null);
+            Thread.currentThread().setContextClassLoader(KafkaConsumer.class.getClassLoader());
             try (KafkaConsumer consumer = new KafkaConsumer<>(cprops);
                     KafkaProducer producer = new KafkaProducer<>(pprops)) {
 
                 // Test producer.
-                ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "test-key", "test-value");
+                ProducerRecord<String, String> record = new ProducerRecord<>(topic, "test-key", "test-value");
                 producer.send(record).get();
 
                 // Test consumer.

plugin/src/main/java/com/redhat/jenkins/plugins/ci/messaging/KafkaMessagingWorker.java

@@ -190,7 +190,7 @@ public boolean connect() {
             // Invalid value org.apache.kafka.common.serialization.StringDeserializer for configuration
             // value.deserializer: Class org.apache.kafka.common.serialization.StringDeserializer could not be found.
             ClassLoader original = Thread.currentThread().getContextClassLoader();
-            Thread.currentThread().setContextClassLoader(null);
+            Thread.currentThread().setContextClassLoader(KafkaConsumer.class.getClassLoader());
             consumer = new KafkaConsumer<>(pdata.mergeProperties(provider.getMergedConsumerProperties()));
             Thread.currentThread().setContextClassLoader(original);
         }
@@ -222,7 +222,7 @@ public SendResult sendMessage(Run<?, ?> build, TaskListener listener, ProviderDa
         String body = "";
         String msgId = "";
         ClassLoader original = Thread.currentThread().getContextClassLoader();
-        Thread.currentThread().setContextClassLoader(null);
+        Thread.currentThread().setContextClassLoader(KafkaProducer.class.getClassLoader());
         try (KafkaProducer<String, String> producer = new KafkaProducer<>(
                 pd.mergeProperties(provider.getMergedProducerProperties()))) {
             EnvVars env = new EnvVars();
@@ -278,7 +278,7 @@ public String waitForMessage(Run<?, ?> build, TaskListener listener, ProviderDat
         }
 
         ClassLoader original = Thread.currentThread().getContextClassLoader();
-        Thread.currentThread().setContextClassLoader(null);
+        Thread.currentThread().setContextClassLoader(KafkaConsumer.class.getClassLoader());
         try (KafkaConsumer<String, String> lconsumer = new KafkaConsumer<>(
                 pd.mergeProperties(provider.getMergedConsumerProperties()))) {
             int timeout = (pd.getTimeout() != null ? pd.getTimeout()