Merge pull request #71 from christophd/pr/add-kafka-consumer-group

chore: Add Kafka consumer group setting

java/steps/yaks-jms/src/main/java/org/citrusframework/yaks/jms/connection/activemq/ActiveMQConnectionFactoryCreator.java

@@ -49,6 +49,6 @@ public ConnectionFactory create(Map<String, String> properties) {
 
     @Override
     public boolean supports(Class<?> type) {
-        return ActiveMQConnectionFactory.class.equals(type);
+        return "org.apache.activemq.ActiveMQConnectionFactory".equals(type.getName());
     }
 }

java/steps/yaks-jms/src/main/java/org/citrusframework/yaks/jms/connection/activemq/artemis/ActiveMQArtemisConnectionFactoryCreator.java

@@ -17,15 +17,13 @@
 
 package org.citrusframework.yaks.jms.connection.activemq.artemis;
 
-import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
-
-import org.citrusframework.yaks.jms.connection.ConnectionFactoryCreator;
-
 import javax.jms.ConnectionFactory;
 import javax.jms.JMSException;
-
 import java.util.Map;
 
+import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
+import org.citrusframework.yaks.jms.connection.ConnectionFactoryCreator;
+
 public class ActiveMQArtemisConnectionFactoryCreator implements ConnectionFactoryCreator {
 
     @Override
@@ -53,6 +51,6 @@ public ConnectionFactory create(Map<String, String> properties) {
 
     @Override
     public boolean supports(Class<?> type) {
-        return ActiveMQConnectionFactory.class.equals(type);
+        return "org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory".equals(type.getName());
     }
 }

java/steps/yaks-kafka/src/main/java/org/citrusframework/yaks/kafka/KafkaSteps.java

@@ -28,6 +28,7 @@
 import com.consol.citrus.kafka.endpoint.KafkaEndpoint;
 import com.consol.citrus.kafka.endpoint.KafkaEndpointBuilder;
 import com.consol.citrus.kafka.message.KafkaMessage;
+import com.consol.citrus.kafka.message.KafkaMessageHeaders;
 import io.cucumber.datatable.DataTable;
 import io.cucumber.java.en.Given;
 import io.cucumber.java.en.Then;
@@ -54,10 +55,12 @@ public void setConnection(DataTable properties) {
 
         String url = connectionProps.getOrDefault("url", "localhost:9092");
         String topic = connectionProps.getOrDefault("topic", "test");
+        String consumerGroup = connectionProps.getOrDefault("consumerGroup", KafkaMessageHeaders.KAFKA_PREFIX + "group");
 
         KafkaEndpointBuilder builder = new KafkaEndpointBuilder()
                 .server(url)
-                .topic(topic);
+                .topic(topic)
+                .consumerGroup(consumerGroup);
 
         kafka = builder.build();
     }

java/steps/yaks-kafka/src/test/resources/org/citrusframework/yaks/kafka/kafka-multiline.feature

@@ -0,0 +1,22 @@
+Feature: Kafka steps
+
+  Background:
+    Given Kafka connection
+        | url           | localhost:9092 |
+        | topic         | hello |
+        | consumerGroup | hello-group |
+
+  Scenario: Send and receive multiline body
+    When send message to Kafka with body
+      """
+      {
+        "message": "Hello from YAKS!"
+      }
+      """
+    Then expect message in Kafka with body
+      """
+      {
+        "message": "Hello from YAKS!"
+      }
+      """
+