From 69b652c71633d6f298b5adf3a0eb0ac7d4e3fe06 Mon Sep 17 00:00:00 2001
From: yatharthranjan <yatharthranjan89@gmail.com>
Date: Tue, 7 Nov 2017 11:09:42 +0000
Subject: [PATCH] Tuning the Client Ids

---
 .idea/.name                                   |  1 -
 .idea/compiler.xml                            | 23 +------------------
 .../monitor/AbstractKafkaMonitor.java         |  7 +++---
 .../radarcns/monitor/BatteryLevelMonitor.java |  2 +-
 .../monitor/KafkaMonitorFactoryTest.java      |  6 +++--
 5 files changed, 10 insertions(+), 29 deletions(-)
 delete mode 100644 .idea/.name

diff --git a/.idea/.name b/.idea/.name
deleted file mode 100644
index 1a452025..00000000
--- a/.idea/.name
+++ /dev/null
@@ -1 +0,0 @@
-radar-backend
\ No newline at end of file
diff --git a/.idea/compiler.xml b/.idea/compiler.xml
index 6e0a691c..69eaff7a 100644
--- a/.idea/compiler.xml
+++ b/.idea/compiler.xml
@@ -1,31 +1,10 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
   <component name="CompilerConfiguration">
-    <resourceExtensions />
-    <wildcardResourcePatterns>
-      <entry name="!?*.java" />
-      <entry name="!?*.form" />
-      <entry name="!?*.class" />
-      <entry name="!?*.groovy" />
-      <entry name="!?*.scala" />
-      <entry name="!?*.flex" />
-      <entry name="!?*.kt" />
-      <entry name="!?*.clj" />
-      <entry name="!?*.aj" />
-    </wildcardResourcePatterns>
-    <annotationProcessing>
-      <profile default="true" name="Default" enabled="false">
-        <processorPath useClasspath="true" />
-      </profile>
-    </annotationProcessing>
     <bytecodeTargetLevel>
-      <module name="commons_avroSources" target="1.7" />
-      <module name="commons_integrationTest" target="1.7" />
-      <module name="commons_main" target="1.7" />
-      <module name="commons_test" target="1.7" />
       <module name="radar-backend_integrationTest" target="1.8" />
       <module name="radar-backend_main" target="1.8" />
       <module name="radar-backend_test" target="1.8" />
     </bytecodeTargetLevel>
   </component>
-</project>
+</project>
\ No newline at end of file
diff --git a/src/main/java/org/radarcns/monitor/AbstractKafkaMonitor.java b/src/main/java/org/radarcns/monitor/AbstractKafkaMonitor.java
index 46380bec..798263e6 100644
--- a/src/main/java/org/radarcns/monitor/AbstractKafkaMonitor.java
+++ b/src/main/java/org/radarcns/monitor/AbstractKafkaMonitor.java
@@ -88,10 +88,11 @@ public AbstractKafkaMonitor(RadarPropertyHandler radar, Collection<String> topic
 
         properties = new Properties();
         String deserializer = KafkaAvroDeserializer.class.getName();
+        String monitorClientId = getClass().getName() + "-" + clientId;
         properties.setProperty(KEY_DESERIALIZER_CLASS_CONFIG, deserializer);
         properties.setProperty(VALUE_DESERIALIZER_CLASS_CONFIG, deserializer);
         properties.setProperty(GROUP_ID_CONFIG, groupId);
-        properties.setProperty(CLIENT_ID_CONFIG, clientId);
+        properties.setProperty(CLIENT_ID_CONFIG, monitorClientId);
         properties.setProperty(ENABLE_AUTO_COMMIT_CONFIG, "true");
         properties.setProperty(AUTO_COMMIT_INTERVAL_MS_CONFIG, "1001");
         properties.setProperty(SESSION_TIMEOUT_MS_CONFIG, "15101");
@@ -105,7 +106,7 @@ public AbstractKafkaMonitor(RadarPropertyHandler radar, Collection<String> topic
         this.topics = topics;
         this.pollTimeout = new AtomicLong(Long.MAX_VALUE);
         this.done = false;
-        this.clientId = clientId;
+        this.clientId = monitorClientId;
         this.groupId = groupId;
 
         PersistentStateStore localStateStore;
@@ -121,7 +122,7 @@ public AbstractKafkaMonitor(RadarPropertyHandler radar, Collection<String> topic
         S localState = stateDefault;
         if (stateStore != null && stateDefault != null) {
             try {
-                localState = stateStore.retrieveState(groupId, clientId, stateDefault);
+                localState = stateStore.retrieveState(groupId, monitorClientId, stateDefault);
                 logger.info("Using existing {} from persistence store.",
                         stateDefault.getClass().getName());
             } catch (IOException ex) {
diff --git a/src/main/java/org/radarcns/monitor/BatteryLevelMonitor.java b/src/main/java/org/radarcns/monitor/BatteryLevelMonitor.java
index 4e1d8738..f8474dd4 100644
--- a/src/main/java/org/radarcns/monitor/BatteryLevelMonitor.java
+++ b/src/main/java/org/radarcns/monitor/BatteryLevelMonitor.java
@@ -60,7 +60,7 @@ public class BatteryLevelMonitor extends
      */
     public BatteryLevelMonitor(RadarPropertyHandler radar, Collection<String> topics,
             EmailSender sender, Status minLevel, long logInterval) {
-        super(radar, topics, "battery_monitors", "2", new BatteryLevelState());
+        super(radar, topics, "battery_monitors", "1", new BatteryLevelState());
 
         Properties props = new Properties();
         props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
diff --git a/src/test/java/org/radarcns/monitor/KafkaMonitorFactoryTest.java b/src/test/java/org/radarcns/monitor/KafkaMonitorFactoryTest.java
index e858c98c..67b7fe22 100644
--- a/src/test/java/org/radarcns/monitor/KafkaMonitorFactoryTest.java
+++ b/src/test/java/org/radarcns/monitor/KafkaMonitorFactoryTest.java
@@ -55,7 +55,8 @@ public void createBatteryMonitor() throws Exception {
         assertEquals(BatteryLevelMonitor.class, monitor.getClass());
         BatteryLevelMonitor batteryMonitor = (BatteryLevelMonitor) monitor;
         batteryMonitor.evaluateRecords(new ConsumerRecords<>(Collections.emptyMap()));
-        assertTrue(new File(config.getPersistencePath(), "battery_monitors_2.yml").isFile());
+        assertTrue(new File(config.getPersistencePath(), "battery_monitors_" +
+                BatteryLevelMonitor.class.getName() + "-1.yml").isFile());
     }
 
     @Test(expected = IOException.class)
@@ -79,7 +80,8 @@ public void createDisconnectMonitor() throws Exception {
         assertEquals(DisconnectMonitor.class, monitor.getClass());
         DisconnectMonitor disconnectMonitor = (DisconnectMonitor) monitor;
         disconnectMonitor.evaluateRecords(new ConsumerRecords<>(Collections.emptyMap()));
-        assertTrue(new File(config.getPersistencePath(), "disconnect_monitor_1.yml").isFile());
+        assertTrue(new File(config.getPersistencePath(), "disconnect_monitor_" +
+                DisconnectMonitor.class.getName() + "-1.yml").isFile());
     }
 
     @Test