Fix sonar issues in dmaap-messagerouter-msgrtr

Fix Either re-interrupt this method or rethrow the "InterruptedException" issues
Fix Either log or rethrow this exception issues

Issue-ID: DMAAP-894
Change-Id: Ic0b296cf400c2529617e8ed9078ab370b10b623c
Signed-off-by: Parshad Patel <pars.patel@samsung.com>

3 files changed, 18 insertions, 17 deletions

diff --git a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java
index b66a251..b3f3c8f 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011Consumer.java
@@ -119,6 +119,7 @@ public class Kafka011Consumer implements Consumer {
 		} catch (InterruptedException x) {
 			log.warn("After size>0, pending msg take() threw InterruptedException. Ignoring. (" + x.getMessage() + ")",
 					x);
+			Thread.currentThread().interrupt();
 		}
 
 		Callable<Boolean> run = new Callable<Boolean>() {
@@ -135,11 +136,10 @@ public class Kafka011Consumer implements Consumer {
 					}
 
 				} catch (KafkaException x) {
-					log.debug(fLogTag + ": KafkaException " + x.getMessage());
+					log.debug(fLogTag + ": KafkaException ", x);
 
 				} catch (java.lang.IllegalStateException | java.lang.IllegalArgumentException x) {
-					log.error(fLogTag + ": Illegal state/arg exception in Kafka consumer; dropping stream. "
-							+ x.getMessage());
+					log.error(fLogTag + ": Illegal state/arg exception in Kafka consumer; dropping stream. ", x);
 
 				}
 
@@ -156,25 +156,26 @@ public class Kafka011Consumer implements Consumer {
 			future.get(consumerPollTimeOut, TimeUnit.SECONDS); // wait 1
 			// second
 		} catch (TimeoutException ex) {
+	        log.error("TimeoutException in in Kafka consumer ", ex);
 			// timed out. Try to stop the code if possible.
 			String apiNodeId = null;
 			try {
 				apiNodeId = InetAddress.getLocalHost().getCanonicalHostName() + ":" + CambriaConstants.kDefault_Port;
 			} catch (UnknownHostException e1) {
-				// TODO Auto-generated catch block
-				log.error("unable to get the localhost address");
+				log.error("unable to get the localhost address ", e1);
 			}
 
 			try {
 				if (fKafkaLiveLockAvoider != null)
 					fKafkaLiveLockAvoider.unlockConsumerGroup(apiNodeId, fTopic + "::" + fGroup);
 			} catch (Exception e) {
-				log.error("unlockConsumerGroup(" + apiNodeId + "," + fTopic + "::" + fGroup);
+				log.error("Exception in unlockConsumerGroup(" + apiNodeId + "," + fTopic + "::" + fGroup, e);
 			}
 
 			forcePollOnConsumer();
 			future.cancel(true);
 		} catch (Exception ex) {
+            log.error("Exception in in Kafka consumer ", ex);
 			// timed out. Try to stop the code if possible.
 			future.cancel(true);
 		}
@@ -307,13 +308,12 @@ public class Kafka011Consumer implements Consumer {
 			// second
 		} catch (TimeoutException ex) {
 			// timed out. Try to stop the code if possible.
-			log.info("Timeout Occured - Kafka connection closure with in 300 seconds by a Executors task");
+			log.info("Timeout Occured - Kafka connection closure with in 300 seconds by a Executors task ", ex);
 			future.cancel(true);
 			setState(Kafka011Consumer.State.OPENED);
 		} catch (Exception ex) {
 			// timed out. Try to stop the code if possible.
-			log.error("Exception occured Occured - Kafka connection closure with in 300 seconds by a Executors task"
-					+ ex);
+			log.error("Exception Occured - Kafka connection closure with in 300 seconds by a Executors task ", ex);
 			future.cancel(true);
 			setState(Kafka011Consumer.State.OPENED);
 			return false;
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011ConsumerUtil.java b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011ConsumerUtil.java
index a93ac33..9f8f26d 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011ConsumerUtil.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/Kafka011ConsumerUtil.java
@@ -72,7 +72,7 @@ public class Kafka011ConsumerUtil {
 					}
 
 				} catch (Exception e) {
-					log.error("Failed and go to Exception block for " + fGroup + " " + e.getMessage());
+					log.error("Failed and go to Exception block for " + fGroup +" ", e);
 				}
 			}
 		});
@@ -108,9 +108,9 @@ public class Kafka011ConsumerUtil {
 					}
 
 				} catch (java.util.ConcurrentModificationException e) {
-					log.error("Error occurs for " + e);
+					log.error("Error occurs for ", e);
 				} catch (Exception e) {
-					log.error("Failed and go to Exception block for " + group + " " + e.getMessage());
+					log.error("Failed and go to Exception block for " + group + " ", e);
 				}
 			}
 		});
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaConsumerCache.java b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaConsumerCache.java
index a38d77b..04d1d9e 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaConsumerCache.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/backends/kafka/KafkaConsumerCache.java
@@ -320,9 +320,9 @@ public class KafkaConsumerCache {
 				curatorConsumerCache.close();
 				log.info("Curator client closed");
 			} catch (ZkInterruptedException e) {
-				log.warn("Curator client close interrupted: " + e.getMessage());
+				log.warn("Curator client close interrupted: ", e);
 			} catch (IOException e) {
-				log.warn("Error while closing curator PathChildrenCache for KafkaConsumerCache" + e.getMessage());
+				log.warn("Error while closing curator PathChildrenCache for KafkaConsumerCache ", e);
 			}
 
 			curatorConsumerCache = null;
@@ -497,10 +497,10 @@ public class KafkaConsumerCache {
 			log.info(" ^ deleted " + fBaseZkPath + "/" + key);
 		} catch (NoNodeException e) {
 			log.warn("A consumer was deleted from " + fApiId
-					+ "'s cache, but no Cambria API node had ownership of it in ZooKeeper");
+					+ "'s cache, but no Cambria API node had ownership of it in ZooKeeper ", e);
 		} catch (Exception e) {
-			log.debug("Unexpected exception while deleting consumer: " + e.getMessage());
-			log.info(" %%%%%%@# Unexpected exception while deleting consumer: " + e.getMessage());
+			log.debug("Unexpected exception while deleting consumer: ", e);
+			log.info(" %%%%%%@# Unexpected exception while deleting consumer: ", e);
 		}
 
 		try {
@@ -648,6 +648,7 @@ public class KafkaConsumerCache {
 			try {
 				curator.setData().forPath(consumerPath, fApiId.getBytes());
 			} catch (KeeperException.NoNodeException e) {
+			    log.info("KeeperException.NoNodeException occured", e);
 				curator.create().creatingParentsIfNeeded().forPath(consumerPath, fApiId.getBytes());
 			}
 			log.info(fApiId + " successfully claimed ownership of consumer " + consumerKey);