Debug log level where appropriate + Throwing an exception where appropriate

Change-Id: Ica18c122c8a2cb8286d4f48c017a06ffe10452e1

@@ -135,7 +135,7 @@ public class ClusterManager implements ClusterService, ClusterAdminService {
                                         nodeInfo.getTcpPort()))
                             .collect(Collectors.toSet());
         } catch (IOException e) {
-            log.warn("Failed to read cluster definition.", e);
+            throw new IllegalStateException("Failed to read cluster definition.", e);
         }
 
         seedNodes.forEach(node -> {
@@ -278,7 +278,7 @@ public class ClusterManager implements ClusterService, ClusterAdminService {
                 }
             });
         } catch (Exception e) {
-            log.trace("Failed to send heartbeat", e);
+            log.debug("Failed to send heartbeat", e);
         }
     }
 
@@ -295,7 +295,7 @@ public class ClusterManager implements ClusterService, ClusterAdminService {
         try {
             messagingService.sendAsync(remoteEp, HEARTBEAT_MESSAGE, messagePayload);
         } catch (IOException e) {
-            log.trace("Sending heartbeat to {} failed", remoteEp, e);
+            log.debug("Sending heartbeat to {} failed", remoteEp, e);
         }
     }
 

@@ -190,7 +190,7 @@ public class DistributedLeadershipManager implements LeadershipService {
 
     @Override
     public void runForLeadership(String path) {
-        log.trace("Running for leadership for topic: {}", path);
+        log.debug("Running for leadership for topic: {}", path);
         activeTopics.add(path);
         tryLeaderLock(path);
     }
@@ -358,7 +358,7 @@ public class DistributedLeadershipManager implements LeadershipService {
                     }
                 }
                 if (localNodeId.equals(nodeId) && !activeTopics.contains(path)) {
-                    log.info("Lock for {} is held by {} when it not running for leadership.", path, nodeId);
+                    log.debug("Lock for {} is held by {} when it not running for leadership.", path, nodeId);
                     try {
                         if (lockMap.remove(path, epoch)) {
                             log.info("Purged stale lock held by {} for {}", nodeId, path);

@@ -222,7 +222,7 @@ public class NettyMessagingService implements MessagingService {
                     try {
                         handler.handle(message);
                     } catch (Exception e) {
-                        log.warn("Failed to process message of type {}", type, e);
+                        log.debug("Failed to process message of type {}", type, e);
                     }
                 });
             }