Merge pull request #36 from RADAR-CNS/fix_client_id

Fix client Id

Author: yatharthranjan

.idea/.name

@@ -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) {

.idea/compiler.xml

@@ -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_1.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
@@ -148,4 +150,4 @@ public static ConfigRadar getBatteryMonitorConfig(int port, TemporaryFolder fold
         config.setBatteryMonitor(getBatteryMonitorConfig(port));
         return config;
     }
-}
+}