Test FTC100 fails since A1-SIM update

[nonrtric.git] / sample-services / ics-producer-consumer / producer / src / main / java / com / demo / producer / repository / Jobs.java

diff --git a/sample-services/ics-producer-consumer/producer/src/main/java/com/demo/producer/repository/Jobs.java b/sample-services/ics-producer-consumer/producer/src/main/java/com/demo/producer/repository/Jobs.java
index f9cffd8..e929e44 100644 (file)
--- a/sample-services/ics-producer-consumer/producer/src/main/java/com/demo/producer/repository/Jobs.java
+++ b/sample-services/ics-producer-consumer/producer/src/main/java/com/demo/producer/repository/Jobs.java
@@ -28,6 +28,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
 
+import com.demo.producer.messages.PropertiesHelper;
 import com.demo.producer.repository.Job.Parameters;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
@@ -55,9 +56,20 @@ public class Jobs {
 
     public void addJob(String id, InfoType type, String owner, Parameters parameters) {
         Job job = new Job(id, type, owner, parameters);
+        setKafkaServersEnvironment(job);
         this.put(job);
     }
 
+    private void setKafkaServersEnvironment(Job job) {
+        String kafkaServers = job.getParameters().getDeliveryInfo().getBootStrapServers();
+        if (kafkaServers != null && !kafkaServers.isEmpty()) {
+            PropertiesHelper.setKafkaServers(kafkaServers);
+            logger.info("Setting variable bootStrapServers: {}", kafkaServers);
+        } else {
+            logger.warn("bootStrapServers is not set for job: {}", job.getId());
+        }
+    }
+
     private synchronized void put(Job job) {
         logger.debug("Put job: {}", job.getId());
         allJobs.put(job.getId(), job);