BACKEND: rules update notification through kafka

Author: jelyoussefi

src/main/java/com/intel/databackend/config/cloudfoundry/ServiceConfig.java

@@ -34,11 +34,12 @@ public class ServiceConfig implements ServiceConfigProvider {
     public static final String KAFKA_SERVICE_NAME = "kafka";
     public static final String KAFKA_SERVICE_URI = "uri";
     public static final String KAFKA_UPS_NAME = "kafka-ups";
-    public static final String KAFKA_UPS_TOPIC = "topic";
+    public static final String KAFKA_UPS_TOPICS = "topics";
     public static final String KAFKA_UPS_ENABLED = "enabled";
     public static final String KAFKA_UPS_PARTITIONS = "partitions";
     public static final String KAFKA_UPS_REPLICATION = "replication";
     public static final String KAFKA_UPS_TIMEOUT_MS = "timeout_ms";
+    public static final String KAFKA_OBSERVATIONS_TOPIC = "observations";
 
     public static final String ZOOKEEPER_BROKER_NAME = "zookeeper";
     public static final String ZOOKEEPER_BROKER_URI = "zk.cluster";
@@ -101,7 +102,11 @@ public Boolean isKafkaEnabled() throws VcapEnvironmentException {
 
     @Override
     public String getKafkaTopicName() throws VcapEnvironmentException {
-        return getFieldValueFromJson(kafkaSettings, KAFKA_UPS_NAME, KAFKA_UPS_TOPIC, String.class);
+     	try {
+            return getFieldValueFromJson(kafkaSettings.getJSONObject(KAFKA_UPS_TOPICS), KAFKA_UPS_TOPICS, KAFKA_OBSERVATIONS_TOPIC, String.class);
+        } catch (JSONException e) {
+            throw new VcapEnvironmentException("Cannot get kafka topic name", e);
+        }
     }
 
     @Override

src/test/java/com/intel/databackend/config/cloudfoundry/ServiceConfigTest.java

@@ -42,7 +42,7 @@ public void Invoke_getUserProvidedServiceCredentialsByName() throws VcapEnvironm
         String topic = "metrics";
 
         Mockito.when(vcapReaderServices.getUserProvidedServiceCredentialsByName(ServiceConfig.KAFKA_UPS_NAME))
-                .thenReturn(new JSONObject("{" + ServiceConfig.KAFKA_UPS_TOPIC + ": " + topic + "}"));
+                .thenReturn(new JSONObject("{" + ServiceConfig.KAFKA_UPS_TOPICS + ": " + "{" + ServiceConfig.KAFKA_OBSERVATIONS_TOPIC + ": " +topic + "} }"));
         serviceConfig.init();
         Assert.assertEquals(serviceConfig.getKafkaTopicName(), topic);
     }
@@ -85,7 +85,7 @@ public void Throws_error_when_response_is_empty() throws VcapEnvironmentExceptio
     @Test(expected = VcapEnvironmentException.class)
     public void Throws_error_when_response_not_contain_key() throws VcapEnvironmentException, JSONException {
         Mockito.when(vcapReaderServices.getVcapServiceCredentialsByType(ServiceConfig.KAFKA_SERVICE_NAME))
-                .thenReturn(new JSONObject("{" + ServiceConfig.KAFKA_UPS_TOPIC + ": test}"));
+                .thenReturn(new JSONObject("{" + ServiceConfig.KAFKA_UPS_TOPICS + ": " + "{" + ServiceConfig.KAFKA_OBSERVATIONS_TOPIC + ": test} }"));;
         serviceConfig.init();
         serviceConfig.getKafkaUri();
     }