log fix

Author: Robert Diers

src/main/java/org/opentesting/services/adapter/kafka/DelayedSender.java

@@ -16,8 +16,8 @@ public class DelayedSender implements Runnable {
 
     private String testid;
     private TestCaseInjectionDTO inject;
-    private Producer<String, String> producer;
-    private ProducerRecord<String, String> kafkarecord;    
+    private Producer<Object, Object> producer;
+    private ProducerRecord<Object, Object> kafkarecord;    
 
     @Override
     public void run() {

src/main/java/org/opentesting/services/adapter/kafka/Kafka.java

@@ -44,7 +44,7 @@ public class Kafka extends Adapter {
     private TaskScheduler taskScheduler;
 
     // producer cache
-    private HashMap<String, Producer<String, String>> producers = new HashMap<>();    
+    private HashMap<String, Producer<Object, Object>> producers = new HashMap<>();    
 
     @Override
     public String getServicename() {
@@ -55,7 +55,7 @@ public String getServicename() {
     @LogExecutionTime
     public boolean inject(String testid, TestCaseInjectionDTO inject) {
 
-        Producer<String, String> producer = null;
+        Producer<Object, Object> producer = null;
         String connectstr = "";
         try {
 
@@ -81,7 +81,7 @@ public boolean inject(String testid, TestCaseInjectionDTO inject) {
             String content = this.getFileAndAddTestData(testid, inject.getSourcefile(), inject.getRandomdata());
 
             // create record and add records
-            ProducerRecord<String, String> kafkarecord = new ProducerRecord<>(
+            ProducerRecord<Object, Object> kafkarecord = new ProducerRecord<>(
                     inject.getService().getCustom("topic").getValue(), UUID.randomUUID().toString(), content);
             headerMap.entrySet().forEach(es -> kafkarecord.headers().add(es.getKey(), es.getValue().getBytes()));
 
@@ -160,12 +160,12 @@ public boolean check(String testid, TestCaseCheckDTO check, Object... args) {
     }
 
     @LogExecutionTime
-    private Producer<String, String> getProducer(String testid, TestCaseServiceDTO service, String connectstr)
+    private Producer<Object, Object> getProducer(String testid, TestCaseServiceDTO service, String connectstr)
             throws ConnectFailedException {
 
         // no reuse because of different password configurations
         String key = this.createConnectionKey(testid, connectstr, service.getUsername(), service.getPassword());
-        Producer<String, String> prod = producers.get(key);
+        Producer<Object, Object> prod = producers.get(key);
 
         // block ones
         if (this.isFailedConnector(key)) {
@@ -222,8 +222,8 @@ public void createRequiredComponents(TestCaseDTO test) {
 
         // remove blocked connections and existing producers
         this.removeFailedConnectorStartingWith(test.getId());
-        HashMap<String, Producer<String, String>> newProducers = new HashMap<>();
-        for (Map.Entry<String,Producer<String, String>> entry : producers.entrySet()) {
+        HashMap<String, Producer<Object, Object>> newProducers = new HashMap<>();
+        for (Map.Entry<String,Producer<Object, Object>> entry : producers.entrySet()) {
             if (entry.getKey().startsWith(test.getId())) {
                 entry.getValue().close();
                 log.info("closed: " + entry.getValue().toString());

src/main/java/org/opentesting/services/adapter/kafka/KafkaTrace.java

@@ -107,7 +107,7 @@ private String getHeader(ConsumerRecord<String, String> data, String header, boo
      * add trace info to record
      * @param kafkarecord kafka record
      */
-    public void generateTraceInfo(ProducerRecord<String, String> kafkarecord) {
+    public void generateTraceInfo(ProducerRecord<Object, Object> kafkarecord) {
 
         // add trace information
         Span span = tracer.currentSpan();

src/main/java/org/opentesting/services/adapter/kafka/OpenTestingKafkaConsumer.java

@@ -50,6 +50,7 @@ public class OpenTestingKafkaConsumer {
     @Value("${kafka.reconnect.backoff.ms}")
     private int reconnectBackoffMs;
 
+    //we need to handle everything as a string to do simple string checks
     private Map<String, KafkaMessageListenerContainer<String, String>> createdConsumers = new HashMap<>();
 
     @LogExecutionTime

src/main/java/org/opentesting/services/adapter/kafka/OpenTestingKafkaMessageListener.java

@@ -61,7 +61,7 @@ public void onMessage(ConsumerRecord<String, String> data) {
             try {
                 return testCheck.executeCheck(c, false, adapter, trace, data.value(), Long.valueOf(data.timestamp()));
             } catch (Exception e) {
-                log.error("Kafka - check execution exception: "+e.getMessage(), e);
+                log.error(testid+": Kafka - check execution exception: "+e.getMessage(), e);
                 return null;
             }
         }).collect(Collectors.toList());

src/main/java/org/opentesting/services/adapter/kafka/OpenTestingKafkaProducerFactory.java

@@ -35,7 +35,7 @@ public class OpenTestingKafkaProducerFactory {
     private int reconnectBackoffMs;
 
     @LogExecutionTime
-    public Producer<String,String> createProducer(String testid, TestCaseServiceDTO service) {        
+    public Producer<Object,Object> createProducer(String testid, TestCaseServiceDTO service) {        
 
         //default properties
         Properties props = new Properties();        

src/main/java/org/opentesting/services/adapter/kafka/README.md

@@ -5,7 +5,7 @@ Using consumer and producer to perform inject and check
 ### Configuration Inject
 
 * please use sourcefile to store message to send
-* StringSerializer is default
+* StringSerializer is default, should be fine as we send file content directly
 * please use empty user and password for unprotected topics
 * random data replacement processed for: sourcefile
 
@@ -23,7 +23,7 @@ Using consumer and producer to perform inject and check
 | jwtpassword | might be used to specify the password in dedicated field (optional) | myPossibillyEncryptedPassword |
 | kafkaheaderjwt | define the kafka header key for requested JWT (optional) | accessToken |
 | senddelay | milliseconds to delay the send, helpful if kafka consumer initialization is to slow (default: not used) | 15000 |
-| value.serializer | you can use any of the available Kafka producer properties (optional) | org.apache.kafka.common.serialization.StringSerializer |
+| producer property | you can use any of the available Kafka producer properties (optional) | see link |
 
 https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html
 
@@ -90,7 +90,7 @@ https://docs.confluent.io/platform/current/installation/configuration/producer-c
 ### Configuration Check
 
 * validation 'request' not required
-* StringDeserializer is default
+* StringDeserializer is default, should be fine as we do only String validations
 * please use validation 'response' to define expected JSON result(s)
 * expectedtype could be 'equals', 'contains', 'containsnot', 'containsoneof'
 * random data replacement processed for: 'response'
@@ -106,7 +106,7 @@ https://docs.confluent.io/platform/current/installation/configuration/producer-c
 | sasl.mechanism | sasl mechanism used in combination with user and password (optional) | SCRAM-SHA-512 |
 | login.module | login module used in combination with user and password (optional) | org.apache.kafka.common.security.scram.ScramLoginModule |
 | stopexpectedlog | hide check compare log actual vs expected (required for high load topics), default = false | true |
-| value.deserializer | you can use any of the available Kafka consumer properties (optional) | org.apache.kafka.common.serialization.StringDeserializer |
+| consumer property | you can use any of the available Kafka consumer properties (optional) | see link |
 
 https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html
 

src/test/resources/opentesting/systemtests/kafka_001/test.json

@@ -88,7 +88,7 @@
                },
                {
                   "key": "group.id",
-                  "value": "e2etests"
+                  "value": "kafka_001.check-kafka-1.group"
                }
             ]
          },

src/test/resources/opentesting/systemtests/kafka_002/check_kafka_1.txt

@@ -0,0 +1 @@
+#replaceme1#

src/test/resources/opentesting/systemtests/kafka_002/source_kafka_1.txt

@@ -0,0 +1,4 @@
+{
+    "hallo": "#replaceme1#",
+    "welt": "#replaceme2#"
+}