[camel-kafka-connector] branch master updated: Prevent NPEs on shutdown when critical errors occur

[acosentino]

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

acosentino pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git


The following commit(s) were added to refs/heads/master by this push:
     new 4e83421  Prevent NPEs on shutdown when critical errors occur
     new 9b5d415  Merge pull request #377 from orpiske/fix-camel-stop-npe
4e83421 is described below

commit 4e8342176d5ab8cad63c144f6ac1bcca587f009c
Author: Otavio Rodolfo Piske <opi...@redhat.com>
AuthorDate: Mon Aug 17 13:51:04 2020 +0200

    Prevent NPEs on shutdown when critical errors occur
---
 .../org/apache/camel/kafkaconnector/CamelSinkTask.java   | 10 +++++++++-
 .../org/apache/camel/kafkaconnector/CamelSourceTask.java | 16 ++++++++++++++--
 .../org/apache/camel/kafkaconnector/DataFormatTest.java  |  2 +-
 3 files changed, 24 insertions(+), 4 deletions(-)

diff --git 
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java 
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
index 57a15ba..c6242cb 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
@@ -153,7 +153,15 @@ public class CamelSinkTask extends SinkTask {
     public void stop() {
         LOG.info("Stopping CamelSinkTask connector task");
         try {
-            cms.stop();
+            if (cms != null) {
+                /*
+                  If the CamelMainSupport instance fails to be instantiated 
(ie.: due to missing classes or similar
+                  issues) then it won't be assigned and de-referencing it 
could cause an NPE.
+                 */
+                cms.stop();
+            } else {
+                LOG.warn("A fatal exception may have occurred and the Camel 
main was not created");
+            }
         } catch (Exception e) {
             throw new ConnectException("Failed to stop Camel context", e);
         } finally {
diff --git 
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java 
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
index acdba59..36aa96e 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
@@ -179,12 +179,24 @@ public class CamelSourceTask extends SourceTask {
     public void stop() {
         LOG.info("Stopping CamelSourceTask connector task");
         try {
-            consumer.stop();
+            if (consumer != null) {
+                consumer.stop();
+            } else {
+                LOG.warn("A critical error may have occurred and there is no 
consumer to stop");
+            }
         } catch (Exception e) {
             LOG.error("Error stopping camel consumer: {}", e.getMessage());
         }
         try {
-            cms.stop();
+            /*
+              If the CamelMainSupport instance fails to be instantiated (ie.: 
due to missing classes or similar
+              issues) then it won't be assigned and de-referencing it could 
cause an NPE.
+             */
+            if (cms != null) {
+                cms.stop();
+            } else {
+                LOG.warn("A fatal exception may have occurred and the Camel 
main was not created");
+            }
         } catch (Exception e) {
             throw new ConnectException("Failed to stop Camel context", e);
         } finally {
diff --git 
a/core/src/test/java/org/apache/camel/kafkaconnector/DataFormatTest.java 
b/core/src/test/java/org/apache/camel/kafkaconnector/DataFormatTest.java
index 6b7abf3..679d554 100644
--- a/core/src/test/java/org/apache/camel/kafkaconnector/DataFormatTest.java
+++ b/core/src/test/java/org/apache/camel/kafkaconnector/DataFormatTest.java
@@ -72,7 +72,7 @@ public class DataFormatTest {
 
         CamelSinkTask camelsinkTask = new CamelSinkTask();
         assertThrows(ConnectException.class, () -> camelsinkTask.start(props));
-        assertThrows(ConnectException.class, () -> camelsinkTask.stop());
+        // No need to check the stop method. The error is already 
thrown/caught during startup.
     }
 
     @Test