Repository: authorjapps/zerocode
Pull request #173 (2018-12-24 13:38:14)

ISS-104 - Kafka stream testing


Patch:
diff --git a/pom.xml b/pom.xml
index 352551ce..0e3e3ddf 100644
--- a/pom.xml
+++ b/pom.xml
@@ -51,6 +51,7 @@
     </distributionManagement>
 
     <properties>
+        <!-- this project properties -->
         <!-- This project properties -->
         <junit.version>4.12</junit.version>
         <jackson-databind.version>2.9.7</jackson-databind.version>
@@ -71,18 +72,27 @@
         <httpclient.version>4.5</httpclient.version>
         <h2.db.version>1.4.191</h2.db.version>
         <extentreports.version>3.1.3</extentreports.version>
-
-        <!-- Maven plugin build related  -->
+        <version.kafka-clients>2.1.0</version.kafka-clients>
+        <!-- maven plugin, build properties  -->
         <maven-compiler-plugin.version>3.2</maven-compiler-plugin.version>
         <java-compiler-source.version>1.8</java-compiler-source.version>
         <java-compiler-target.version>1.8</java-compiler-target.version>
         <maven-assembly-plugin.version>2.5.4</maven-assembly-plugin.version>
-
-        <!-- Build will not fail if error occurs during javadoc generation -->
+        <!-- Release Build will not fail if error occurs during javadoc generation -->
         <maven.javadoc.failOnError>false</maven.javadoc.failOnError>
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+            <version>2.6.2</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>${version.kafka-clients}</version>
+        </dependency>
         <dependency>
             <groupId>org.json</groupId>
             <artifactId>json</artifactId>
diff --git a/src/main/java/org/jsmart/zerocode/core/di/SmartUtilsModule.java b/src/main/java/org/jsmart/zerocode/core/di/SmartUtilsModule.java
deleted file mode 100644
index bfefd20c..00000000
--- a/src/main/java/org/jsmart/zerocode/core/di/SmartUtilsModule.java
+++ /dev/null
@@ -1,4 +0,0 @@
-package org.jsmart.zerocode.core.di;
-
-public class SmartUtilsModule {
-}
diff --git a/src/main/java/org/jsmart/zerocode/core/di/ApplicationMainModule.java b/src/main/java/org/jsmart/zerocode/core/di/main/ApplicationMainModule.java
similarity index 75%
rename from src/main/java/org/jsmart/zerocode/core/di/ApplicationMainModule.java
rename to src/main/java/org/jsmart/zerocode/core/di/main/ApplicationMainModule.java
index eef3fc51..551692e6 100644
--- a/src/main/java/org/jsmart/zerocode/core/di/ApplicationMainModule.java
+++ b/src/main/java/org/jsmart/zerocode/core/di/main/ApplicationMainModule.java
@@ -1,8 +1,11 @@
-package org.jsmart.zerocode.core.di;
+package org.jsmart.zerocode.core.di.main;
 
 
 import com.google.inject.AbstractModule;
 import com.google.inject.name.Names;
+import org.jsmart.zerocode.core.di.module.GsonModule;
+import org.jsmart.zerocode.core.di.module.HttpClientModule;
+import org.jsmart.zerocode.core.di.module.ObjectMapperModule;
 import org.jsmart.zerocode.core.engine.executor.JavaExecutor;
 import org.jsmart.zerocode.core.engine.executor.JavaExecutorImpl;
 import org.jsmart.zerocode.core.engine.executor.JsonServiceExecutor;
@@ -37,6 +40,8 @@ public class ApplicationMainModule extends AbstractModule {
          */
         install(new ObjectMapperModule());
         install(new HttpClientModule());
+        install(new GsonModule());
+        //install(new KafkaModule());
 
         /*
          * Bind Direct classes, classes to interfaces etc
@@ -47,12 +52,10 @@ public class ApplicationMainModule extends AbstractModule {
         bind(ZeroCodeJsonTestProcesor.class).to(ZeroCodeJsonTestProcesorImpl.class);
         bind(ZeroCodeReportGenerator.class).to(ZeroCodeReportGeneratorImpl.class);
         bind(ZeroCodeExternalFileProcessor.class).to(ZeroCodeExternalFileProcessorImpl.class);
-        //bind(ExecutorServiceRunner.class).in(Singleton.class);
-        //bind(SmartUtils.class);
 
-        /*
-		 * Bind properties for localhost, CI, PRE-PROD etc
-		 */
+        // ------------------------------------------------
+        // Bind properties for localhost, CI, DIT, SIT etc
+        // ------------------------------------------------
         Names.bindProperties(binder(), getProperties(serverEnv));
     }
 
@@ -77,16 +80,19 @@ public class ApplicationMainModule extends AbstractModule {
 
     private void checkAndLoadOldProperties(Properties properties) {
 
-        if(properties.get(WEB_APPLICATION_ENDPOINT_HOST) == null){
-            properties.setProperty(WEB_APPLICATION_ENDPOINT_HOST, (String)properties.get(RESTFUL_APPLICATION_ENDPOINT_HOST));
+        if(properties.get(WEB_APPLICATION_ENDPOINT_HOST) == null && properties.get(RESTFUL_APPLICATION_ENDPOINT_HOST) != null){
+            Object oldPropertyValue = properties.get(RESTFUL_APPLICATION_ENDPOINT_HOST);
+            properties.setProperty(WEB_APPLICATION_ENDPOINT_HOST, oldPropertyValue != null ? oldPropertyValue.toString() : null);
         }
 
-        if(properties.get(WEB_APPLICATION_ENDPOINT_PORT) == null){
-            properties.setProperty(WEB_APPLICATION_ENDPOINT_PORT, (String)properties.get(RESTFUL_APPLICATION_ENDPOINT_PORT));
+        if(properties.get(WEB_APPLICATION_ENDPOINT_PORT) == null && properties.get(RESTFUL_APPLICATION_ENDPOINT_PORT) != null){
+            Object oldPropertyValue = properties.get(RESTFUL_APPLICATION_ENDPOINT_PORT);
+            properties.setProperty(WEB_APPLICATION_ENDPOINT_PORT, oldPropertyValue != null ? oldPropertyValue.toString() : null);
         }
 
-        if(properties.get(WEB_APPLICATION_ENDPOINT_CONTEXT) == null){
-            properties.setProperty(WEB_APPLICATION_ENDPOINT_CONTEXT, (String)properties.get(RESTFUL_APPLICATION_ENDPOINT_CONTEXT));
+        if(properties.get(WEB_APPLICATION_ENDPOINT_CONTEXT) == null && properties.get(RESTFUL_APPLICATION_ENDPOINT_CONTEXT) != null){
+            Object oldPropertyValue = properties.get(RESTFUL_APPLICATION_ENDPOINT_CONTEXT);
+            properties.setProperty(WEB_APPLICATION_ENDPOINT_CONTEXT, oldPropertyValue != null ? oldPropertyValue.toString() : null);
         }
 
     }
diff --git a/src/main/java/org/jsmart/zerocode/core/di/module/GsonModule.java b/src/main/java/org/jsmart/zerocode/core/di/module/GsonModule.java
new file mode 100644
index 00000000..3a8e4f2f
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/di/module/GsonModule.java
@@ -0,0 +1,18 @@
+package org.jsmart.zerocode.core.di.module;
+
+import com.google.gson.Gson;
+import com.google.inject.Binder;
+import com.google.inject.Module;
+import org.jsmart.zerocode.core.di.provider.GsonSerDeProvider;
+
+import javax.inject.Singleton;
+
+
+public class GsonModule implements Module {
+
+    @Override
+    public void configure(Binder binder) {
+        binder.bind(Gson.class).toProvider(GsonSerDeProvider.class).in(Singleton.class);
+    }
+}
+
diff --git a/src/main/java/org/jsmart/zerocode/core/di/HttpClientModule.java b/src/main/java/org/jsmart/zerocode/core/di/module/HttpClientModule.java
similarity index 76%
rename from src/main/java/org/jsmart/zerocode/core/di/HttpClientModule.java
rename to src/main/java/org/jsmart/zerocode/core/di/module/HttpClientModule.java
index 237ac9fe..b5fa1f7d 100644
--- a/src/main/java/org/jsmart/zerocode/core/di/HttpClientModule.java
+++ b/src/main/java/org/jsmart/zerocode/core/di/module/HttpClientModule.java
@@ -1,9 +1,9 @@
-package org.jsmart.zerocode.core.di;
+package org.jsmart.zerocode.core.di.module;
 
 import com.google.inject.Binder;
 import com.google.inject.Module;
+import org.jsmart.zerocode.core.di.provider.DefaultGuiceHttpClientProvider;
 import org.jsmart.zerocode.core.httpclient.BasicHttpClient;
-import org.jsmart.zerocode.core.httpclient.ssl.SslTrustHttpClient;
 
 import javax.inject.Singleton;
 
diff --git a/src/main/java/org/jsmart/zerocode/core/di/module/KafkaModule.java b/src/main/java/org/jsmart/zerocode/core/di/module/KafkaModule.java
new file mode 100644
index 00000000..927922d3
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/di/module/KafkaModule.java
@@ -0,0 +1,15 @@
+//package org.jsmart.zerocode.core.di.module;
+//
+//import com.google.inject.Binder;
+//import com.google.inject.Module;
+//import org.jsmart.zerocode.core.di.provider.KafkaServicesProvider;
+//import org.jsmart.zerocode.core.kafka.KafkaService;
+//
+//import javax.inject.Singleton;
+//
+//public class KafkaModule implements Module {
+//
+//    public void configure(Binder binder) {
+//        binder.bind(KafkaService.class).toProvider(KafkaServicesProvider.class).in(Singleton.class);
+//    }
+//}
\ No newline at end of file
diff --git a/src/main/java/org/jsmart/zerocode/core/di/ObjectMapperModule.java b/src/main/java/org/jsmart/zerocode/core/di/module/ObjectMapperModule.java
similarity index 51%
rename from src/main/java/org/jsmart/zerocode/core/di/ObjectMapperModule.java
rename to src/main/java/org/jsmart/zerocode/core/di/module/ObjectMapperModule.java
index 7c23422a..34f62289 100644
--- a/src/main/java/org/jsmart/zerocode/core/di/ObjectMapperModule.java
+++ b/src/main/java/org/jsmart/zerocode/core/di/module/ObjectMapperModule.java
@@ -1,8 +1,9 @@
-package org.jsmart.zerocode.core.di;
+package org.jsmart.zerocode.core.di.module;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.inject.Binder;
 import com.google.inject.Module;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 
 import javax.inject.Singleton;
 
@@ -15,16 +16,3 @@ public class ObjectMapperModule implements Module {
     }
 }
 
-
-
-/*
- * Another way : By extending from AbstractModule, "binder" is still available via binder()
- */
-
-//public class ObjectMapperModule extends AbstractModule {
-//
-//    @Override
-//    public void configure() {
-//        bind(ObjectMapper.class).toProvider(ObjectMapperProvider.class).in(Singleton.class);
-//    }
-//}
diff --git a/src/main/java/org/jsmart/zerocode/core/di/RuntimeHttpClientModule.java b/src/main/java/org/jsmart/zerocode/core/di/module/RuntimeHttpClientModule.java
similarity index 92%
rename from src/main/java/org/jsmart/zerocode/core/di/RuntimeHttpClientModule.java
rename to src/main/java/org/jsmart/zerocode/core/di/module/RuntimeHttpClientModule.java
index 7c315c69..e17618bb 100644
--- a/src/main/java/org/jsmart/zerocode/core/di/RuntimeHttpClientModule.java
+++ b/src/main/java/org/jsmart/zerocode/core/di/module/RuntimeHttpClientModule.java
@@ -1,4 +1,4 @@
-package org.jsmart.zerocode.core.di;
+package org.jsmart.zerocode.core.di.module;
 
 import com.google.inject.Binder;
 import com.google.inject.Module;
diff --git a/src/main/java/org/jsmart/zerocode/core/di/module/RuntimeKafkaClientModule.java b/src/main/java/org/jsmart/zerocode/core/di/module/RuntimeKafkaClientModule.java
new file mode 100644
index 00000000..798cc844
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/di/module/RuntimeKafkaClientModule.java
@@ -0,0 +1,18 @@
+package org.jsmart.zerocode.core.di.module;
+
+import com.google.inject.Binder;
+import com.google.inject.Module;
+import org.jsmart.zerocode.core.kafka.client.BasicKafkaClient;
+
+public class RuntimeKafkaClientModule implements Module {
+
+    private final Class<? extends BasicKafkaClient> customKafkaClientClazz;
+
+    public RuntimeKafkaClientModule(Class<? extends BasicKafkaClient> customKafkaClientClazz) {
+        this.customKafkaClientClazz = customKafkaClientClazz;
+    }
+
+    public void configure(Binder binder) {
+        binder.bind(BasicKafkaClient.class).to(customKafkaClientClazz);
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/org/jsmart/zerocode/core/di/module/SmartUtilsModule.java b/src/main/java/org/jsmart/zerocode/core/di/module/SmartUtilsModule.java
new file mode 100644
index 00000000..c84a314f
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/di/module/SmartUtilsModule.java
@@ -0,0 +1,4 @@
+package org.jsmart.zerocode.core.di.module;
+
+public class SmartUtilsModule {
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/di/DefaultGuiceHttpClientProvider.java b/src/main/java/org/jsmart/zerocode/core/di/provider/DefaultGuiceHttpClientProvider.java
similarity index 89%
rename from src/main/java/org/jsmart/zerocode/core/di/DefaultGuiceHttpClientProvider.java
rename to src/main/java/org/jsmart/zerocode/core/di/provider/DefaultGuiceHttpClientProvider.java
index 40e0f5fe..9424fdd3 100644
--- a/src/main/java/org/jsmart/zerocode/core/di/DefaultGuiceHttpClientProvider.java
+++ b/src/main/java/org/jsmart/zerocode/core/di/provider/DefaultGuiceHttpClientProvider.java
@@ -1,4 +1,4 @@
-package org.jsmart.zerocode.core.di;
+package org.jsmart.zerocode.core.di.provider;
 
 import org.jsmart.zerocode.core.httpclient.BasicHttpClient;
 import org.jsmart.zerocode.core.httpclient.ssl.SslTrustHttpClient;
diff --git a/src/main/java/org/jsmart/zerocode/core/di/provider/GsonSerDeProvider.java b/src/main/java/org/jsmart/zerocode/core/di/provider/GsonSerDeProvider.java
new file mode 100644
index 00000000..ff7546ef
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/di/provider/GsonSerDeProvider.java
@@ -0,0 +1,16 @@
+package org.jsmart.zerocode.core.di.provider;
+
+import com.google.gson.Gson;
+
+import javax.inject.Provider;
+
+public class GsonSerDeProvider implements Provider<Gson> {
+
+    @Override
+    public Gson get() {
+
+        return (new Gson());
+
+    }
+
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/di/provider/KafkaServicesProvider.java b/src/main/java/org/jsmart/zerocode/core/di/provider/KafkaServicesProvider.java
new file mode 100644
index 00000000..8f409135
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/di/provider/KafkaServicesProvider.java
@@ -0,0 +1,19 @@
+//package org.jsmart.zerocode.core.di.provider;
+//
+//import org.jsmart.zerocode.core.kafka.KafkaService;
+//import org.jsmart.zerocode.core.kafka.KafkaServiceImpl;
+//
+//import javax.inject.Provider;
+//
+//public class KafkaServicesProvider implements Provider<KafkaService> {
+//
+//    @Override
+//    public KafkaService get() {
+//
+//        KafkaService kafkaService = new KafkaServiceImpl();
+//        // Read the properties etc
+//
+//        return kafkaService;
+//    }
+//
+//}
\ No newline at end of file
diff --git a/src/main/java/org/jsmart/zerocode/core/di/ObjectMapperProvider.java b/src/main/java/org/jsmart/zerocode/core/di/provider/ObjectMapperProvider.java
similarity index 90%
rename from src/main/java/org/jsmart/zerocode/core/di/ObjectMapperProvider.java
rename to src/main/java/org/jsmart/zerocode/core/di/provider/ObjectMapperProvider.java
index f5a67eb2..6e146fb7 100644
--- a/src/main/java/org/jsmart/zerocode/core/di/ObjectMapperProvider.java
+++ b/src/main/java/org/jsmart/zerocode/core/di/provider/ObjectMapperProvider.java
@@ -1,4 +1,4 @@
-package org.jsmart.zerocode.core.di;
+package org.jsmart.zerocode.core.di.provider;
 
 import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.databind.ObjectMapper;
diff --git a/src/main/java/org/jsmart/zerocode/core/domain/MockStep.java b/src/main/java/org/jsmart/zerocode/core/domain/MockStep.java
index 988637ea..99e5b3c9 100644
--- a/src/main/java/org/jsmart/zerocode/core/domain/MockStep.java
+++ b/src/main/java/org/jsmart/zerocode/core/domain/MockStep.java
@@ -5,9 +5,8 @@ import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
-import org.jsmart.zerocode.core.di.ObjectMapperProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 
-import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 
diff --git a/src/main/java/org/jsmart/zerocode/core/domain/ReportProperties.java b/src/main/java/org/jsmart/zerocode/core/domain/ReportProperties.java
deleted file mode 100644
index 7b3d3bc9..00000000
--- a/src/main/java/org/jsmart/zerocode/core/domain/ReportProperties.java
+++ /dev/null
@@ -1,20 +0,0 @@
-//package org.jsmart.zerocode.core.domain;
-//
-//import java.lang.annotation.Documented;
-//import java.lang.annotation.ElementType;
-//import java.lang.annotation.Inherited;
-//import java.lang.annotation.Retention;
-//import java.lang.annotation.RetentionPolicy;
-//import java.lang.annotation.Target;
-//
-//@Documented
-//@Target(ElementType.TYPE)
-//@Retention(RetentionPolicy.RUNTIME)
-//@Inherited
-//public @interface ReportProperties {
-//
-//    boolean searchableReport() default true;
-//    boolean filterByAuthor() default true;
-//    boolean spikeReport() default false;
-//
-//}
\ No newline at end of file
diff --git a/src/main/java/org/jsmart/zerocode/core/domain/UseKafkaClient.java b/src/main/java/org/jsmart/zerocode/core/domain/UseKafkaClient.java
new file mode 100644
index 00000000..48ea45d5
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/domain/UseKafkaClient.java
@@ -0,0 +1,15 @@
+package org.jsmart.zerocode.core.domain;
+
+import org.jsmart.zerocode.core.kafka.client.BasicKafkaClient;
+
+import java.lang.annotation.*;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+@Inherited
+public @interface UseKafkaClient {
+    /**
+     * @return a Http Client implementation class which will override the default implementation of RestEasy client
+     */
+    Class<? extends BasicKafkaClient> value();
+}
\ No newline at end of file
diff --git a/src/main/java/org/jsmart/zerocode/core/domain/ZerocodeConstants.java b/src/main/java/org/jsmart/zerocode/core/domain/ZerocodeConstants.java
new file mode 100644
index 00000000..1e85f1ba
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/domain/ZerocodeConstants.java
@@ -0,0 +1,12 @@
+package org.jsmart.zerocode.core.domain;
+
+public interface ZerocodeConstants {
+    public static final String PROPERTY_KEY_HOST = "restful.application.endpoint.host";
+    public static final String PROPERTY_KEY_PORT = "restful.application.endpoint.context";
+
+    public static final String KAFKA = "kafka";
+    public static final String KAFKA_TOPIC = "kafka-topic:";
+    public static final String OK = "Ok";
+    public static final String FAILED = "Failed";
+
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/domain/builders/ZeroCodeReportBuilder.java b/src/main/java/org/jsmart/zerocode/core/domain/builders/ZeroCodeReportBuilder.java
index 508b7e4c..b10841ab 100644
--- a/src/main/java/org/jsmart/zerocode/core/domain/builders/ZeroCodeReportBuilder.java
+++ b/src/main/java/org/jsmart/zerocode/core/domain/builders/ZeroCodeReportBuilder.java
@@ -2,7 +2,7 @@ package org.jsmart.zerocode.core.domain.builders;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.jsmart.zerocode.core.di.ObjectMapperProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.jsmart.zerocode.core.domain.reports.ZeroCodeExecResult;
 import org.jsmart.zerocode.core.domain.reports.ZeroCodeReport;
 
diff --git a/src/main/java/org/jsmart/zerocode/core/engine/executor/JsonServiceExecutor.java b/src/main/java/org/jsmart/zerocode/core/engine/executor/JsonServiceExecutor.java
index 90473765..840fdb8d 100644
--- a/src/main/java/org/jsmart/zerocode/core/engine/executor/JsonServiceExecutor.java
+++ b/src/main/java/org/jsmart/zerocode/core/engine/executor/JsonServiceExecutor.java
@@ -3,7 +3,11 @@ package org.jsmart.zerocode.core.engine.executor;
 import com.fasterxml.jackson.core.JsonProcessingException;
 
 public interface JsonServiceExecutor {
-    public String executeJavaService(String serviceName, String methodName, String requestJson) throws JsonProcessingException;
+    String executeJavaService(String serviceName, String methodName, String requestJson) throws JsonProcessingException;
+
+    String executeRESTService(String urlName, String methodName, String requestJson);
+
+    String executeKafkaService(String kafkaServers, String urlName, String methodName, String requestJson);
+
 
-    public String executeRESTService(String urlName, String methodName, String requestJson);
 }
diff --git a/src/main/java/org/jsmart/zerocode/core/engine/executor/JsonServiceExecutorImpl.java b/src/main/java/org/jsmart/zerocode/core/engine/executor/JsonServiceExecutorImpl.java
index bfab94fc..a07066a5 100644
--- a/src/main/java/org/jsmart/zerocode/core/engine/executor/JsonServiceExecutorImpl.java
+++ b/src/main/java/org/jsmart/zerocode/core/engine/executor/JsonServiceExecutorImpl.java
@@ -10,6 +10,7 @@ import com.jayway.jsonpath.PathNotFoundException;
 import org.jsmart.zerocode.core.domain.MockSteps;
 import org.jsmart.zerocode.core.domain.Response;
 import org.jsmart.zerocode.core.httpclient.BasicHttpClient;
+import org.jsmart.zerocode.core.kafka.client.BasicKafkaClient;
 import org.jsmart.zerocode.core.utils.SmartUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -20,15 +21,12 @@ import java.util.HashMap;
 import java.util.List;
 
 import static org.apache.commons.lang.StringUtils.isEmpty;
-import static org.jsmart.zerocode.core.engine.mocker.RestEndPointMocker.createWithLocalMock;
-import static org.jsmart.zerocode.core.engine.mocker.RestEndPointMocker.createWithVirtuosoMock;
-import static org.jsmart.zerocode.core.engine.mocker.RestEndPointMocker.createWithWireMock;
+import static org.jsmart.zerocode.core.engine.mocker.RestEndPointMocker.*;
 import static org.jsmart.zerocode.core.utils.SmartUtils.prettyPrintJson;
 
 public class JsonServiceExecutorImpl implements JsonServiceExecutor {
     private static final Logger LOGGER = LoggerFactory.getLogger(JsonServiceExecutorImpl.class);
 
-    //guice
     @Inject
     private JavaExecutor javaExecutor;
 
@@ -36,23 +34,25 @@ public class JsonServiceExecutorImpl implements JsonServiceExecutor {
     private ObjectMapper objectMapper;
 
     @Inject
-    SmartUtils smartUtils;
+    private SmartUtils smartUtils;
 
     @Inject
-    BasicHttpClient httpClient;
+    private BasicHttpClient httpClient;
+
+    @Inject
+    private BasicKafkaClient kafkaClient;
 
     @Inject(optional = true)
     @Named("mock.api.port")
     private int mockPort;
 
-    //guice
-
     public JsonServiceExecutorImpl() {
     }
 
+    @Override
     public String executeJavaService(String serviceName, String methodName, String requestJson) throws JsonProcessingException {
 
-        if( javaExecutor == null) {
+        if (javaExecutor == null) {
             throw new RuntimeException("Can not proceed as the framework could not load the executors. ");
         }
 
@@ -75,10 +75,11 @@ public class JsonServiceExecutorImpl implements JsonServiceExecutor {
         }
     }
 
+    @Override
     public String executeRESTService(String urlName, String methodName, String requestJson) {
 
         try {
-            String responseJson =  executeRESTInternal(urlName, methodName, requestJson);
+            String responseJson = executeRESTInternal(urlName, methodName, requestJson);
 
             return responseJson;
 
@@ -96,6 +97,11 @@ public class JsonServiceExecutorImpl implements JsonServiceExecutor {
         }
     }
 
+    @Override
+    public String executeKafkaService(String kafkaServers, String topicName, String operation, String requestJson) {
+        return kafkaClient.execute(kafkaServers, topicName, operation, requestJson);
+    }
+
     private String executeRESTInternal(String httpUrl, String methodName, String requestJson) throws Exception {
 
         HashMap queryParams = (HashMap) readJsonPathOrElseNull(requestJson, "$.queryParams");
@@ -124,46 +130,46 @@ public class JsonServiceExecutorImpl implements JsonServiceExecutor {
 
         final MultivaluedMap responseHeaders = serverResponse.getMetadata();
 
-        final String responseBodyAsString = (String)serverResponse.getEntity();
-    
+        final String responseBodyAsString = (String) serverResponse.getEntity();
+
         Response zeroCodeResponse = deriveZeroCodeResponseFrom(responseStatus, responseHeaders, responseBodyAsString);
-        
+
         final String zeroCodeResponseString = objectMapper.writeValueAsString(zeroCodeResponse);
 
         return prettyPrintJson(zeroCodeResponseString);
     }
-    
+
     private Response deriveZeroCodeResponseFrom(int responseStatus,
-                    MultivaluedMap responseHeaders,
-                    String responseBodyAsString)
-                    throws IOException {
-        
+                                                MultivaluedMap responseHeaders,
+                                                String responseBodyAsString)
+            throws IOException {
+
         final JsonNode jsonBody;
         final String rawBody;
-        
-        if(isEmpty(responseBodyAsString)){
+
+        if (isEmpty(responseBodyAsString)) {
             jsonBody = null;
             rawBody = null;
-    
-        } else if(isParsableJson(responseBodyAsString)){
+
+        } else if (isParsableJson(responseBodyAsString)) {
             jsonBody = objectMapper.readValue(responseBodyAsString, JsonNode.class);
             rawBody = null;
-            
+
         } else {
             jsonBody = null;
             rawBody = responseBodyAsString;
-        
+
         }
-        
+
         return new Response(responseStatus, responseHeaders, jsonBody, rawBody, null);
     }
-    
+
     private boolean completedMockingEndPoints(String httpUrl, String requestJson, String methodName, Object bodyContent) throws java.io.IOException {
-        if(httpUrl.contains("/$MOCK") && methodName.equals("$USE.WIREMOCK")){
+        if (httpUrl.contains("/$MOCK") && methodName.equals("$USE.WIREMOCK")) {
 
             MockSteps mockSteps = smartUtils.getMapper().readValue(requestJson, MockSteps.class);
 
-            if(mockPort > 0){
+            if (mockPort > 0) {
                 createWithWireMock(mockSteps, mockPort);
 
                 LOGGER.info("#SUCCESS: End points simulated via wiremock.");
@@ -175,9 +181,7 @@ public class JsonServiceExecutorImpl implements JsonServiceExecutor {
                     "Usage: e.g. in your <env host config .properties> file provide- \n " +
                     "mock.api.port=8888\n\n");
             return false;
-        }
-
-        else if(httpUrl.contains("/$MOCK") && methodName.equals("$USE.VIRTUOSO")){
+        } else if (httpUrl.contains("/$MOCK") && methodName.equals("$USE.VIRTUOSO")) {
             LOGGER.info("\n#body:\n" + bodyContent);
 
             //read the content of the "request". This contains the complete rest API.
@@ -185,9 +189,7 @@ public class JsonServiceExecutorImpl implements JsonServiceExecutor {
 
             LOGGER.info("#SUCCESS: End point simulated via virtuoso.");
             return true;
-        }
-
-        else if(httpUrl.contains("/$MOCK") && methodName.equals("$USE.SIMULATOR")){
+        } else if (httpUrl.contains("/$MOCK") && methodName.equals("$USE.SIMULATOR")) {
             LOGGER.info("\n#body:\n" + bodyContent);
 
             //read the content of the "request". This contains the complete rest API.
@@ -201,25 +203,25 @@ public class JsonServiceExecutorImpl implements JsonServiceExecutor {
     }
 
     private Object readJsonPathOrElseNull(String requestJson, String jsonPath) {
-        try{
+        try {
             return JsonPath.read(requestJson, jsonPath);
-        } catch(PathNotFoundException pEx){
+        } catch (PathNotFoundException pEx) {
             LOGGER.debug("No " + jsonPath + " was present in the request. returned null.");
-            return  null;
+            return null;
         }
     }
-    
+
     private boolean isParsableJson(String potentialJsonString) {
         try {
             objectMapper.readTree(potentialJsonString);
             return true;
         } catch (IOException e) {
             LOGGER.warn("\n---------------------------------------------\n\n"
-                         + "\t\t\t\t\t\t * Warning *  \n\nOutput was not a valid JSON body. It was treated as a simple rawBody."
-                        + " If it was intentional, you can ignore this warning. "
-                         + "\n -OR- Update your assertions block with \"rawBody\" instead of \"body\" "
-                         + "\n e.g. \"rawBody\" : \"an expected string \""
-                         + "\n\n---------------------------------------------");
+                    + "\t\t\t\t\t\t * Warning *  \n\nOutput was not a valid JSON body. It was treated as a simple rawBody."
+                    + " If it was intentional, you can ignore this warning. "
+                    + "\n -OR- Update your assertions block with \"rawBody\" instead of \"body\" "
+                    + "\n e.g. \"rawBody\" : \"an expected string \""
+                    + "\n\n---------------------------------------------");
             return false;
         }
     }
diff --git a/src/main/java/org/jsmart/zerocode/core/httpclient/utils/FileUploadUtils.java b/src/main/java/org/jsmart/zerocode/core/httpclient/utils/FileUploadUtils.java
index 78d431fa..8774de19 100644
--- a/src/main/java/org/jsmart/zerocode/core/httpclient/utils/FileUploadUtils.java
+++ b/src/main/java/org/jsmart/zerocode/core/httpclient/utils/FileUploadUtils.java
@@ -5,7 +5,7 @@ import org.apache.http.client.methods.RequestBuilder;
 import org.apache.http.entity.mime.MultipartEntityBuilder;
 import org.apache.http.entity.mime.content.FileBody;
 import org.apache.http.entity.mime.content.StringBody;
-import org.jsmart.zerocode.core.di.ObjectMapperProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 
 import java.io.File;
 import java.io.IOException;
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/ConsumedRecords.java b/src/main/java/org/jsmart/zerocode/core/kafka/ConsumedRecords.java
new file mode 100644
index 00000000..ade36b60
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/ConsumedRecords.java
@@ -0,0 +1,31 @@
+package org.jsmart.zerocode.core.kafka;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+import java.util.List;
+
+public class ConsumedRecords {
+    private List<ConsumerRecord> records;
+    private Integer size;
+
+    public ConsumedRecords(List<ConsumerRecord> records) {
+        this.records = records;
+        this.size = records != null? records.size() : 0;
+    }
+
+    public List<ConsumerRecord> getRecords() {
+        return records;
+    }
+
+    public Integer getSize() {
+        return records != null? records.size() : 0;
+    }
+
+    @Override
+    public String toString() {
+        return "ConsumedRecords{" +
+                "records=" + records +
+                ", size=" + size +
+                '}';
+    }
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/ConsumerCreator.java b/src/main/java/org/jsmart/zerocode/core/kafka/ConsumerCreator.java
new file mode 100644
index 00000000..b2a2cc41
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/ConsumerCreator.java
@@ -0,0 +1,29 @@
+package org.jsmart.zerocode.core.kafka;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+
+import java.util.Collections;
+import java.util.Properties;
+
+public class ConsumerCreator {
+
+	public static Consumer<Long, String> createConsumer() {
+		final Properties props = new Properties();
+		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstants.KAFKA_BROKERS);
+		props.put(ConsumerConfig.GROUP_ID_CONFIG, KafkaConstants.GROUP_ID_CONFIG);
+		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
+		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+		props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, KafkaConstants.MAX_POLL_RECORDS);
+		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, KafkaConstants.OFFSET_RESET_EARLIER);
+
+		final Consumer<Long, String> consumer = new KafkaConsumer<>(props);
+		consumer.subscribe(Collections.singletonList(KafkaConstants.TOPIC_NAME));
+		return consumer;
+	}
+
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/DeliveryStatus.java b/src/main/java/org/jsmart/zerocode/core/kafka/DeliveryStatus.java
new file mode 100644
index 00000000..25b81f10
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/DeliveryStatus.java
@@ -0,0 +1,72 @@
+package org.jsmart.zerocode.core.kafka;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+import java.util.Objects;
+
+@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
+public class DeliveryStatus {
+    private final String status;
+    private final String message;
+    private final Integer recordCount;
+
+    @JsonCreator
+    public DeliveryStatus(
+            @JsonProperty("status") String status,
+            @JsonProperty("message") String message,
+            @JsonProperty("recordCount") Integer recordCount) {
+        this.status = status;
+        this.message = message;
+        this.recordCount = recordCount;
+    }
+
+    public DeliveryStatus(String status, String message) {
+        this(status, message, null);
+    }
+    public DeliveryStatus(String status, Integer recordCount) {
+        this(status, null, recordCount);
+    }
+
+    public DeliveryStatus(String status) {
+        this(status, null, null);
+    }
+
+    public String getStatus() {
+        return status;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public Integer getRecordCount() {
+        return recordCount;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        DeliveryStatus that = (DeliveryStatus) o;
+        return Objects.equals(status, that.status) &&
+                Objects.equals(message, that.message) &&
+                Objects.equals(recordCount, that.recordCount);
+    }
+
+    @Override
+    public int hashCode() {
+
+        return Objects.hash(status, message, recordCount);
+    }
+
+    @Override
+    public String toString() {
+        return "DeliveryStatus{" +
+                "status='" + status + '\'' +
+                ", message='" + message + '\'' +
+                ", recordCount='" + recordCount + '\'' +
+                '}';
+    }
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/KafkaConstants.java b/src/main/java/org/jsmart/zerocode/core/kafka/KafkaConstants.java
new file mode 100644
index 00000000..e64f2961
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/KafkaConstants.java
@@ -0,0 +1,25 @@
+package org.jsmart.zerocode.core.kafka;
+
+public interface KafkaConstants {
+	public static String KAFKA_BROKERS = "192.168.99.1:9092";
+	
+	public static Integer MESSAGE_COUNT=110;
+	
+	public static String CLIENT_ID="client1";
+	
+	public static String TOPIC_NAME="demo";
+	
+	public static String GROUP_ID_CONFIG="consumerGroup10";
+	
+	public static Integer MAX_NO_MESSAGE_FOUND_COUNT=5;
+
+	public static Integer MAX_NO_OF_RETRY_POLLS_OR_TIME_OUTS =5;
+
+	public static Long DEFAULT_POLLING_TIME_MILLI_SEC = 100L;
+
+	public static String OFFSET_RESET_LATEST="latest";
+	
+	public static String OFFSET_RESET_EARLIER="earliest";
+	
+	public static Integer MAX_POLL_RECORDS=1;
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/KafkaService.java b/src/main/java/org/jsmart/zerocode/core/kafka/KafkaService.java
new file mode 100755
index 00000000..09adb322
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/KafkaService.java
@@ -0,0 +1,10 @@
+package org.jsmart.zerocode.core.kafka;
+
+public interface KafkaService {
+
+	String consume(String serviceName, String methodName, String requestJson);
+
+    String produce(String kafkaServers, String topicName, String requestJson);
+
+    String execute(String kafkaServers, String topicName, String operation, String requestJson);
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/KafkaServiceImpl.java b/src/main/java/org/jsmart/zerocode/core/kafka/KafkaServiceImpl.java
new file mode 100644
index 00000000..7b50bccc
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/KafkaServiceImpl.java
@@ -0,0 +1,37 @@
+//package org.jsmart.zerocode.core.kafka;
+//
+//import com.google.inject.Inject;
+//import com.google.inject.name.Named;
+//import org.jsmart.zerocode.core.kafka.client.BasicKafkaClient;
+//
+//public class KafkaServiceImpl implements KafkaService {
+//    @Inject(optional = true)
+//    @Named("consumer.commitAsync")
+//    private boolean commitAsync;
+//
+//    @Inject(optional = true)
+//    @Named("producer.key1")
+//    private String key1;
+//
+//    @Inject
+//    private BasicKafkaClient kafkaClient;
+//
+//    @Override
+//    public String consume(String serviceName, String methodName, String requestJson) {
+//        System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>> commitAsync = " + commitAsync);
+//        return null;
+//    }
+//
+//    @Override
+//    public String produce(String kafkaServers, String topicName, String requestJson) {
+//        System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>> key1 = " + key1);
+//        return "{\"status\": \"Ok\"}";
+//    }
+//
+//    @Override
+//    public String execute(String brokers, String topicName, String operation, String requestJson) {
+//
+//        return kafkaClient.execute(brokers, topicName, operation, requestJson);
+//    }
+//}
+//
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/client/BasicKafkaClient.java b/src/main/java/org/jsmart/zerocode/core/kafka/client/BasicKafkaClient.java
new file mode 100644
index 00000000..c15c9960
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/client/BasicKafkaClient.java
@@ -0,0 +1,54 @@
+package org.jsmart.zerocode.core.kafka.client;
+
+import com.google.inject.Inject;
+import org.jsmart.zerocode.core.kafka.receive.KafkaReceiver;
+import org.jsmart.zerocode.core.kafka.send.KafkaSender;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class BasicKafkaClient {
+    private Logger LOGGER = LoggerFactory.getLogger(BasicKafkaClient.class);
+
+    @Inject
+    private  KafkaSender sender;
+
+    @Inject
+    private  KafkaReceiver receiver;
+
+
+    public BasicKafkaClient() {
+    }
+
+    public String execute(String brokers, String topicName, String operation, String requestJson) {
+        LOGGER.info("brokers:{}, topicName:{}, operation:{}, requestJson:{}", brokers, topicName, operation, requestJson);
+
+        try {
+            switch (operation) {
+                case "send":
+                case "load":
+                case "publish":
+                case "produce":
+                    return sender.send(brokers, topicName, requestJson);
+
+                case "unload":
+                case "consume":
+                case "receive":
+                case "subscribe":
+                    return receiver.receive(brokers, topicName, requestJson);
+
+                case "poll":
+                    throw new RuntimeException("poll - Not yet Implemented");
+
+                default:
+                    throw new RuntimeException("Unsupported. Framework could not assume a default Kafka operation");
+            }
+
+        } catch (Throwable exx) {
+
+            LOGGER.error("Exception during operation:{}, topicName:{}, error:{}", operation, topicName, exx);
+
+            throw new RuntimeException(exx);
+        }
+
+    }
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/client/ZerocodeCustomKafkaClient.java b/src/main/java/org/jsmart/zerocode/core/kafka/client/ZerocodeCustomKafkaClient.java
new file mode 100644
index 00000000..9fb21344
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/client/ZerocodeCustomKafkaClient.java
@@ -0,0 +1,14 @@
+package org.jsmart.zerocode.core.kafka.client;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ZerocodeCustomKafkaClient extends BasicKafkaClient{
+    private static final Logger LOGGER = LoggerFactory.getLogger(ZerocodeCustomKafkaClient.class);
+
+    public ZerocodeCustomKafkaClient() {
+        super();
+    }
+
+}
+
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/common/CommonConfigs.java b/src/main/java/org/jsmart/zerocode/core/kafka/common/CommonConfigs.java
new file mode 100644
index 00000000..142d0da2
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/common/CommonConfigs.java
@@ -0,0 +1,6 @@
+package org.jsmart.zerocode.core.kafka.common;
+
+public class CommonConfigs {
+    public static final String BOOTSTRAP_SERVERS = "bootstrap.servers";
+
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigs.java b/src/main/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigs.java
new file mode 100644
index 00000000..53b6fb7c
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigs.java
@@ -0,0 +1,99 @@
+package org.jsmart.zerocode.core.kafka.consume;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+import java.util.Objects;
+
+//@JsonIgnoreProperties(ignoreUnknown = true) //<--- Do not enable this. All properties to be aware of and processed
+@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
+public class ConsumerLocalConfigs {
+    private final String fileDumpTo;
+    private final String fileDumpType;
+    private final Boolean commitAsync;
+    private final Boolean commitSync;
+    private final Boolean showRecordsInResponse;
+    private final Integer maxNoOfRetryPollsOrTimeouts;
+    private final Long pollingTime;
+
+    @JsonCreator
+    public ConsumerLocalConfigs(
+            @JsonProperty("fileDumpTo") String fileDumpTo,
+            @JsonProperty("fileDumpType") String fileDumpType,
+            @JsonProperty("commitAsync") Boolean commitAsync,
+            @JsonProperty("commitSync") Boolean commitSync,
+            @JsonProperty("showRecordsInResponse") Boolean showRecordsInResponse,
+            @JsonProperty("maxNoOfRetryPollsOrTimeouts") Integer maxNoOfRetryPollsOrTimeouts,
+            @JsonProperty("pollingTime")Long pollingTime) {
+        this.fileDumpTo = fileDumpTo;
+        this.fileDumpType = fileDumpType;
+        this.commitAsync = commitAsync;
+        this.commitSync = commitSync;
+        this.showRecordsInResponse = showRecordsInResponse;
+        this.maxNoOfRetryPollsOrTimeouts = maxNoOfRetryPollsOrTimeouts;
+        this.pollingTime = pollingTime;
+    }
+
+    public String getFileDumpTo() {
+        return fileDumpTo;
+    }
+
+    public String getFileDumpType() {
+        return fileDumpType;
+    }
+
+    public Boolean getCommitAsync() {
+        return commitAsync;
+    }
+
+    public Boolean getCommitSync() {
+        return commitSync;
+    }
+
+    public Boolean getShowRecordsInResponse() {
+        return showRecordsInResponse;
+    }
+
+    public Integer getMaxNoOfRetryPollsOrTimeouts() {
+        return maxNoOfRetryPollsOrTimeouts;
+    }
+
+    public Long getPollingTime() {
+        return pollingTime;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ConsumerLocalConfigs that = (ConsumerLocalConfigs) o;
+        return Objects.equals(fileDumpTo, that.fileDumpTo) &&
+                Objects.equals(fileDumpType, that.fileDumpType) &&
+                Objects.equals(commitAsync, that.commitAsync) &&
+                Objects.equals(commitSync, that.commitSync) &&
+                Objects.equals(showRecordsInResponse, that.showRecordsInResponse) &&
+                Objects.equals(maxNoOfRetryPollsOrTimeouts, that.maxNoOfRetryPollsOrTimeouts) &&
+                Objects.equals(pollingTime, that.pollingTime);
+    }
+
+    @Override
+    public int hashCode() {
+
+        return Objects.hash(fileDumpTo, fileDumpType, commitAsync, commitSync,
+                showRecordsInResponse, maxNoOfRetryPollsOrTimeouts, pollingTime);
+    }
+
+    @Override
+    public String toString() {
+        return "ConsumerLocalConfigs{" +
+                "fileDumpTo='" + fileDumpTo + '\'' +
+                ", fileDumpType='" + fileDumpType + '\'' +
+                ", commitAsync=" + commitAsync +
+                ", commitSync=" + commitSync +
+                ", showRecordsInResponse=" + showRecordsInResponse +
+                ", maxNoOfRetryPollsOrTimeouts=" + maxNoOfRetryPollsOrTimeouts +
+                ", pollingTime=" + pollingTime +
+                '}';
+    }
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrap.java b/src/main/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrap.java
new file mode 100644
index 00000000..d945cae3
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrap.java
@@ -0,0 +1,43 @@
+package org.jsmart.zerocode.core.kafka.consume;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+
+import java.util.Objects;
+
+@JsonSerialize(include = JsonSerialize.Inclusion.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class ConsumerLocalConfigsWrap {
+    private final ConsumerLocalConfigs consumerLocalConfigs;
+
+    @JsonCreator
+    public ConsumerLocalConfigsWrap(@JsonProperty("consumerLocalConfigs") ConsumerLocalConfigs consumerLocalConfigs) {
+        this.consumerLocalConfigs = consumerLocalConfigs;
+    }
+
+    public ConsumerLocalConfigs getConsumerLocalConfigs() {
+        return consumerLocalConfigs;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ConsumerLocalConfigsWrap that = (ConsumerLocalConfigsWrap) o;
+        return Objects.equals(consumerLocalConfigs, that.consumerLocalConfigs);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(consumerLocalConfigs);
+    }
+
+    @Override
+    public String toString() {
+        return "ConsumerLocalConfigsWrap{" +
+                "consumerLocalConfigs=" + consumerLocalConfigs +
+                '}';
+    }
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaFileRecordHelper.java b/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaFileRecordHelper.java
new file mode 100644
index 00000000..ad06394b
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaFileRecordHelper.java
@@ -0,0 +1,77 @@
+package org.jsmart.zerocode.core.kafka.helper;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigs;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.List;
+
+import static org.jsmart.zerocode.core.kafka.helper.KafkaHelper.validateConsumeProperties;
+
+public class KafkaFileRecordHelper {
+    public static void handleRecordsDump(ConsumerLocalConfigs consumeLocalTestProps, List<ConsumerRecord> fetchedRecords) {
+        String fileDumpType = consumeLocalTestProps != null ? consumeLocalTestProps.getFileDumpType() : null;
+
+        if (fileDumpType != null) {
+
+            validateConsumeProperties(consumeLocalTestProps);
+
+            switch (fileDumpType) {
+                case "RAW":
+                    dumpRecords(consumeLocalTestProps.getFileDumpTo(), fetchedRecords);
+                    break;
+
+                case "BIN":
+                    //TODO - Handle image data etc.
+                    break;
+
+                case "JSON":
+                    // TODO - dump JSON data to a file
+                    break;
+
+                default:
+                    throw new RuntimeException("Unsupported fileDumpType - '" + fileDumpType + "'");
+            }
+        }
+    }
+
+    public static void dumpRecords(String fileName, List<ConsumerRecord> fetchedRecords) {
+
+        File file = createCascadeIfNotExisting(fileName);
+
+        try {
+            FileWriter writer = new FileWriter(file.getAbsoluteFile());
+
+            for (ConsumerRecord aRecord : fetchedRecords) {
+                String key = aRecord.key() != null ? aRecord.key().toString() : "";
+                String value = aRecord.value() != null ? aRecord.value().toString() : "";
+                writer.write(key + value + osIndependentNewLine());
+            }
+            writer.close();
+        } catch (IOException exx) {
+            throw new RuntimeException("Could not write to file '" + fileName + "' exception >> " + exx);
+        }
+    }
+
+    private static File createCascadeIfNotExisting(String fileName) {
+        try {
+            Path path = Paths.get(fileName);
+            Files.createDirectories(path.getParent());
+
+            File file = new File(fileName);
+
+            return file;
+        } catch (IOException exx) {
+            throw new RuntimeException("Create file '" + fileName + "' Exception" + exx);
+        }
+    }
+
+    private static String osIndependentNewLine() {
+        return System.getProperty("line.separator");
+    }
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaHelper.java b/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaHelper.java
new file mode 100644
index 00000000..093111b1
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaHelper.java
@@ -0,0 +1,163 @@
+package org.jsmart.zerocode.core.kafka.helper;
+
+import com.google.common.io.Resources;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
+import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigs;
+import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigsWrap;
+import org.jsmart.zerocode.core.kafka.receive.ConsumerCommonConfigs;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.Properties;
+
+import static java.util.Optional.ofNullable;
+import static org.jsmart.zerocode.core.kafka.KafkaConstants.DEFAULT_POLLING_TIME_MILLI_SEC;
+import static org.jsmart.zerocode.core.kafka.KafkaConstants.MAX_NO_OF_RETRY_POLLS_OR_TIME_OUTS;
+import static org.jsmart.zerocode.core.kafka.common.CommonConfigs.BOOTSTRAP_SERVERS;
+
+public class KafkaHelper {
+    public static Producer<Long, String> createProducer(String bootStrapServers, String producerPropertyFile) {
+
+        try (InputStream propsIs = Resources.getResource(producerPropertyFile).openStream()) {
+            Properties properties = new Properties();
+            properties.load(propsIs);
+            properties.put(BOOTSTRAP_SERVERS, bootStrapServers);
+
+            return new KafkaProducer(properties);
+
+        } catch (IOException e) {
+            throw new RuntimeException("Exception while reading kafka producer properties" + e);
+        }
+    }
+
+    public static Consumer<Long, String> createConsumer(String bootStrapServers, String consumerPropertyFile, String topic) {
+        try (InputStream propsIs = Resources.getResource(consumerPropertyFile).openStream()) {
+            Properties properties = new Properties();
+            properties.load(propsIs);
+            properties.put("bootstrap.servers", bootStrapServers);
+
+            final Consumer<Long, String> consumer = new KafkaConsumer<>(properties);
+            consumer.subscribe(Collections.singletonList(topic));
+            return consumer;
+
+        } catch (IOException e) {
+            throw new RuntimeException("Exception while reading kafka properties" + e);
+        }
+    }
+
+    public static void validateConsumeProperties(ConsumerLocalConfigs consumeLocalTestProps) {
+        if (null != consumeLocalTestProps.getFileDumpType() && consumeLocalTestProps.getFileDumpTo() == null) {
+            throw new RuntimeException("Found type, but no fileName. Try e.g. 'fileDumpTo':'target/temp/abc.txt' ");
+        }
+    }
+
+    public static void validateLocalConfigs(ConsumerLocalConfigs consumeLocalTestProps) {
+        if (consumeLocalTestProps != null) {
+            Boolean localCommitSync = consumeLocalTestProps.getCommitSync();
+            Boolean localCommitAsync = consumeLocalTestProps.getCommitAsync();
+
+            validateIfBothEnabled(localCommitSync, localCommitAsync);
+        }
+    }
+
+    public static void validateCommonConfigs(ConsumerCommonConfigs consumerCommonConfigs) {
+        validateIfBothEnabled(consumerCommonConfigs.getCommitSync(), consumerCommonConfigs.getCommitAsync());
+    }
+
+    private static void validateIfBothEnabled(Boolean commitSync, Boolean commitAsync) {
+        if ((commitSync != null && commitAsync != null)  && commitSync == true && commitAsync == true) {
+            throw new RuntimeException("\n********* Both commitSync and commitAsync can not be true *********\n");
+        }
+    }
+
+    public static ConsumerLocalConfigs deriveEffectiveConfigs(ConsumerLocalConfigs consumerLocalTestConfigs, ConsumerCommonConfigs consumerCommonConfigs) {
+
+        validateCommonConfigs(consumerCommonConfigs);
+        validateLocalConfigs(consumerLocalTestConfigs);
+
+        return createEffective(consumerCommonConfigs, consumerLocalTestConfigs);
+    }
+
+
+    public static ConsumerLocalConfigs createEffective(ConsumerCommonConfigs consumerCommon, ConsumerLocalConfigs consumerLocal) {
+        if(consumerLocal == null){
+            return new ConsumerLocalConfigs(
+                    consumerCommon.getFileDumpTo(),
+                    consumerCommon.getFileDumpType(),
+                    consumerCommon.getCommitAsync(),
+                    consumerCommon.getCommitSync(),
+                    consumerCommon.getShowRecordsInResponse(),
+                    consumerCommon.getMaxNoOfRetryPollsOrTimeouts(),
+                    consumerCommon.getPollingTime());
+        }
+
+        // Handle fileDumpTo
+        String effectiveFileDumpTo = ofNullable(consumerLocal.getFileDumpTo()).orElse(consumerCommon.getFileDumpTo());
+
+        // Handle fileDumpType
+        String effectiveFileDumpType = ofNullable(consumerLocal.getFileDumpType()).orElse(consumerCommon.getFileDumpType());
+
+        // Handle showRecordsInResponse
+        Boolean effectiveShowRecordsInResponse = ofNullable(consumerLocal.getShowRecordsInResponse()).orElse(consumerCommon.getShowRecordsInResponse());
+
+        // Handle maxNoOfRetryPollsOrTimeouts
+        Integer effectiveMaxNoOfRetryPollsOrTimeouts = ofNullable(consumerLocal.getMaxNoOfRetryPollsOrTimeouts()).orElse(consumerCommon.getMaxNoOfRetryPollsOrTimeouts());
+
+        // Handle pollingTime
+        Long effectivePollingTime = ofNullable(consumerLocal.getPollingTime()).orElse(consumerCommon.getPollingTime());
+
+        // Handle commitSync and commitAsync -START
+        Boolean effectiveCommitSync;
+        Boolean effectiveCommitAsync;
+
+        Boolean localCommitSync = consumerLocal.getCommitSync();
+        Boolean localCommitAsync = consumerLocal.getCommitAsync();
+
+        if (localCommitSync == null && localCommitAsync == null) {
+            effectiveCommitSync = consumerCommon.getCommitSync();
+            effectiveCommitAsync = consumerCommon.getCommitAsync();
+
+        } else {
+            effectiveCommitSync = localCommitSync;
+            effectiveCommitAsync = localCommitAsync;
+        }
+
+        return new ConsumerLocalConfigs(
+                effectiveFileDumpTo,
+                effectiveFileDumpType,
+                effectiveCommitAsync,
+                effectiveCommitSync,
+                effectiveShowRecordsInResponse,
+                effectiveMaxNoOfRetryPollsOrTimeouts,
+                effectivePollingTime);
+    }
+
+    public static ConsumerLocalConfigs readConsumerLocalTestProperties(String requestJsonWithConfigWrapped) {
+        try {
+            ConsumerLocalConfigsWrap consumerLocalConfigsWrap = (new ObjectMapperProvider().get())
+                    .readValue(requestJsonWithConfigWrapped, ConsumerLocalConfigsWrap.class);
+
+            return consumerLocalConfigsWrap.getConsumerLocalConfigs();
+
+        } catch (IOException exx) {
+            throw new RuntimeException(exx);
+        }
+    }
+
+    public static Integer getMaxTimeOuts(ConsumerLocalConfigs effectiveLocalTestProps) {
+        return ofNullable(effectiveLocalTestProps.getMaxNoOfRetryPollsOrTimeouts())
+                .orElse(MAX_NO_OF_RETRY_POLLS_OR_TIME_OUTS);
+    }
+
+
+    public static Long getPollTime(ConsumerLocalConfigs effectiveLocal) {
+        return ofNullable(effectiveLocal.getPollingTime())
+                .orElse(DEFAULT_POLLING_TIME_MILLI_SEC);
+    }
+
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/receive/ConsumerCommonConfigs.java b/src/main/java/org/jsmart/zerocode/core/kafka/receive/ConsumerCommonConfigs.java
new file mode 100644
index 00000000..64068654
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/receive/ConsumerCommonConfigs.java
@@ -0,0 +1,99 @@
+package org.jsmart.zerocode.core.kafka.receive;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import com.google.inject.name.Named;
+
+@Singleton
+public class ConsumerCommonConfigs {
+
+    @Inject(optional = true)
+    @Named("consumer.commitSync")
+    private Boolean commitSync;
+
+    @Inject(optional = true)
+    @Named("consumer.commitAsync")
+    private Boolean commitAsync;
+
+    @Inject(optional = true)
+    @Named("consumer.fileDumpTo")
+    private String fileDumpTo;
+
+    @Inject(optional = true)
+    @Named("consumer.fileDumpType")
+    private String fileDumpType;
+
+    @Inject(optional = true)
+    @Named("consumer.showRecordsInResponse")
+    private Boolean showRecordsInResponse;
+
+    @Inject(optional = true)
+    @Named("consumer.maxNoOfRetryPollsOrTimeouts")
+    private Integer maxNoOfRetryPollsOrTimeouts;
+
+    @Inject(optional = true)
+    @Named("consumer.pollingTime")
+    private Long pollingTime;
+
+    public ConsumerCommonConfigs() {
+    }
+
+    public ConsumerCommonConfigs(Boolean commitSync,
+                                 Boolean commitAsync,
+                                 String fileDumpTo,
+                                 String fileDumpType,
+                                 Boolean showRecordsInResponse,
+                                 Integer maxNoOfRetryPollsOrTimeouts,
+                                 Long pollingTime
+
+    ) {
+        this.commitSync = commitSync;
+        this.commitAsync = commitAsync;
+        this.fileDumpTo = fileDumpTo;
+        this.fileDumpType = fileDumpType;
+        this.showRecordsInResponse = showRecordsInResponse;
+        this.maxNoOfRetryPollsOrTimeouts = maxNoOfRetryPollsOrTimeouts;
+        this.pollingTime = pollingTime;
+    }
+
+    public Boolean getCommitSync() {
+        return commitSync;
+    }
+
+    public Boolean getCommitAsync() {
+        return commitAsync;
+    }
+
+    public String getFileDumpTo() {
+        return fileDumpTo;
+    }
+
+    public String getFileDumpType() {
+        return fileDumpType;
+    }
+
+    public Boolean getShowRecordsInResponse() {
+        return showRecordsInResponse;
+    }
+
+    public Integer getMaxNoOfRetryPollsOrTimeouts() {
+        return maxNoOfRetryPollsOrTimeouts;
+    }
+
+    public Long getPollingTime() {
+        return pollingTime;
+    }
+
+    @Override
+    public String toString() {
+        return "ConsumerCommonConfigs{" +
+                "commitSync=" + commitSync +
+                ", commitAsync=" + commitAsync +
+                ", fileDumpTo='" + fileDumpTo + '\'' +
+                ", fileDumpType='" + fileDumpType + '\'' +
+                ", showRecordsInResponse=" + showRecordsInResponse +
+                ", maxNoOfRetryPollsOrTimeouts=" + maxNoOfRetryPollsOrTimeouts +
+                ", pollingTime=" + pollingTime +
+                '}';
+    }
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiver.java b/src/main/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiver.java
new file mode 100644
index 00000000..9a045bfb
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiver.java
@@ -0,0 +1,151 @@
+package org.jsmart.zerocode.core.kafka.receive;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.gson.Gson;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import com.google.inject.name.Named;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.jsmart.zerocode.core.di.provider.GsonSerDeProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
+import org.jsmart.zerocode.core.kafka.ConsumedRecords;
+import org.jsmart.zerocode.core.kafka.DeliveryStatus;
+import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static java.time.Duration.ofMillis;
+import static org.jsmart.zerocode.core.domain.ZerocodeConstants.OK;
+import static org.jsmart.zerocode.core.kafka.helper.KafkaFileRecordHelper.handleRecordsDump;
+import static org.jsmart.zerocode.core.kafka.helper.KafkaHelper.*;
+import static org.jsmart.zerocode.core.utils.SmartUtils.prettyPrintJson;
+
+@Singleton
+public class KafkaReceiver {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaReceiver.class);
+
+    private static Gson gson = new GsonSerDeProvider().get();
+
+    private final ObjectMapper objectMapper = new ObjectMapperProvider().get();
+
+    @Inject(optional = true)
+    @Named("kafka.consumer.properties")
+    private String consumerPropertyFile;
+
+    @Inject
+    private ConsumerCommonConfigs consumerCommonConfigs;
+
+    public String receive(String kafkaServers, String topicName, String requestJsonWithConfig) throws IOException {
+
+        ConsumerLocalConfigs consumerLocalConfigs = readConsumerLocalTestProperties(requestJsonWithConfig);
+
+        ConsumerLocalConfigs effectiveLocal = deriveEffectiveConfigs(consumerLocalConfigs, consumerCommonConfigs);
+
+        LOGGER.info("\n### Consumer Effective configs:{}\n", effectiveLocal);
+
+        Consumer<Long, String> consumer = createConsumer(kafkaServers, consumerPropertyFile, topicName);
+
+        final List<ConsumerRecord> fetchedRecords = new ArrayList<>();
+
+        int noOfTimeOuts = 0;
+
+        while (true) {
+            LOGGER.info("polling records  - noOfTimeOuts reached : " + noOfTimeOuts);
+
+            final ConsumerRecords<Long, String> records = consumer.poll(ofMillis(getPollTime(effectiveLocal)));
+
+            //String jsonRecords = gson.toJson(records);
+            //System.out.println("jsonRecords>>>>>>>>>>\n" + jsonRecords);
+
+            if (records.count() == 0) {
+                noOfTimeOuts++;
+                if (noOfTimeOuts > getMaxTimeOuts(effectiveLocal)) {
+                    break;
+                } else {
+                    continue;
+                }
+            } else {
+                LOGGER.info("Got {} records after {} timeouts\n", records.count(), noOfTimeOuts);
+                // -----------------------------------
+                // reset after it fetched some records
+                // -----------------------------------
+                noOfTimeOuts = 0;
+            }
+
+            if (records != null) {
+                records.forEach(thisRecord -> {
+                    fetchedRecords.add(thisRecord);
+                    LOGGER.info("\nRecord Key - {} , Record value - {}, Record partition - {}, Record offset - {}",
+                            thisRecord.key(), thisRecord.value(), thisRecord.partition(), thisRecord.offset());
+                });
+            }
+
+            handleCommitSyncAsync(consumer, effectiveLocal);
+        }
+
+        consumer.close();
+
+        handleRecordsDump(effectiveLocal, fetchedRecords);
+
+        return prepareResult(effectiveLocal, fetchedRecords);
+
+    }
+
+    private String prepareResult(ConsumerLocalConfigs consumeLocalTestProps, List<ConsumerRecord> fetchedRecords) throws JsonProcessingException {
+        if (consumeLocalTestProps != null && !consumeLocalTestProps.getShowRecordsInResponse()) {
+
+            return objectMapper.writeValueAsString(new DeliveryStatus(OK, fetchedRecords.size()));
+
+        } else {
+            //TODO - inject this Gson
+            return prettyPrintJson(gson.toJson(new ConsumedRecords(fetchedRecords)));
+
+        }
+    }
+
+    private void handleCommitSyncAsync(Consumer<Long, String> consumer, ConsumerLocalConfigs consumeLocalTestProps) {
+        if(consumeLocalTestProps == null){
+            LOGGER.warn("[No local test configs]-Kafka client neither did `commitAsync()` nor `commitSync()`");
+            return;
+        }
+
+        Boolean effectiveCommitSync;
+        Boolean effectiveCommitAsync;
+
+        Boolean localCommitSync = consumeLocalTestProps.getCommitSync();
+        Boolean localCommitAsync = consumeLocalTestProps.getCommitAsync();
+
+        if (localCommitSync == null && localCommitAsync == null) {
+            effectiveCommitSync = consumerCommonConfigs.getCommitSync();
+            effectiveCommitAsync = consumerCommonConfigs.getCommitAsync();
+
+        } else {
+            effectiveCommitSync = localCommitSync;
+            effectiveCommitAsync = localCommitAsync;
+        }
+
+        if (effectiveCommitSync != null && effectiveCommitSync == true) {
+            consumer.commitSync();
+
+        } else if (effectiveCommitAsync != null && effectiveCommitAsync == true) {
+            consumer.commitAsync();
+
+        } else {
+            LOGGER.warn("Kafka client neither configured for `commitAsync()` nor `commitSync()`");
+        }
+
+        // ---------------------------------------------------
+        // Leave this to the user to commit it explicitly
+        // ---------------------------------------------------
+    }
+
+
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/kafka/send/KafkaSender.java b/src/main/java/org/jsmart/zerocode/core/kafka/send/KafkaSender.java
new file mode 100644
index 00000000..a69a6be3
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/kafka/send/KafkaSender.java
@@ -0,0 +1,66 @@
+package org.jsmart.zerocode.core.kafka.send;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.gson.Gson;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import com.google.inject.name.Named;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.jsmart.zerocode.core.di.provider.GsonSerDeProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
+import org.jsmart.zerocode.core.kafka.DeliveryStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.jsmart.zerocode.core.domain.ZerocodeConstants.FAILED;
+import static org.jsmart.zerocode.core.domain.ZerocodeConstants.OK;
+import static org.jsmart.zerocode.core.kafka.helper.KafkaHelper.createProducer;
+import static org.jsmart.zerocode.core.utils.SmartUtils.prettyPrintJson;
+
+@Singleton
+public class KafkaSender {
+    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSender.class);
+
+    @Inject(optional = true)
+    @Named("kafka.producer.properties")
+    private String producerPropertyFile;
+
+
+    private final ObjectMapper objectMapper = new ObjectMapperProvider().get();
+    private final Gson gson = new GsonSerDeProvider().get();
+
+
+    public String send(String brokers, String topicName, String requestJson) throws JsonProcessingException {
+        Producer<Long, String> producer = createProducer(brokers, producerPropertyFile);
+        String status = objectMapper.writeValueAsString(new DeliveryStatus(OK));
+
+        for (int index = 0; index < 1; index++) {
+
+            // Very basic constructor, use other ones for 'partition' key etc
+            final ProducerRecord<Long, String> record = new ProducerRecord<>(topicName, requestJson);
+
+            try {
+                RecordMetadata metadata = producer.send(record).get();
+                LOGGER.info("Record sent with key " + index
+                        + ", to partition " + metadata.partition()
+                        + ", with offset " + metadata.offset());
+
+
+            } catch (Exception e) {
+                LOGGER.info("Error in sending record. Exception - {} ", e);
+                String failedStatus = objectMapper.writeValueAsString(new DeliveryStatus(FAILED, e.getMessage()));
+
+                // TODO- Also send RecordMetadata metadata from above
+                return prettyPrintJson(failedStatus);
+
+            }
+        }
+
+        return prettyPrintJson(status);
+
+    }
+
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodeMultiStepsScenarioRunnerImpl.java b/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodeMultiStepsScenarioRunnerImpl.java
index c6576846..aa0bb0ee 100644
--- a/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodeMultiStepsScenarioRunnerImpl.java
+++ b/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodeMultiStepsScenarioRunnerImpl.java
@@ -3,7 +3,6 @@ package org.jsmart.zerocode.core.runner;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.google.inject.name.Named;
-import org.apache.commons.lang.StringUtils;
 import org.jsmart.zerocode.core.domain.ScenarioSpec;
 import org.jsmart.zerocode.core.domain.Step;
 import org.jsmart.zerocode.core.domain.builders.ZeroCodeExecResultBuilder;
@@ -16,7 +15,6 @@ import org.jsmart.zerocode.core.engine.preprocessor.StepExecutionState;
 import org.jsmart.zerocode.core.engine.preprocessor.ZeroCodeExternalFileProcessor;
 import org.jsmart.zerocode.core.engine.preprocessor.ZeroCodeJsonTestProcesor;
 import org.jsmart.zerocode.core.logbuilder.LogCorrelationshipPrinter;
-import org.jsmart.zerocode.core.utils.ServiceType;
 import org.junit.runner.Description;
 import org.junit.runner.notification.RunNotifier;
 import org.slf4j.Logger;
@@ -25,9 +23,10 @@ import java.time.LocalDateTime;
 import java.util.List;
 import java.util.function.BiConsumer;
 
-import static org.apache.commons.lang.StringUtils.isEmpty;
+import static org.jsmart.zerocode.core.domain.ZerocodeConstants.*;
 import static org.jsmart.zerocode.core.domain.builders.ZeroCodeExecResultBuilder.newInstance;
 import static org.jsmart.zerocode.core.engine.mocker.RestEndPointMocker.wireMockServer;
+import static org.jsmart.zerocode.core.utils.ServiceTypeUtils.serviceType;
 import static org.jsmart.zerocode.core.utils.RunnerUtils.getFullyQualifiedUrl;
 import static org.jsmart.zerocode.core.utils.SmartUtils.prettyPrintJson;
 import static org.slf4j.LoggerFactory.getLogger;
@@ -46,18 +45,28 @@ public class ZeroCodeMultiStepsScenarioRunnerImpl implements ZeroCodeMultiStepsS
     @Inject
     private JsonServiceExecutor serviceExecutor;
 
-    @Inject
+    @Inject(optional = true)
     @Named("web.application.endpoint.host")
     private String host;
 
-    @Inject
+    @Inject(optional = true)
     @Named("web.application.endpoint.port")
     private String port;
 
-    @Inject
+    @Inject(optional = true)
     @Named("web.application.endpoint.context")
     private String applicationContext;
 
+    //---------------------------------------------------
+    //kafka bootstrap servers or brokers comma separated
+    //---------------------------------------------------
+    @Inject(optional = true)
+    @Named("kafka.bootstrap.servers")
+    private String kafkaServers;
+
+
+    //guice -ends
+
     private LogCorrelationshipPrinter logCorrelationshipPrinter;
 
     private static StepNotificationHandler notificationHandler = new StepNotificationHandler();
@@ -105,7 +114,7 @@ public class ZeroCodeMultiStepsScenarioRunnerImpl implements ZeroCodeMultiStepsS
                     StepExecutionState stepExecutionState = new StepExecutionState();
 
                     /*
-                     * Create the step name as it is for the 1st stepLoopTimes ie i=0.
+                     * Create the step name as it is for the 1st stepLoopTimes i.e. i=0.
                      * Rest of the loops suffix as i ie stepName1, stepName2, stepName3 etc
                      */
                     final String thisStepName = thisStep.getName() + (i == 0 ? "" : i);
@@ -126,7 +135,9 @@ public class ZeroCodeMultiStepsScenarioRunnerImpl implements ZeroCodeMultiStepsS
                         String serviceName = thisStep.getUrl();
                         String operationName = thisStep.getOperation();
 
+                        // --------------------------------
                         // Resolve the URL patterns if any
+                        // --------------------------------
                         serviceName = zeroCodeJsonTestProcesor.resolveStringJson(
                                 serviceName,
                                 scenarioExecutionState.getResolvedScenarioState()
@@ -161,6 +172,24 @@ public class ZeroCodeMultiStepsScenarioRunnerImpl implements ZeroCodeMultiStepsS
                                 executionResult = serviceExecutor.executeJavaService(serviceName, operationName, resolvedRequestJson);
                                 break;
 
+                            case KAFKA_CALL:
+                                if(kafkaServers == null){
+                                    throw new RuntimeException(">>> 'kafka.bootstrap.servers' property can not be null for kafka operations");
+                                }
+                                printBrokerProperties();
+                                logCorrelationshipPrinter.aRequestBuilder()
+                                        .stepLoop(i)
+                                        .relationshipId(logPrefixRelationshipId)
+                                        .requestTimeStamp(requestTimeStamp)
+                                        .step(thisStepName)
+                                        .url(serviceName)
+                                        .method(operationName)
+                                        .request(prettyPrintJson(resolvedRequestJson));
+
+                                String topicName = serviceName.substring(KAFKA_TOPIC.length());
+                                executionResult = serviceExecutor.executeKafkaService(kafkaServers, topicName, operationName, resolvedRequestJson);
+                                break;
+
                             case NONE:
                                 logCorrelationshipPrinter.aRequestBuilder()
                                         .stepLoop(i)
@@ -357,18 +386,28 @@ public class ZeroCodeMultiStepsScenarioRunnerImpl implements ZeroCodeMultiStepsS
         this.applicationContext = applicationContext;
     }
 
-    private ServiceType serviceType(String serviceName, String methodName) {
-        ServiceType serviceType;
+    private String getFullyQualifiedRestUrl(String serviceEndPoint) {
 
-        if (StringUtils.isEmpty(serviceName) || isEmpty(methodName)) {
-            serviceType = ServiceType.NONE;
-        } else if (serviceName != null && serviceName.contains("/")) {
-            serviceType = ServiceType.REST_CALL;
-        } else {
-            serviceType = ServiceType.JAVA_CALL;
+        if(host == null || port == null){
+            throw new RuntimeException("'" + PROPERTY_KEY_HOST + "' or 'port' - can not be null");
+        }
+
+        if(applicationContext == null){
+            throw new RuntimeException("'" + PROPERTY_KEY_PORT + "' key must be present even if empty or blank");
         }
 
-        return serviceType;
+        if (serviceEndPoint.startsWith("http://") || serviceEndPoint.startsWith("https://")) {
+
+            return serviceEndPoint;
+
+        } else {
+            /*
+             * Make sure your property file contains context-path with a front slash like "/google-map".
+             * -OR-
+             * Empty context path is also ok if it requires. In this case do not put a front slash.
+             */
+            return String.format("%s:%s%s%s", host, port, applicationContext, serviceEndPoint);
+        }
     }
 
     private void stopWireMockServer() {
@@ -378,4 +417,13 @@ public class ZeroCodeMultiStepsScenarioRunnerImpl implements ZeroCodeMultiStepsS
             LOGGER.info("Scenario: All mockings done via WireMock server. Dependant end points executed. Stopped WireMock.");
         }
     }
+
+    private void printBrokerProperties() {
+
+        System.out.println("---------------------------------------------------------");
+        System.out.println(String.format("kafka.bootstrap.servers - %s", kafkaServers));
+        System.out.println("---------------------------------------------------------");
+
+    }
+
 }
diff --git a/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodePackageRunner.java b/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodePackageRunner.java
index 5e07eb78..cc02ad20 100644
--- a/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodePackageRunner.java
+++ b/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodePackageRunner.java
@@ -3,8 +3,8 @@ package org.jsmart.zerocode.core.runner;
 import com.google.inject.Inject;
 import com.google.inject.Injector;
 import com.google.inject.util.Modules;
-import org.jsmart.zerocode.core.di.ApplicationMainModule;
-import org.jsmart.zerocode.core.di.RuntimeHttpClientModule;
+import org.jsmart.zerocode.core.di.main.ApplicationMainModule;
+import org.jsmart.zerocode.core.di.module.RuntimeHttpClientModule;
 import org.jsmart.zerocode.core.domain.ScenarioSpec;
 import org.jsmart.zerocode.core.domain.TargetEnv;
 import org.jsmart.zerocode.core.domain.TestPackageRoot;
diff --git a/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodeUnitRunner.java b/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodeUnitRunner.java
index 511ce38c..43cb0a85 100644
--- a/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodeUnitRunner.java
+++ b/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodeUnitRunner.java
@@ -3,14 +3,17 @@ package org.jsmart.zerocode.core.runner;
 import com.google.inject.Guice;
 import com.google.inject.Injector;
 import com.google.inject.util.Modules;
-import org.jsmart.zerocode.core.di.ApplicationMainModule;
-import org.jsmart.zerocode.core.di.RuntimeHttpClientModule;
+import org.jsmart.zerocode.core.di.main.ApplicationMainModule;
+import org.jsmart.zerocode.core.di.module.RuntimeHttpClientModule;
+import org.jsmart.zerocode.core.di.module.RuntimeKafkaClientModule;
 import org.jsmart.zerocode.core.domain.*;
 import org.jsmart.zerocode.core.domain.builders.ZeroCodeExecResultBuilder;
 import org.jsmart.zerocode.core.domain.builders.ZeroCodeReportBuilder;
 import org.jsmart.zerocode.core.engine.listener.ZeroCodeTestReportListener;
 import org.jsmart.zerocode.core.httpclient.BasicHttpClient;
 import org.jsmart.zerocode.core.httpclient.ssl.SslTrustHttpClient;
+import org.jsmart.zerocode.core.kafka.client.BasicKafkaClient;
+import org.jsmart.zerocode.core.kafka.client.ZerocodeCustomKafkaClient;
 import org.jsmart.zerocode.core.logbuilder.LogCorrelationshipPrinter;
 import org.jsmart.zerocode.core.report.ZeroCodeReportGenerator;
 import org.jsmart.zerocode.core.utils.SmartUtils;
@@ -41,21 +44,19 @@ import static org.jsmart.zerocode.core.utils.RunnerUtils.getEnvSpecificConfigFil
 public class ZeroCodeUnitRunner extends BlockJUnit4ClassRunner {
     private static final Logger LOGGER = LoggerFactory.getLogger(ZeroCodeUnitRunner.class);
 
-    protected boolean passed;
-    protected boolean testRunCompleted;
     private ZeroCodeMultiStepsScenarioRunner zeroCodeMultiStepsScenarioRunner;
     private final Class<?> testClass;
     private Injector injector;
     private SmartUtils smartUtils;
-
     private HostProperties hostProperties;
     private String host;
     private String context;
     private int port;
     private List<String> smartTestCaseNames = new ArrayList<>();
     private String currentTestCase;
-
     private LogCorrelationshipPrinter logCorrelationshipPrinter;
+    protected boolean testRunCompleted;
+    protected boolean passed;
 
     private ZeroCodeMultiStepsScenarioRunner multiStepsRunner;
 
@@ -137,6 +138,39 @@ public class ZeroCodeUnitRunner extends BlockJUnit4ClassRunner {
         return zeroCodeMultiStepsScenarioRunner;
     }
 
+    public Injector getMainModuleInjector() {
+        // Synchronise this with an object lock e.g. synchronized (ZeroCodeUnitRunner.class) {}
+        synchronized (this) {
+            final TargetEnv envAnnotation = testClass.getAnnotation(TargetEnv.class);
+            String serverEnv = envAnnotation != null ? envAnnotation.value() : "config_hosts.properties";
+
+            serverEnv = getEnvSpecificConfigFile(serverEnv, testClass);
+
+            Class<? extends BasicHttpClient> runtimeHttpClient = getCustomHttpClientOrDefault();
+            Class<? extends BasicKafkaClient> runtimeKafkaClient = getCustomKafkaClientOrDefault();
+
+            injector = Guice.createInjector(Modules.override(new ApplicationMainModule(serverEnv))
+                    .with(
+                            new RuntimeHttpClientModule(runtimeHttpClient),
+                            new RuntimeKafkaClientModule(runtimeKafkaClient)
+                    )
+            );
+
+            return injector;
+        }
+    }
+
+    private Class<? extends BasicKafkaClient> getCustomKafkaClientOrDefault() {
+        final UseKafkaClient kafkaClientAnnotated = testClass.getAnnotation(UseKafkaClient.class);
+        return kafkaClientAnnotated != null ? kafkaClientAnnotated.value() : ZerocodeCustomKafkaClient.class;
+    }
+
+    private Class<? extends BasicHttpClient> getCustomHttpClientOrDefault() {
+        final UseHttpClient httpClientAnnotated = testClass.getAnnotation(UseHttpClient.class);
+        return httpClientAnnotated != null ? httpClientAnnotated.value() : SslTrustHttpClient.class;
+    }
+
+
     protected SmartUtils getInjectedSmartUtilsClass() {
         return getMainModuleInjector().getInstance(SmartUtils.class);
     }
@@ -178,25 +212,6 @@ public class ZeroCodeUnitRunner extends BlockJUnit4ClassRunner {
         notifier.fireTestFinished(description);
     }
 
-    private Injector getMainModuleInjector() {
-        // Synchronise this with an object lock e.g. synchronized (ZeroCodeUnitRunner.class) {}
-        synchronized (this) {
-            final TargetEnv envAnnotation = testClass.getAnnotation(TargetEnv.class);
-            String serverEnv = envAnnotation != null ? envAnnotation.value() : "config_hosts.properties";
-
-            serverEnv = getEnvSpecificConfigFile(serverEnv, testClass);
-
-            final UseHttpClient httpClientAnnotated = testClass.getAnnotation(UseHttpClient.class);
-            Class<? extends BasicHttpClient> runtimeHttpClient =
-                    httpClientAnnotated != null ? httpClientAnnotated.value() : SslTrustHttpClient.class;
-
-            injector = Guice.createInjector(Modules.override(new ApplicationMainModule(serverEnv))
-                    .with(new RuntimeHttpClientModule(runtimeHttpClient)));
-
-            return injector;
-        }
-    }
-
     private List<String> getSmartChildrenList() {
         List<FrameworkMethod> children = getChildren();
         children.forEach(
diff --git a/src/main/java/org/jsmart/zerocode/core/utils/HelperJsonUtils.java b/src/main/java/org/jsmart/zerocode/core/utils/HelperJsonUtils.java
index 3e264f6d..2c23b556 100755
--- a/src/main/java/org/jsmart/zerocode/core/utils/HelperJsonUtils.java
+++ b/src/main/java/org/jsmart/zerocode/core/utils/HelperJsonUtils.java
@@ -4,7 +4,7 @@ import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.lang.StringUtils;
 import org.jboss.resteasy.client.ClientRequest;
-import org.jsmart.zerocode.core.di.ObjectMapperProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
diff --git a/src/main/java/org/jsmart/zerocode/core/utils/RunnerUtils.java b/src/main/java/org/jsmart/zerocode/core/utils/RunnerUtils.java
index 705656aa..518ae247 100644
--- a/src/main/java/org/jsmart/zerocode/core/utils/RunnerUtils.java
+++ b/src/main/java/org/jsmart/zerocode/core/utils/RunnerUtils.java
@@ -14,7 +14,7 @@ import static org.jsmart.zerocode.core.utils.SmartUtils.getEnvPropertyValue;
 
 /**
  * This code is used by Unit and Package runner.
- * Hence coded at a single place here.
+ * Hence resides at a common place.
  */
 public class RunnerUtils {
     private static final Logger LOGGER = LoggerFactory.getLogger(RunnerUtils.class);
diff --git a/src/main/java/org/jsmart/zerocode/core/utils/ServiceType.java b/src/main/java/org/jsmart/zerocode/core/utils/ServiceType.java
index 3f298031..5addfc8d 100644
--- a/src/main/java/org/jsmart/zerocode/core/utils/ServiceType.java
+++ b/src/main/java/org/jsmart/zerocode/core/utils/ServiceType.java
@@ -2,6 +2,7 @@ package org.jsmart.zerocode.core.utils;
 
 public enum ServiceType {
     REST_CALL,
+    KAFKA_CALL,
     JAVA_CALL,
     NONE;
 }
diff --git a/src/main/java/org/jsmart/zerocode/core/utils/ServiceTypeUtils.java b/src/main/java/org/jsmart/zerocode/core/utils/ServiceTypeUtils.java
new file mode 100644
index 00000000..e1acc9db
--- /dev/null
+++ b/src/main/java/org/jsmart/zerocode/core/utils/ServiceTypeUtils.java
@@ -0,0 +1,30 @@
+package org.jsmart.zerocode.core.utils;
+
+import org.apache.commons.lang.StringUtils;
+
+import static org.apache.commons.lang.StringUtils.isEmpty;
+import static org.jsmart.zerocode.core.domain.ZerocodeConstants.KAFKA;
+
+public class ServiceTypeUtils {
+
+    public static ServiceType serviceType(String serviceName, String methodName) {
+        ServiceType serviceType;
+
+        if (StringUtils.isEmpty(serviceName) || isEmpty(methodName)) {
+            serviceType = ServiceType.NONE;
+
+        } else if (serviceName != null && serviceName.contains("/")) {
+            serviceType = ServiceType.REST_CALL;
+
+        } else if (serviceName != null && serviceName.contains(KAFKA)) {
+            serviceType = ServiceType.KAFKA_CALL;
+
+        } else {
+            serviceType = ServiceType.JAVA_CALL;
+
+        }
+
+        return serviceType;
+    }
+
+}
diff --git a/src/main/java/org/jsmart/zerocode/core/utils/SmartUtils.java b/src/main/java/org/jsmart/zerocode/core/utils/SmartUtils.java
index 293c8adc..6e6ff235 100644
--- a/src/main/java/org/jsmart/zerocode/core/utils/SmartUtils.java
+++ b/src/main/java/org/jsmart/zerocode/core/utils/SmartUtils.java
@@ -10,7 +10,7 @@ import com.google.common.io.Resources;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import org.apache.commons.lang.text.StrSubstitutor;
-import org.jsmart.zerocode.core.di.ObjectMapperProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.jsmart.zerocode.core.domain.ScenarioSpec;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;


Test patch:
diff --git a/src/main/java/org/jsmart/zerocode/core/engine/preprocessor/ZeroCodeJsonTestProcesorImpl.java b/src/main/java/org/jsmart/zerocode/core/engine/preprocessor/ZeroCodeJsonTestProcesorImpl.java
index 68ea9891..3a45b300 100644
--- a/src/main/java/org/jsmart/zerocode/core/engine/preprocessor/ZeroCodeJsonTestProcesorImpl.java
+++ b/src/main/java/org/jsmart/zerocode/core/engine/preprocessor/ZeroCodeJsonTestProcesorImpl.java
@@ -64,6 +64,7 @@ public class ZeroCodeJsonTestProcesorImpl implements ZeroCodeJsonTestProcesor {
     private static final String LOCALDATETIME_NOW = "LOCAL.DATETIME.NOW:";
     private static final String XML_FILE = "XML.FILE:";
     private static final String RANDOM_UU_ID = "RANDOM.UUID";
+    private static final String RECORD_DUMP = "RECORD.DUMP:";
 
     private static final List<String> availableTokens = Arrays.asList(
             PREFIX_ASU,
@@ -73,7 +74,8 @@ public class ZeroCodeJsonTestProcesorImpl implements ZeroCodeJsonTestProcesor {
             LOCALDATE_TODAY,
             LOCALDATETIME_NOW,
             XML_FILE,
-            RANDOM_UU_ID
+            RANDOM_UU_ID,
+            RECORD_DUMP
     );
 
     /*
@@ -129,6 +131,7 @@ public class ZeroCodeJsonTestProcesorImpl implements ZeroCodeJsonTestProcesor {
                         String formatPattern = runTimeToken.substring(LOCALDATETIME_NOW.length());
                         DateTimeFormatter formatter = DateTimeFormatter.ofPattern(formatPattern);
                         parammap.put(runTimeToken, LocalDateTime.now().format(formatter));
+
                     } else if (runTimeToken.startsWith(XML_FILE)) {
                         String xmlFileResource = runTimeToken.substring(XML_FILE.length());
                         final String xmlString = getXmlContent(xmlFileResource);
diff --git a/src/test/java/org/jsmart/zerocode/converter/MimeTypeConverterTest.java b/src/test/java/org/jsmart/zerocode/converter/MimeTypeConverterTest.java
index 3bcef8a5..76cf1d99 100644
--- a/src/test/java/org/jsmart/zerocode/converter/MimeTypeConverterTest.java
+++ b/src/test/java/org/jsmart/zerocode/converter/MimeTypeConverterTest.java
@@ -3,7 +3,7 @@ package org.jsmart.zerocode.converter;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.jayway.jsonpath.JsonPath;
-import org.jsmart.zerocode.core.di.ObjectMapperProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.junit.Before;
 import org.junit.Test;
 
diff --git a/src/test/java/org/jsmart/zerocode/core/di/ApplicationMainModuleBackwordCompatibilityTest.java b/src/test/java/org/jsmart/zerocode/core/di/ApplicationMainModuleBackwordCompatibilityTest.java
index 310e9201..29011746 100644
--- a/src/test/java/org/jsmart/zerocode/core/di/ApplicationMainModuleBackwordCompatibilityTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/di/ApplicationMainModuleBackwordCompatibilityTest.java
@@ -2,6 +2,7 @@ package org.jsmart.zerocode.core.di;
 
 import com.google.inject.Inject;
 import com.google.inject.name.Named;
+import org.jsmart.zerocode.core.di.main.ApplicationMainModule;
 import org.jsmart.zerocode.core.utils.SmartUtils;
 import org.jukito.JukitoRunner;
 import org.jukito.TestModule;
@@ -13,7 +14,7 @@ import org.junit.runner.RunWith;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
-//this config file is to test backword compatibility, remove after release 1.2.9
+//this config file is to test backword compatibility, remove this test only after release 1.2.9 or later releases
 @RunWith(JukitoRunner.class)
 public class ApplicationMainModuleBackwordCompatibilityTest {
 
diff --git a/src/test/java/org/jsmart/zerocode/core/di/ApplicationMainModuleTest.java b/src/test/java/org/jsmart/zerocode/core/di/ApplicationMainModuleTest.java
index 4c3112c1..e1289d60 100644
--- a/src/test/java/org/jsmart/zerocode/core/di/ApplicationMainModuleTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/di/ApplicationMainModuleTest.java
@@ -2,6 +2,7 @@ package org.jsmart.zerocode.core.di;
 
 import com.google.inject.Inject;
 import com.google.inject.name.Named;
+import org.jsmart.zerocode.core.di.main.ApplicationMainModule;
 import org.jsmart.zerocode.core.utils.SmartUtils;
 import org.jukito.JukitoRunner;
 import org.jukito.TestModule;
diff --git a/src/test/java/org/jsmart/zerocode/core/domain/MockStepTest.java b/src/test/java/org/jsmart/zerocode/core/domain/MockStepTest.java
index 525e73fc..13bc61c4 100644
--- a/src/test/java/org/jsmart/zerocode/core/domain/MockStepTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/domain/MockStepTest.java
@@ -1,7 +1,7 @@
 package org.jsmart.zerocode.core.domain;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.jsmart.zerocode.core.di.ObjectMapperProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.jsmart.zerocode.core.utils.SmartUtils;
 import org.junit.Test;
 
diff --git a/src/test/java/org/jsmart/zerocode/core/domain/ScenarioSpecTest.java b/src/test/java/org/jsmart/zerocode/core/domain/ScenarioSpecTest.java
index f2d426fc..fdaa14e3 100644
--- a/src/test/java/org/jsmart/zerocode/core/domain/ScenarioSpecTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/domain/ScenarioSpecTest.java
@@ -2,7 +2,7 @@ package org.jsmart.zerocode.core.domain;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.jsmart.zerocode.core.di.ApplicationMainModule;
+import org.jsmart.zerocode.core.di.main.ApplicationMainModule;
 import org.jsmart.zerocode.core.utils.SmartUtils;
 import org.jukito.JukitoRunner;
 import org.jukito.TestModule;
diff --git a/src/test/java/org/jsmart/zerocode/core/domain/StepTest.java b/src/test/java/org/jsmart/zerocode/core/domain/StepTest.java
index 7deeaabb..600f7e3f 100644
--- a/src/test/java/org/jsmart/zerocode/core/domain/StepTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/domain/StepTest.java
@@ -5,7 +5,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.io.Resources;
 import com.google.inject.Inject;
 import com.jayway.jsonpath.JsonPath;
-import org.jsmart.zerocode.core.di.ApplicationMainModule;
+import org.jsmart.zerocode.core.di.main.ApplicationMainModule;
 import org.jsmart.zerocode.core.utils.SmartUtils;
 import org.jukito.JukitoRunner;
 import org.jukito.TestModule;
diff --git a/src/test/java/org/jsmart/zerocode/core/domain/reports/ZeroCodeReportTest.java b/src/test/java/org/jsmart/zerocode/core/domain/reports/ZeroCodeReportTest.java
index f9bd9b59..2f8eb957 100644
--- a/src/test/java/org/jsmart/zerocode/core/domain/reports/ZeroCodeReportTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/domain/reports/ZeroCodeReportTest.java
@@ -2,7 +2,7 @@ package org.jsmart.zerocode.core.domain.reports;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.jsmart.zerocode.core.di.ObjectMapperProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.jsmart.zerocode.core.domain.builders.ZeroCodeReportBuilder;
 import org.jsmart.zerocode.core.domain.builders.ZeroCodeReportStepBuilder;
 import org.jsmart.zerocode.core.domain.builders.ZeroCodeExecResultBuilder;
diff --git a/src/test/java/org/jsmart/zerocode/core/domain/reports/chart/ZeroCodeChartKeyValueTest.java b/src/test/java/org/jsmart/zerocode/core/domain/reports/chart/ZeroCodeChartKeyValueTest.java
index 187a4741..ce198400 100644
--- a/src/test/java/org/jsmart/zerocode/core/domain/reports/chart/ZeroCodeChartKeyValueTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/domain/reports/chart/ZeroCodeChartKeyValueTest.java
@@ -1,7 +1,7 @@
 package org.jsmart.zerocode.core.domain.reports.chart;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.jsmart.zerocode.core.di.ObjectMapperProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.jsmart.zerocode.core.domain.builders.ZeroCodeChartKeyValueArrayBuilder;
 import org.jsmart.zerocode.core.domain.builders.ZeroCodeChartKeyValueBuilder;
 import org.junit.Test;
@@ -10,7 +10,6 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.stream.Collectors;
 
-import static javax.management.remote.JMXConnectionNotification.FAILED;
 import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
diff --git a/src/test/java/org/jsmart/zerocode/core/engine/executor/JavaExecutorImplTest.java b/src/test/java/org/jsmart/zerocode/core/engine/executor/JavaExecutorImplTest.java
index 382f6627..83d403c0 100644
--- a/src/test/java/org/jsmart/zerocode/core/engine/executor/JavaExecutorImplTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/engine/executor/JavaExecutorImplTest.java
@@ -3,7 +3,7 @@ package org.jsmart.zerocode.core.engine.executor;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.inject.Guice;
 import com.google.inject.Injector;
-import org.jsmart.zerocode.core.di.ApplicationMainModule;
+import org.jsmart.zerocode.core.di.main.ApplicationMainModule;
 import org.jsmart.zerocode.core.domain.ScenarioSpec;
 import org.jsmart.zerocode.core.utils.SmartUtils;
 import org.junit.Before;
diff --git a/src/test/java/org/jsmart/zerocode/core/engine/executor/JsonServiceExecutorImplTest.java b/src/test/java/org/jsmart/zerocode/core/engine/executor/JsonServiceExecutorImplTest.java
index 8efee5af..eedcf235 100644
--- a/src/test/java/org/jsmart/zerocode/core/engine/executor/JsonServiceExecutorImplTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/engine/executor/JsonServiceExecutorImplTest.java
@@ -5,7 +5,7 @@ import com.google.inject.Injector;
 import com.jayway.jsonpath.JsonPath;
 
 import org.jsmart.simulator.main.SimpleRestJsonSimulatorsMain;
-import org.jsmart.zerocode.core.di.ApplicationMainModule;
+import org.jsmart.zerocode.core.di.main.ApplicationMainModule;
 import org.jsmart.zerocode.core.domain.ScenarioSpec;
 import org.jsmart.zerocode.core.utils.SmartUtils;
 import org.junit.After;
diff --git a/src/test/java/org/jsmart/zerocode/core/engine/mocker/RestEndPointMockerTest.java b/src/test/java/org/jsmart/zerocode/core/engine/mocker/RestEndPointMockerTest.java
index b30ddbdc..478f1417 100644
--- a/src/test/java/org/jsmart/zerocode/core/engine/mocker/RestEndPointMockerTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/engine/mocker/RestEndPointMockerTest.java
@@ -16,7 +16,7 @@ import org.apache.http.impl.client.HttpClients;
 import org.jboss.resteasy.client.ClientRequest;
 import org.jboss.resteasy.client.ClientResponse;
 import org.jboss.resteasy.client.core.executors.ApacheHttpClientExecutor;
-import org.jsmart.zerocode.core.di.ApplicationMainModule;
+import org.jsmart.zerocode.core.di.main.ApplicationMainModule;
 import org.jsmart.zerocode.core.domain.MockStep;
 import org.jsmart.zerocode.core.domain.MockSteps;
 import org.jsmart.zerocode.core.domain.ScenarioSpec;
diff --git a/src/test/java/org/jsmart/zerocode/core/engine/preprocessor/ZeroCodeExternalFileProcessorImplTest.java b/src/test/java/org/jsmart/zerocode/core/engine/preprocessor/ZeroCodeExternalFileProcessorImplTest.java
index 1e40d087..38967afb 100644
--- a/src/test/java/org/jsmart/zerocode/core/engine/preprocessor/ZeroCodeExternalFileProcessorImplTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/engine/preprocessor/ZeroCodeExternalFileProcessorImplTest.java
@@ -2,7 +2,7 @@ package org.jsmart.zerocode.core.engine.preprocessor;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.jsmart.zerocode.core.di.ObjectMapperProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.jsmart.zerocode.core.domain.Step;
 import org.junit.Test;
 
diff --git a/src/test/java/org/jsmart/zerocode/core/engine/preprocessor/ZeroCodeJsonTestProcesorImplTest.java b/src/test/java/org/jsmart/zerocode/core/engine/preprocessor/ZeroCodeJsonTestProcesorImplTest.java
index b881c84c..a3ab3e32 100644
--- a/src/test/java/org/jsmart/zerocode/core/engine/preprocessor/ZeroCodeJsonTestProcesorImplTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/engine/preprocessor/ZeroCodeJsonTestProcesorImplTest.java
@@ -4,7 +4,7 @@ import com.google.inject.Guice;
 import com.google.inject.Injector;
 import com.jayway.jsonpath.JsonPath;
 import org.jsmart.simulator.main.SimpleRestJsonSimulatorsMain;
-import org.jsmart.zerocode.core.di.ApplicationMainModule;
+import org.jsmart.zerocode.core.di.main.ApplicationMainModule;
 import org.jsmart.zerocode.core.domain.ScenarioSpec;
 import org.jsmart.zerocode.core.engine.assertion.AssertionReport;
 import org.jsmart.zerocode.core.engine.assertion.JsonAsserter;
diff --git a/src/test/java/org/jsmart/zerocode/core/kafka/DeliveryStatusTest.java b/src/test/java/org/jsmart/zerocode/core/kafka/DeliveryStatusTest.java
new file mode 100644
index 00000000..33270f3a
--- /dev/null
+++ b/src/test/java/org/jsmart/zerocode/core/kafka/DeliveryStatusTest.java
@@ -0,0 +1,37 @@
+package org.jsmart.zerocode.core.kafka;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+
+public class DeliveryStatusTest {
+
+    @Test
+    public void testSerDeser() throws IOException {
+        DeliveryStatus deliveryStatus = new DeliveryStatus("Ok", "test message", 10);
+        ObjectMapper objectMapper = new ObjectMapperProvider().get();
+
+        String json = objectMapper.writeValueAsString(deliveryStatus);
+        assertThat(json, is("{\"status\":\"Ok\",\"message\":\"test message\",\"recordCount\":10}"));
+
+        DeliveryStatus javaPojo = objectMapper.readValue(json, DeliveryStatus.class);
+        assertThat(javaPojo, is(deliveryStatus));
+    }
+
+    @Test
+    public void testSerDeser_statusOnly() throws IOException {
+        DeliveryStatus deliveryStatus = new DeliveryStatus("Ok", null, null);
+        ObjectMapper objectMapper = new ObjectMapperProvider().get();
+
+        String json = objectMapper.writeValueAsString(deliveryStatus);
+        assertThat(json, is("{\"status\":\"Ok\"}"));
+
+        DeliveryStatus javaPojo = objectMapper.readValue(json, DeliveryStatus.class);
+        assertThat(javaPojo, is(deliveryStatus));
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrapTest.java b/src/test/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrapTest.java
new file mode 100644
index 00000000..706c3524
--- /dev/null
+++ b/src/test/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrapTest.java
@@ -0,0 +1,111 @@
+package org.jsmart.zerocode.core.kafka.consume;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+import static org.skyscreamer.jsonassert.JSONAssert.assertEquals;
+import static org.skyscreamer.jsonassert.JSONCompareMode.LENIENT;
+
+public class ConsumerLocalConfigsWrapTest {
+    ObjectMapper objectMapper = new ObjectMapperProvider().get();
+
+    @Test
+    public void testSerDeser() throws IOException {
+        ConsumerLocalConfigsWrap javaObject = new ConsumerLocalConfigsWrap(
+                new ConsumerLocalConfigs("RAW:/target/ttt",
+                        "RAW",
+                        true,
+                        null,
+                        null,
+                        3,
+                        50L));
+        ObjectMapper objectMapper = new ObjectMapperProvider().get();
+
+        String json = objectMapper.writeValueAsString(javaObject);
+        assertEquals("{\n" +
+                "   \"consumerLocalConfigs\": {\n" +
+                "      \"fileDumpTo\": \"RAW:/target/ttt\",\n" +
+                "      \"fileDumpType\": \"RAW\",\n" +
+                "      \"commitAsync\": true,\n" +
+                "      \"maxNoOfRetryPollsOrTimeouts\": 3,\n" +
+                "      \"pollingTime\": 50\n" +
+                "   }\n" +
+                "}",
+                json, LENIENT);
+
+        ConsumerLocalConfigsWrap javaPojo = objectMapper.readValue(json, ConsumerLocalConfigsWrap.class);
+        assertThat(javaPojo, is(javaObject));
+    }
+
+    @Test
+    public void testSerDeser_oneFieldOnly() throws IOException {
+        ConsumerLocalConfigsWrap javaObject = new ConsumerLocalConfigsWrap(
+                new ConsumerLocalConfigs("JSON:/target/ttt",
+                        "RAW",
+                        null,
+                        null,
+                        null,
+                        3,
+                        null));
+
+        String json = objectMapper.writeValueAsString(javaObject);
+        assertEquals("{\n" +
+                        "   \"consumerLocalConfigs\": {\n" +
+                        "      \"fileDumpTo\": \"JSON:/target/ttt\",\n" +
+                        "      \"fileDumpType\": \"RAW\",\n" +
+                        "      \"maxNoOfRetryPollsOrTimeouts\": 3\n" +
+                        "   }\n" +
+                        "}",
+                json, LENIENT);
+
+        ConsumerLocalConfigsWrap javaPojo = objectMapper.readValue(json, ConsumerLocalConfigsWrap.class);
+        assertThat(javaPojo, is(javaObject));
+    }
+
+    @Test
+    public void testDser_json() throws IOException {
+        String json = "{\n" +
+                "                \"consumerLocalConfigs\": {\n" +
+                "                    \"fileDumpTo\": \"target/temp/demo.txt\",\n" +
+                "                    \"fileDumpType\": \"RAW\",\n" +
+                "                    \"commitAsync\":true,\n" +
+                "                    \"showRecordsInResponse\":false\n" +
+                "                }\n" +
+                "\n" +
+                "            }";
+
+        ConsumerLocalConfigsWrap javaPojo = objectMapper.readValue(json, ConsumerLocalConfigsWrap.class);
+
+        assertThat(javaPojo.getConsumerLocalConfigs().getFileDumpTo(), is("target/temp/demo.txt"));
+        assertThat(javaPojo.getConsumerLocalConfigs().getShowRecordsInResponse(), is(false));
+
+
+    }
+
+    @Test(expected = UnrecognizedPropertyException.class)
+    public void testDser_UnRecognizedField() throws IOException {
+        String json = "{\n" +
+                "                \"consumerLocalConfigs\": {\n" +
+                "                    \"fileDumpTo\": \"target/temp/demo.txt\",\n" +
+                "                    \"fileDumpType\": \"RAW\",\n" +
+                "                    \"commitAsync\":true,\n" +
+                "                    \"showRecordsInResponse\":false,\n" +
+                "                    \"MAX_NO_OF_RETRY_POLLS_OR_TIME_OUTS\": 5\n" +
+                "                }\n" +
+                "\n" +
+                "            }";
+
+        ConsumerLocalConfigsWrap javaPojo = objectMapper.readValue(json, ConsumerLocalConfigsWrap.class);
+
+        assertThat(javaPojo.getConsumerLocalConfigs().getFileDumpTo(), is("target/temp/demo.txt"));
+        assertThat(javaPojo.getConsumerLocalConfigs().getShowRecordsInResponse(), is(false));
+
+
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaFileRecordHelperTest.java b/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaFileRecordHelperTest.java
new file mode 100644
index 00000000..fa676ac5
--- /dev/null
+++ b/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaFileRecordHelperTest.java
@@ -0,0 +1,41 @@
+package org.jsmart.zerocode.core.kafka.helper;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThat;
+
+public class KafkaFileRecordHelperTest {
+
+    @Test
+    public void test_fileDump_RAW() throws IOException {
+        ConsumerRecord consumerRecord1 = new ConsumerRecord("topic.x", 1, 1, "key1", "value1");
+        ConsumerRecord consumerRecord2 = new ConsumerRecord("topic.x", 2, 2, "key2", "value2");
+
+        List<ConsumerRecord> records = new ArrayList<>();
+        records.add(consumerRecord1);
+        records.add(consumerRecord2);
+
+        String relativeFileName = "target/tempX/ttt/demo1.txt";
+        //String absoluteFileName = "also works for full path - Tested Ok";
+        KafkaFileRecordHelper.dumpRecords(relativeFileName, records);
+
+        long lineCount = Files.lines(Paths.get(relativeFileName)).count();
+        assertThat(lineCount, is(2L));
+    }
+
+    // Caused by: java.lang.RuntimeException: Create file '//////zerocode/target/temp/demo2.txt'
+    // Exceptionjava.nio.file.AccessDeniedException: /zerocode
+    @Test(expected = RuntimeException.class)
+    public void test_dodgyFile() throws IOException {
+        String badAbsoluteFileName = "//////zerocode/target/temp/demo2.txt";
+        KafkaFileRecordHelper.dumpRecords(badAbsoluteFileName, new ArrayList<>());
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaHelperTest.java b/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaHelperTest.java
new file mode 100644
index 00000000..b854d028
--- /dev/null
+++ b/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaHelperTest.java
@@ -0,0 +1,123 @@
+package org.jsmart.zerocode.core.kafka.helper;
+
+import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigs;
+import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigsWrap;
+import org.jsmart.zerocode.core.kafka.receive.ConsumerCommonConfigs;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+import static org.hamcrest.core.IsNull.nullValue;
+import static org.jsmart.zerocode.core.kafka.helper.KafkaHelper.deriveEffectiveConfigs;
+
+public class KafkaHelperTest {
+
+    ConsumerCommonConfigs consumerCommon;
+    ConsumerLocalConfigs consumerLocal;
+
+    //ObjectMapper objectMapper = (new ObjectMapperProvider()).get();
+
+    @Rule
+    public ExpectedException expectedException = ExpectedException.none();
+
+    @Test
+    public void test_syncAsyncTrueCommon() throws Exception{
+        consumerCommon = new ConsumerCommonConfigs(true, true, "aTestFile", "JSON", true, 3, 50L);
+
+        expectedException.expectMessage("Both commitSync and commitAsync can not be true");
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(null, consumerCommon);
+    }
+
+    @Test
+    public void test_syncAsyncTrueLocal() throws Exception{
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L);
+        consumerLocal = new ConsumerLocalConfigs("sTestLocalFile", "RAW", true, true, false, 3, 50L);
+        ConsumerLocalConfigsWrap localConfigsWrap = new ConsumerLocalConfigsWrap(consumerLocal);
+
+        expectedException.expectMessage("Both commitSync and commitAsync can not be true");
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+    }
+
+    @Test
+    public void test_effectiveConfigsIsLocal() throws Exception{
+
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L);
+        consumerLocal = new ConsumerLocalConfigs("sTestLocalFile", "RAW", true, false, false, 3, 150L);
+
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+
+        assertThat(consumerEffectiveConfigs.getFileDumpTo(), is("sTestLocalFile"));
+        assertThat(consumerEffectiveConfigs.getFileDumpType(), is("RAW"));
+        assertThat(consumerEffectiveConfigs.getCommitAsync(), is(true));
+        assertThat(consumerEffectiveConfigs.getCommitSync(), is(false));
+        assertThat(consumerEffectiveConfigs.getShowRecordsInResponse(), is(false));
+        assertThat(consumerEffectiveConfigs.getMaxNoOfRetryPollsOrTimeouts(), is(3));
+        assertThat(consumerEffectiveConfigs.getPollingTime(), is(consumerLocal.getPollingTime()));
+    }
+
+    @Test
+    public void test_effectiveConfigsIsCentrall() throws Exception{
+
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L);
+        consumerLocal = null;
+
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+
+        assertThat(consumerEffectiveConfigs.getFileDumpTo(), is("aTestFile"));
+        assertThat(consumerEffectiveConfigs.getFileDumpType(), is("JSON"));
+        assertThat(consumerEffectiveConfigs.getCommitAsync(), is(false));
+        assertThat(consumerEffectiveConfigs.getCommitSync(), is(true));
+        assertThat(consumerEffectiveConfigs.getShowRecordsInResponse(), is(true));
+        assertThat(consumerEffectiveConfigs.getPollingTime(), is(consumerCommon.getPollingTime()));
+    }
+
+    @Test
+    public void test_effectiveCommitAsync_true() throws Exception{
+
+        consumerCommon = new ConsumerCommonConfigs(true, null, "aTestFile", "JSON", true, 3, 50L);
+        consumerLocal = new ConsumerLocalConfigs("sTestLocalFile", "RAW", true, false, false, 3, 50L);
+
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+
+        assertThat(consumerEffectiveConfigs.getCommitAsync(), is(true));
+        assertThat(consumerEffectiveConfigs.getCommitSync(), is(false));
+    }
+
+    @Test
+    public void test_effectiveCommitSync_true() throws Exception{
+
+        consumerCommon = new ConsumerCommonConfigs(null, true, "aTestFile", "JSON", true, 3, 50L);
+        consumerLocal = new ConsumerLocalConfigs("sTestLocalFile", "RAW", null, true, false, 3, 50L);
+
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+
+        assertThat(consumerEffectiveConfigs.getCommitSync(), is(true));
+        assertThat(consumerEffectiveConfigs.getCommitAsync(), nullValue());
+    }
+
+    @Test
+    public void test_effectiveCommitSyncFromCommon_true() throws Exception{
+
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L);
+        consumerLocal = new ConsumerLocalConfigs("sTestLocalFile", "RAW", null, null, false, 3, 50L);
+
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+
+        assertThat(consumerEffectiveConfigs.getCommitSync(), is(true));
+        assertThat(consumerEffectiveConfigs.getCommitAsync(), is(false));
+    }
+
+    @Test
+    public void test_effectiveCommitAsyncFromCommon_true() throws Exception{
+
+        consumerCommon = new ConsumerCommonConfigs(null, true, "aTestFile", "JSON", true, 3,50L);
+        consumerLocal = new ConsumerLocalConfigs("sTestLocalFile", "RAW", true, false, false, 3, 150L);
+
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+
+        assertThat(consumerEffectiveConfigs.getCommitAsync(), is(true));
+        assertThat(consumerEffectiveConfigs.getCommitSync(), is(false));
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiverTest.java b/src/test/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiverTest.java
new file mode 100644
index 00000000..56b2a132
--- /dev/null
+++ b/src/test/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiverTest.java
@@ -0,0 +1,6 @@
+package org.jsmart.zerocode.core.kafka.receive;
+
+public class KafkaReceiverTest {
+    KafkaReceiver kafkaReceiver;
+
+}
\ No newline at end of file
diff --git a/src/test/java/org/jsmart/zerocode/core/runner/ZeroCodeMultiStepsScenarioRunnerImplE2ETest.java b/src/test/java/org/jsmart/zerocode/core/runner/ZeroCodeMultiStepsScenarioRunnerImplE2ETest.java
index 9372da34..2e3d4f54 100644
--- a/src/test/java/org/jsmart/zerocode/core/runner/ZeroCodeMultiStepsScenarioRunnerImplE2ETest.java
+++ b/src/test/java/org/jsmart/zerocode/core/runner/ZeroCodeMultiStepsScenarioRunnerImplE2ETest.java
@@ -1,7 +1,7 @@
 package org.jsmart.zerocode.core.runner;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.jsmart.zerocode.core.di.ObjectMapperProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.jsmart.zerocode.core.domain.reports.ZeroCodeReport;
 import org.jsmart.zerocode.core.runner.e2e.IgnoreTestFailureFlagRunE2E;
 import org.junit.Test;
diff --git a/src/test/java/org/jsmart/zerocode/core/utils/SmartUtilsTest.java b/src/test/java/org/jsmart/zerocode/core/utils/SmartUtilsTest.java
index 3e81f14a..712c984c 100644
--- a/src/test/java/org/jsmart/zerocode/core/utils/SmartUtilsTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/utils/SmartUtilsTest.java
@@ -1,7 +1,7 @@
 package org.jsmart.zerocode.core.utils;
 
 import com.google.inject.Inject;
-import org.jsmart.zerocode.core.di.ApplicationMainModule;
+import org.jsmart.zerocode.core.di.main.ApplicationMainModule;
 import org.jsmart.zerocode.core.domain.ScenarioSpec;
 import org.jsmart.zerocode.core.domain.Step;
 import org.jukito.JukitoRunner;
diff --git a/src/test/java/org/jsmart/zerocode/core/verification/loopreport/SmartJUnitNavigatorReportGenTest.java b/src/test/java/org/jsmart/zerocode/core/verification/loopreport/SmartJUnitNavigatorReportGenTest.java
index d921c7dd..f863e49d 100644
--- a/src/test/java/org/jsmart/zerocode/core/verification/loopreport/SmartJUnitNavigatorReportGenTest.java
+++ b/src/test/java/org/jsmart/zerocode/core/verification/loopreport/SmartJUnitNavigatorReportGenTest.java
@@ -3,7 +3,7 @@ package org.jsmart.zerocode.core.verification.loopreport;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.io.FilenameUtils;
-import org.jsmart.zerocode.core.di.ObjectMapperProvider;
+import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.jsmart.zerocode.core.domain.reports.ZeroCodeReport;
 import org.jsmart.zerocode.core.utils.SmartUtils;
 import org.junit.Test;
diff --git a/src/test/java/org/jsmart/zerocode/kafka/consume/KafkaConsumeTest.java b/src/test/java/org/jsmart/zerocode/kafka/consume/KafkaConsumeTest.java
new file mode 100644
index 00000000..4dab74e2
--- /dev/null
+++ b/src/test/java/org/jsmart/zerocode/kafka/consume/KafkaConsumeTest.java
@@ -0,0 +1,39 @@
+package org.jsmart.zerocode.kafka.consume;
+
+import org.jsmart.zerocode.core.domain.JsonTestCase;
+import org.jsmart.zerocode.core.domain.TargetEnv;
+import org.jsmart.zerocode.core.runner.ZeroCodeUnitRunner;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@Ignore("Ignored for CI build until docker ci build setup is complete. But can be run manually against a kafka server.")
+@TargetEnv("hosts_servers/kafka_test_server.properties")
+@RunWith(ZeroCodeUnitRunner.class)
+public class KafkaConsumeTest {
+
+    @Test
+    @JsonTestCase("kafka/consume/test_kafka_subscribe.json")
+    public void testKafkaConsume() throws Exception {
+    }
+
+    @Test
+    @JsonTestCase("kafka/consume/file_dump/test_kafka_consume_raw_record_dump.json")
+    public void testKafkaRawRecordDump() throws Exception {
+    }
+
+    @Ignore("TODO wip_")
+    @Test
+    @JsonTestCase("kafka/consume/WIP_test_kafka_consume_with_properties.json")
+    public void testKafkaLocalProperties() throws Exception {
+    }
+
+    @Ignore("TODO wip_")
+    @Test
+    @JsonTestCase("kafka/consume/file_dump/WIP_test_kafka_consume_json_dump.json")
+    public void testKafkaJsonDump() throws Exception {
+    }
+
+
+
+}
diff --git a/src/test/java/org/jsmart/zerocode/kafka/customclient/KafkaProduceCustomClientTest.java b/src/test/java/org/jsmart/zerocode/kafka/customclient/KafkaProduceCustomClientTest.java
new file mode 100644
index 00000000..efcec810
--- /dev/null
+++ b/src/test/java/org/jsmart/zerocode/kafka/customclient/KafkaProduceCustomClientTest.java
@@ -0,0 +1,23 @@
+package org.jsmart.zerocode.kafka.customclient;
+
+import org.jsmart.zerocode.core.domain.JsonTestCase;
+import org.jsmart.zerocode.core.domain.TargetEnv;
+import org.jsmart.zerocode.core.domain.UseKafkaClient;
+import org.jsmart.zerocode.core.kafka.client.ZerocodeCustomKafkaClient;
+import org.jsmart.zerocode.core.runner.ZeroCodeUnitRunner;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@Ignore("Ignored for CI build until docker ci build setup is complete. But can be run manually against a kafka server.")
+@TargetEnv("hosts_servers/kafka_test_server.properties")
+@UseKafkaClient(ZerocodeCustomKafkaClient.class)
+@RunWith(ZeroCodeUnitRunner.class)
+public class KafkaProduceCustomClientTest {
+
+    @Test
+    @JsonTestCase("kafka/produce/test_kafka_publish.json")
+    public void testPublish() throws Exception {
+    }
+
+}
diff --git a/src/test/java/org/jsmart/zerocode/kafka/produce/KafkaProduceTest.java b/src/test/java/org/jsmart/zerocode/kafka/produce/KafkaProduceTest.java
new file mode 100644
index 00000000..181c36b2
--- /dev/null
+++ b/src/test/java/org/jsmart/zerocode/kafka/produce/KafkaProduceTest.java
@@ -0,0 +1,20 @@
+package org.jsmart.zerocode.kafka.produce;
+
+import org.jsmart.zerocode.core.domain.JsonTestCase;
+import org.jsmart.zerocode.core.domain.TargetEnv;
+import org.jsmart.zerocode.core.runner.ZeroCodeUnitRunner;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@Ignore("Ignored for CI build until docker ci build setup is complete. But can be run manually against a kafka server.")
+@TargetEnv("hosts_servers/kafka_test_server.properties")
+@RunWith(ZeroCodeUnitRunner.class)
+public class KafkaProduceTest {
+
+    @Test
+    @JsonTestCase("kafka/produce/test_kafka_publish.json")
+    public void testPublish() throws Exception {
+    }
+
+}
diff --git a/src/test/java/org/jsmart/zerocode/kafka/produce/KafkaPublishFailureTest.java b/src/test/java/org/jsmart/zerocode/kafka/produce/KafkaPublishFailureTest.java
new file mode 100644
index 00000000..c7bb63a1
--- /dev/null
+++ b/src/test/java/org/jsmart/zerocode/kafka/produce/KafkaPublishFailureTest.java
@@ -0,0 +1,21 @@
+package org.jsmart.zerocode.kafka.produce;
+
+import org.jsmart.zerocode.core.domain.JsonTestCase;
+import org.jsmart.zerocode.core.domain.TargetEnv;
+import org.jsmart.zerocode.core.runner.ZeroCodeUnitRunner;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@TargetEnv("hosts_servers/kafka_test_bad_server.properties")
+@RunWith(ZeroCodeUnitRunner.class)
+public class KafkaPublishFailureTest {
+
+    @Ignore("Sometimes - 2018-10-06 23:33:15,673 [main] WARN org.apache.kafka.common.utils.AppInfoParser - Error registering AppInfo mbean\n" +
+            "javax.management.InstanceAlreadyExistsException: kafka.producer:type=app-info,id=zerocode-producer\n")
+    @Test
+    @JsonTestCase("kafka/produce/test_kafka_publish_failed.json")
+    public void testPublish() throws Exception {
+    }
+
+}
diff --git a/src/test/resources/hosts_servers/kafka_consumer.properties b/src/test/resources/hosts_servers/kafka_consumer.properties
new file mode 100755
index 00000000..8db7d575
--- /dev/null
+++ b/src/test/resources/hosts_servers/kafka_consumer.properties
@@ -0,0 +1,26 @@
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+#          kafka consumer properties
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+group.id=consumerGroup14
+key.deserializer=org.apache.kafka.common.serialization.LongDeserializer
+value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
+max.poll.records=2
+enable.auto.commit=false
+auto.offset.reset=earliest
+
+#
+#bootstrap.servers=localhost:9092
+#group.id=None
+#enable.auto.commit=true
+#key.deserializer=org.apache.kafka.common.serialization.LongDeserializer
+#value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
+#
+## fast session timeout makes it more fun to play with failover
+#session.timeout.ms=10000
+#
+## These buffer sizes seem to be needed to avoid consumer switching to
+## a mode where it processes one bufferful every 5 seconds with multiple
+## timeouts along the way.  No idea why this happens.
+#fetch.min.bytes=50000
+#receive.buffer.bytes=262144
+#max.partition.fetch.bytes=2097152
\ No newline at end of file
diff --git a/src/test/resources/hosts_servers/kafka_producer.properties b/src/test/resources/hosts_servers/kafka_producer.properties
new file mode 100755
index 00000000..c4bb9581
--- /dev/null
+++ b/src/test/resources/hosts_servers/kafka_producer.properties
@@ -0,0 +1,16 @@
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+#          kafka producer properties
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+client.id=zerocode-producer
+key.serializer=org.apache.kafka.common.serialization.LongSerializer
+value.serializer=org.apache.kafka.common.serialization.StringSerializer
+
+#acks=all
+#retries=0
+#batch.size=16384
+#client.id=fcd-producer
+#auto.commit.interval.ms=1000
+#linger.ms=0
+#key.serializer=org.apache.kafka.common.serialization.StringSerializer
+#value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
+#block.on.buffer.full=true
diff --git a/src/test/resources/hosts_servers/kafka_test_bad_server.properties b/src/test/resources/hosts_servers/kafka_test_bad_server.properties
new file mode 100755
index 00000000..34e96e93
--- /dev/null
+++ b/src/test/resources/hosts_servers/kafka_test_bad_server.properties
@@ -0,0 +1,6 @@
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+# kafka bootstrap servers comma separated
+# e.g. localhost:9092,host2:9093
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+kafka.bootstrap.servers=127.1.1.1:9092
+
diff --git a/src/test/resources/hosts_servers/kafka_test_server.properties b/src/test/resources/hosts_servers/kafka_test_server.properties
new file mode 100755
index 00000000..d25e8c42
--- /dev/null
+++ b/src/test/resources/hosts_servers/kafka_test_server.properties
@@ -0,0 +1,40 @@
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+# kafka bootstrap servers comma separated
+# e.g. localhost:9092,host2:9093
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+kafka.bootstrap.servers=localhost:9092
+
+kafka.producer.properties=hosts_servers/kafka_producer.properties
+kafka.consumer.properties=hosts_servers/kafka_consumer.properties
+
+# --------------------------------------------------------------------
+# Optional local consumer properties common/central to all test cases.
+# These can be overwritten by the tests locally.
+# --------------------------------------------------------------------
+# If this property is set, then the consumer does a commitSync after reading the message(s)
+# Make sure you don't set both commitSync and commitAsync to true
+consumer.commitSync = true
+# If this property is set, then the consumer does a commitAsync after reading the message(s)
+# Make sure you don't set both commitSync and commitAsync to true
+consumer.commitAsync = false
+# All records those were read are dumped to this specified file path
+# This path can be a relative path or an absolute path. If the file
+# does not exist, it creates the file and dumps the records
+consumer.fileDumpTo= target/temp/demo.txt
+# While dumping the file, it has to be dumped in this format
+consumer.fileDumpType= RAW
+# If this property is set to true, all records are shown in the response.
+# When dealing with large number of records, you might not be interested
+# in the individual records, but interested in the recordCount
+# i.e. total number of records consumed
+consumer.showRecordsInResponse=false
+# That means if any record(s) are read, then this counter is reset to 0(zero) and the consumer
+# polls again. So if no records are fetched for a specific poll interval, then the consumer
+# gives a retry retrying until this max number polls/reties reached.
+consumer.maxNoOfRetryPollsOrTimeouts = 5
+# Polling time in milli seconds i.e how long the consumer should poll before
+# the next retry poll
+consumer.pollingTime = 1000
+
+# local producer properties
+producer.key1=value1-testv ycvb
diff --git a/src/test/resources/kafka/commands/test_kafka_command_execution.json b/src/test/resources/kafka/commands/test_kafka_command_execution.json
new file mode 100755
index 00000000..d57d43e7
--- /dev/null
+++ b/src/test/resources/kafka/commands/test_kafka_command_execution.json
@@ -0,0 +1,18 @@
+{
+    "scenarioName": "Meta-data - Execute a command or equivalent Java function",
+    "steps": [
+        {
+            "name": "load_kafka",
+            "url": "kafka-topic://or//kafka-broker//against-what",
+            "operation": "any-kafka-command-to-execute against the topic demo",
+            "request": {
+                "name": "demo",
+                "broker-name": "kafka-server.corpbank.com"
+            },
+            "assertions": {
+                "status": "OK"
+            }
+        }
+
+    ]
+}
diff --git a/src/test/resources/kafka/consume/WIP_test_kafka_consume_with_properties.json b/src/test/resources/kafka/consume/WIP_test_kafka_consume_with_properties.json
new file mode 100755
index 00000000..be6fa6ba
--- /dev/null
+++ b/src/test/resources/kafka/consume/WIP_test_kafka_consume_with_properties.json
@@ -0,0 +1,31 @@
+{
+    "scenarioName": "Unload - consume a message from kafka -wtih local optional properties",
+    "steps": [
+        {
+            "name": "load_kafka",
+            "url": "kafka-topic:demo",
+            "operation": "load",
+            "request": {
+                "key": "${RANDOM.NUMBER}",
+                "value": "Hello World"
+            },
+            "assertions": {
+                "status" : "Ok"
+            }
+        },
+        {
+            "name": "onload_kafka",
+            "url": "kafka-topic:demo",
+            "operation": "unload",
+            "request": {
+                "consumer_properties": {
+                    "commitAsync":true,
+                    "MAX_NO_OF_RETRY_POLLS_OR_TIME_OUTS": 5
+                }
+            },
+            "assertions": {
+                "records.SIZE" : 1
+            }
+        }
+    ]
+}
diff --git a/src/test/resources/kafka/consume/file_dump/WIP_test_kafka_consume_file_dump_record_count.json b/src/test/resources/kafka/consume/file_dump/WIP_test_kafka_consume_file_dump_record_count.json
new file mode 100755
index 00000000..8ee9068d
--- /dev/null
+++ b/src/test/resources/kafka/consume/file_dump/WIP_test_kafka_consume_file_dump_record_count.json
@@ -0,0 +1,27 @@
+{
+    "scenarioName": "Unload - consume a message from kafka -fileDumpTo record count",
+    "steps": [
+        {
+            "name": "load_kafka",
+            "url": "kafka-topic:demo",
+            "operation": "produce",
+            "request": {
+                "key": "${RANDOM.NUMBER}",
+                "value": "Hello World"
+            },
+            "assertions": {
+                "status" : "Ok"
+            }
+        },
+        {
+            "name": "record_count_from_file",
+            "url": "file:target/filename",
+            "operation": "countRecord",
+            "request": {
+            },
+            "assertions": {
+                "record_count": 1
+            }
+        }
+    ]
+}
diff --git a/src/test/resources/kafka/consume/file_dump/WIP_test_kafka_consume_json_dump.json b/src/test/resources/kafka/consume/file_dump/WIP_test_kafka_consume_json_dump.json
new file mode 100755
index 00000000..c86e27fc
--- /dev/null
+++ b/src/test/resources/kafka/consume/file_dump/WIP_test_kafka_consume_json_dump.json
@@ -0,0 +1,29 @@
+{
+    "scenarioName": "Unload - consume a message from kafka -json dump to file",
+    "steps": [
+        {
+            "name": "load_kafka",
+            "url": "kafka-topic:demo",
+            "operation": "produce",
+            "request": {
+                "key": "${RANDOM.NUMBER}",
+                "value": "Hello World"
+            },
+            "assertions": {
+                "status" : "Ok"
+            }
+        },
+        {
+            "name": "unload_kafka",
+            "url": "kafka-topic:demo",
+            "operation": "consume",
+            "request": {
+                "action": "$JSON.DUMP:target/filename"
+            },
+            "assertions": {
+                "status" : "Ok",
+                "record_count": 1
+            }
+        }
+    ]
+}
diff --git a/src/test/resources/kafka/consume/file_dump/test_kafka_consume_raw_record_dump.json b/src/test/resources/kafka/consume/file_dump/test_kafka_consume_raw_record_dump.json
new file mode 100755
index 00000000..4fff1490
--- /dev/null
+++ b/src/test/resources/kafka/consume/file_dump/test_kafka_consume_raw_record_dump.json
@@ -0,0 +1,53 @@
+{
+   "scenarioName": "Unload - consume a message from kafka - raw dump to file",
+   "steps": [
+      {
+         "name": "unload_and_commit",
+         "url": "kafka-topic:demo",
+         "operation": "consume",
+         "request": {
+            "consumerLocalConfigs": {
+               //"commitSync": true,  //Uncomment to test validation
+               //"commitAsync": true, //Uncomment to test validation
+               "showRecordsInResponse": false,
+               "maxNoOfRetryPollsOrTimeouts": 5
+            }
+         },
+         "assertions": {
+         }
+      },
+      {
+         "name": "load_kafka",
+         "url": "kafka-topic:demo",
+         "operation": "produce",
+         "request": {
+            "key": "${RANDOM.NUMBER}",
+            "value": "Hello World"
+         },
+         "assertions": {
+            "status": "Ok"
+         }
+      },
+      {
+         "name": "dump_file_record_count",
+         "url": "kafka-topic:demo",
+         "operation": "consume",
+         "request": {
+            "consumerLocalConfigs": {
+               "fileDumpTo": "target/temp/demo.txt",
+               "fileDumpType": "RAW",
+               //"commitSync":true, // not effective, bacaz it does a commit after consuming the records.
+               // See step 'unload_and_commit', after which the reading starts fresh from the next offset.
+               // Otherwise you get the earlier records form last committed offset.
+               "showRecordsInResponse": false,
+               "maxNoOfRetryPollsOrTimeouts": 5,
+               "pollingTime": 1000
+            }
+         },
+         "assertions": {
+            "status": "Ok",
+            "recordCount": 1
+         }
+      }
+   ]
+}
diff --git a/src/test/resources/kafka/consume/test_kafka_subscribe.json b/src/test/resources/kafka/consume/test_kafka_subscribe.json
new file mode 100755
index 00000000..b7e0b71d
--- /dev/null
+++ b/src/test/resources/kafka/consume/test_kafka_subscribe.json
@@ -0,0 +1,30 @@
+{
+    "scenarioName": "Unload - consume a message from kafka",
+    "steps": [
+        {
+            "name": "load_kafka",
+            "url": "kafka-topic:demo",
+            "operation": "load",
+            "request": {
+                "key": "${RANDOM.NUMBER}",
+                "value": "Hello World"
+            },
+            "assertions": {
+                "status" : "Ok"
+            }
+        },
+        {
+            "name": "onload_kafka",
+            "url": "kafka-topic:demo",
+            "operation": "unload",
+            "request": {
+            },
+            "assertions": {
+                // this will be 1  only when consumer does a commit i.e. commitAsync or sync,
+                // otherwise 3 or more as earler messages will still be there
+                // TODO- commitAsync - sometimes delayed as it is async(provide a helloworld example).
+                "recordCount" : "$GT.0"
+            }
+        }
+    ]
+}
diff --git a/src/test/resources/kafka/metadata/WIP_test_kafka_topic_details.json b/src/test/resources/kafka/metadata/WIP_test_kafka_topic_details.json
new file mode 100755
index 00000000..2d3ee64e
--- /dev/null
+++ b/src/test/resources/kafka/metadata/WIP_test_kafka_topic_details.json
@@ -0,0 +1,18 @@
+{
+    "scenarioName": "Meta-data - Get topic details",
+    "steps": [
+        {
+            "name": "load_kafka",
+            "url": "kafka-topic://demo",
+            "operation": "any-kafka-command-to-execute against the topic demo",
+            "request": {
+                "name": "demo",
+                "broker-name": "kafka-server.corpbank.com"
+            },
+            "assertions": {
+                "status": "OK"
+            }
+        }
+
+    ]
+}
diff --git a/src/test/resources/kafka/pfiles/data_1of2.json b/src/test/resources/kafka/pfiles/data_1of2.json
new file mode 100644
index 00000000..fb974e4b
--- /dev/null
+++ b/src/test/resources/kafka/pfiles/data_1of2.json
@@ -0,0 +1,2 @@
+{"key":"1539010017093","value":"Hello World 1"}
+{"key":"1539010017094","value":"Hello World 2"}
diff --git a/src/test/resources/kafka/produce/WIP_test_kafka_publish_inline_server.json b/src/test/resources/kafka/produce/WIP_test_kafka_publish_inline_server.json
new file mode 100755
index 00000000..abd8b688
--- /dev/null
+++ b/src/test/resources/kafka/produce/WIP_test_kafka_publish_inline_server.json
@@ -0,0 +1,17 @@
+{
+    "scenarioName": "Load - publish a message to kafka",
+    "steps": [
+        {
+            "name": "load_kafka",
+            "url": "kafka-topic:192.168.99.1:9092:demo",
+            "operation": "load",
+            "request": {
+                "key": "${RANDOM.NUMBER}",
+                "value": "Hello World"
+            },
+            "assertions": {
+                "status": "Ok"
+            }
+        }
+    ]
+}
diff --git a/src/test/resources/kafka/produce/file_produce/WIP_test_kafka_produce_from_file_WIP_.json b/src/test/resources/kafka/produce/file_produce/WIP_test_kafka_produce_from_file_WIP_.json
new file mode 100755
index 00000000..554e8a45
--- /dev/null
+++ b/src/test/resources/kafka/produce/file_produce/WIP_test_kafka_produce_from_file_WIP_.json
@@ -0,0 +1,42 @@
+{
+    "scenarioName": "Prouce - messages to kafka - from file(s)",
+    "steps": [
+        {
+            "name": "load_kafka_from_files",
+            "url": "kafka-topic:demo",
+            "operation": "produce",
+            "request": {
+                "produce_local_conf": {
+                    "files": [
+                        {
+                            "fileName":"kafka/pfiles/data_1of2.json"
+                            //"fileType":"RAW"
+                        }
+                    ]
+                }
+            },
+            "assertions": {
+                "status": "Ok"
+            }
+        },
+        {
+            "name": "dump_file_record_count",
+            "url": "kafka-topic:demo",
+            "operation": "consume",
+            "request": {
+                "consumerLocalConfigs": {
+                    "fileDumpTo": "target/temp/demo.txt",
+                    "fileDumpType": "RAW",
+                    "commitAsync":true,
+                    "showRecordsInResponse":false,
+                    "MAX_NO_OF_RETRY_POLLS_OR_TIME_OUTS": 5
+                }
+
+            },
+            "assertions": {
+                "status": "Ok",
+                "recordCount": 1
+            }
+        }
+    ]
+}
diff --git a/src/test/resources/kafka/produce/test_kafka_publish.json b/src/test/resources/kafka/produce/test_kafka_publish.json
new file mode 100755
index 00000000..c8d26a29
--- /dev/null
+++ b/src/test/resources/kafka/produce/test_kafka_publish.json
@@ -0,0 +1,17 @@
+{
+    "scenarioName": "Load - publish a message to kafka - vanilla",
+    "steps": [
+        {
+            "name": "load_kafka",
+            "url": "kafka-topic:demo",
+            "operation": "load",
+            "request": {
+                "key": "${RANDOM.NUMBER}",
+                "value": "Hello World"
+            },
+            "assertions": {
+                "status" : "Ok"
+            }
+        }
+    ]
+}
diff --git a/src/test/resources/kafka/produce/test_kafka_publish_failed.json b/src/test/resources/kafka/produce/test_kafka_publish_failed.json
new file mode 100755
index 00000000..be1e63f5
--- /dev/null
+++ b/src/test/resources/kafka/produce/test_kafka_publish_failed.json
@@ -0,0 +1,28 @@
+{
+    "scenarioName": "Load - publish a message to kafka - valid-ip format, but no kafka",
+    "steps": [
+        {
+            "name": "load_kafka",
+            "url": "kafka-topic:demo-x",
+            "operation": "load",
+            "request": {
+                "key": "${RANDOM.NUMBER}",
+                "value": "Hello World"
+            },
+            "assertions": {
+                "status" : "Failed",
+                "message" : "org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms."
+            }
+        }
+    ]
+}
+
+// When you provide localhost:9092 : Then it means, the host is valid n able to ping localhost, if there is no kafka,
+// then only, it keeps on trying with below message-
+
+// 2018-09-28 13:07:03,628 [kafka-producer-network-thread | client1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=client1] Error while fetching metadata with correlation id 559 : {kafka-topic:demo=INVALID_TOPIC_EXCEPTION}
+// 2018-09-28 13:07:03,735 [kafka-producer-network-thread | client1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=client1] Error while fetching metadata with correlation id 560 : {kafka-topic:demo=INVALID_TOPIC_EXCEPTION}
+// Error in sending record
+// java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
+
+// Note - the final exception message is the same, hence asserted.
\ No newline at end of file