diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/java/org/apache/wayang/hackit/shipper/kafka/receiver/ReceiverKafka.java b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/java/org/apache/wayang/hackit/shipper/kafka/receiver/ReceiverKafka.java
new file mode 100644
index 0000000..49dab87
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/java/org/apache/wayang/hackit/shipper/kafka/receiver/ReceiverKafka.java
@@ -0,0 +1,89 @@
+package org.apache.wayang.hackit.shipper.kafka.receiver;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.PSProtocol;
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.receiver.Receiver;
+import org.apache.wayang.plugin.hackit.core.tuple.HackitTuple;
+import org.apache.wayang.plugin.hackit.core.tuple.header.Header;
+
+import java.util.*;
+
+public class ReceiverKafka<K, T> extends Receiver<HackitTuple<K, T>> implements PSProtocol {
+
+    //TODO Get from configuration
+    static Map<String, String> KAFKA_MAPPING;
+    static {
+        KAFKA_MAPPING = new HashMap<>();
+        KAFKA_MAPPING.put("127.0.0.1", "127.0.0.1");
+    }
+    static Integer numPartitions = 1;
+    static Short replicationFactor = 1;
+
+
+    Consumer<K, T> consumer;
+    Properties config;
+    List<String> topics;
+
+    public ReceiverKafka(Properties config){
+        this.config = config;
+        this.topics = new ArrayList<>();
+    }
+
+
+    @Override
+    public PSProtocol addTopic(String... topic) {
+
+        this.topics.addAll(Arrays.asList(topic));
+        this.consumer.subscribe(this.topics);
+        return this;
+    }
+
+    @Override
+    public PSProtocol addExchange(String exchange) {
+        return null;
+    }
+
+    @Override
+    public void init() {
+        this.consumer =
+                new KafkaConsumer<>(config);
+    }
+
+    @Override
+    public Iterator<HackitTuple<K, T>> getElements() {
+
+        final int giveUp = 100;   int noRecordsCount = 0;
+        ConsumerRecords<K, T> consumerRecords;
+        while (true) {
+            consumerRecords =
+                    consumer.poll(1000);
+
+            if (consumerRecords.count()==0) {
+                noRecordsCount++;
+                if (noRecordsCount > giveUp) break;
+                else continue;
+            }
+
+            List<HackitTuple<K, T>> list = new ArrayList<>();
+            consumerRecords.forEach(record ->{
+                HackitTuple<K, T> result = new HackitTuple<>(
+                        record.value()
+                );
+                // System.out.println("received " + record.value());
+                list.add(result);
+            });
+
+            consumer.commitAsync();
+
+            return list.listIterator();
+        }
+        return null;
+    }
+
+    @Override
+    public void close() {
+        consumer.close();
+    }
+}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/java/org/apache/wayang/hackit/shipper/kafka/sender/SenderKafka.java b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/java/org/apache/wayang/hackit/shipper/kafka/sender/SenderKafka.java
new file mode 100644
index 0000000..635f7ee
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/java/org/apache/wayang/hackit/shipper/kafka/sender/SenderKafka.java
@@ -0,0 +1,100 @@
+package org.apache.wayang.hackit.shipper.kafka.sender;
+
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.PSProtocol;
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.sender.Sender;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+public class SenderKafka<K, T> implements Sender<T>, PSProtocol {
+
+    //TODO Get from configuration
+    static Map<String, String> KAFKA_MAPPING;
+    static {
+        KAFKA_MAPPING = new HashMap<>();
+        KAFKA_MAPPING.put("127.0.0.1", "127.0.0.1");
+    }
+    static Integer numPartitions = 1;
+    static Short replicationFactor = 1;
+
+
+    Producer<K, T> producer;
+    Properties config;
+    List<String> topics;
+
+    public SenderKafka(Properties config){
+        this.config = config;
+        this.topics = new ArrayList<>();
+    }
+
+    public PSProtocol preAddTopic(String... topic) {
+        AdminClient ad = AdminClient.create(config);
+        List<String> topicsToCreate = new ArrayList<>();
+        Collection<NewTopic> l = Arrays.stream(topic)
+                .map(t -> {
+                    topicsToCreate.add(t);
+                    return new NewTopic(t, numPartitions, replicationFactor);
+                })
+                .collect(Collectors.toList());
+
+        try {
+            final CreateTopicsResult result = ad.createTopics(l);
+            result.all().get();
+            this.topics.addAll(topicsToCreate);
+            topicsToCreate.clear();
+        } catch (final Exception e) {
+            throw new RuntimeException("Failed to create topic:" + topic, e);
+        }
+
+        return this;
+    }
+
+    @Override
+    public PSProtocol addTopic(String... topic) {
+
+        this.topics.addAll(Arrays.stream(topic).collect(Collectors.toList()));
+        return this;
+    }
+
+    /*Not used by Kafka*/
+    @Override
+    public PSProtocol addExchange(String exchange) {
+        return null;
+    }
+
+    @Override
+    public void init() {
+        this.producer = new KafkaProducer<>(config);
+    }
+
+    @Override
+    public void send(T value) {
+
+        for (String topic : topics) {
+            producer.send(
+                    new ProducerRecord<>(topic, null, value));
+        }
+
+    }
+
+    public void send(K key, T value) {
+
+        for (String topic : topics) {
+            producer.send(
+                    new ProducerRecord<>(topic, key, value));
+            // System.out.println(value + " sent to " + topic);
+        }
+
+    }
+
+    @Override
+    public void close() {
+        producer.close();
+    }
+}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/test/java/org/apache/wayang/hackit/shipper/kafka/receiver/ReceiverKafkaTest.java b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/test/java/org/apache/wayang/hackit/shipper/kafka/receiver/ReceiverKafkaTest.java
new file mode 100644
index 0000000..84f4a82
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/test/java/org/apache/wayang/hackit/shipper/kafka/receiver/ReceiverKafkaTest.java
@@ -0,0 +1,32 @@
+package org.apache.wayang.hackit.shipper.kafka.receiver;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.wayang.plugin.hackit.core.tuple.HackitTuple;
+import org.junit.jupiter.api.Test;
+
+import java.util.Iterator;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class ReceiverKafkaTest {
+
+    @Test
+    void receiveMessage() {
+
+        final Properties props = new Properties();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+        props.put(ConsumerConfig.GROUP_ID_CONFIG,  "1");
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+
+        ReceiverKafka<String, String> receiver = new ReceiverKafka<>(props);
+        receiver.init();
+        receiver.addTopic("debug");
+        Iterator<HackitTuple<String, String>> results = receiver.getElements();
+        results.forEachRemaining(t -> System.out.println(t.getValue()));
+        receiver.close();
+    }
+}
\ No newline at end of file
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/test/java/org/apache/wayang/hackit/shipper/kafka/sender/SenderKafkaTest.java b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/test/java/org/apache/wayang/hackit/shipper/kafka/sender/SenderKafkaTest.java
new file mode 100644
index 0000000..d1803e4
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/test/java/org/apache/wayang/hackit/shipper/kafka/sender/SenderKafkaTest.java
@@ -0,0 +1,37 @@
+package org.apache.wayang.hackit.shipper.kafka.sender;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.jupiter.api.Test;
+
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class SenderKafkaTest {
+
+    @Test
+    void sendMessage() {
+
+        Properties props = new Properties();
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+        //Set acknowledgements for producer requests.
+        props.put(ProducerConfig.ACKS_CONFIG,  "1");
+        //If the request fails, the producer can automatically retry,
+        props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 12);
+        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
+        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1);
+        props.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 1);
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+
+        SenderKafka<String, String> sender = new SenderKafka<>(props);
+        sender.init();
+        sender.addTopic("debug");
+        sender.send("perro", "gato");
+        sender.send("pulpo", "atun");
+        sender.close();
+    }
+
+}
\ No newline at end of file