diff --git a/contrib/bolts/kafka/src/java/BUILD b/contrib/bolts/kafka/src/java/BUILD
new file mode 100644
index 0000000..1df060c
--- /dev/null
+++ b/contrib/bolts/kafka/src/java/BUILD
@@ -0,0 +1,29 @@
+licenses(["notice"])
+
+package(default_visibility = ["//visibility:public"])
+
+load("//tools/rules:build_defs.bzl", "DOCLINT_HTML_AND_SYNTAX")
+load("//tools/rules:javadoc.bzl", "java_doc")
+
+java_doc(
+    name = "heron-kafka-bolt-javadoc",
+    libs = [":heron-kafka-bolt-java"],
+    pkgs = ["org/apache/heron/bolts/kafka"],
+    title = "Kafka Bolt Documentation",
+)
+
+kafka_bolt_deps = [
+    "//storm-compatibility/src/java:storm-compatibility-java-neverlink",
+    "//heron/api/src/java:api-java-low-level",
+    "//heron/common/src/java:basics-java",
+    "//heron/common/src/java:config-java",
+    "//third_party/java:logging",
+    "@org_apache_kafka_kafka_clients//jar",
+]
+
+java_library(
+    name = "heron-kafka-bolt-java",
+    srcs = glob(["org/apache/heron/bolts/kafka/**/*.java"]),
+    javacopts = DOCLINT_HTML_AND_SYNTAX,
+    deps = kafka_bolt_deps,
+)
\ No newline at end of file
diff --git a/contrib/bolts/kafka/src/java/org/apache/heron/bolts/kafka/DefaultKafkaProducerFactory.java b/contrib/bolts/kafka/src/java/org/apache/heron/bolts/kafka/DefaultKafkaProducerFactory.java
new file mode 100644
index 0000000..3fdd223
--- /dev/null
+++ b/contrib/bolts/kafka/src/java/org/apache/heron/bolts/kafka/DefaultKafkaProducerFactory.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.heron.bolts.kafka;
+
+import java.util.Map;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+
+@SuppressWarnings("unused")
+public class DefaultKafkaProducerFactory<K, V> implements KafkaProducerFactory<K, V> {
+  private static final long serialVersionUID = -2184222924531815883L;
+  private Map<String, Object> configs;
+
+  public DefaultKafkaProducerFactory(Map<String, Object> configs) {
+    this.configs = configs;
+  }
+
+  @Override
+  public Producer<K, V> create() {
+    return new KafkaProducer<>(configs);
+  }
+}
diff --git a/contrib/bolts/kafka/src/java/org/apache/heron/bolts/kafka/KafkaBolt.java b/contrib/bolts/kafka/src/java/org/apache/heron/bolts/kafka/KafkaBolt.java
new file mode 100644
index 0000000..95ce3f8
--- /dev/null
+++ b/contrib/bolts/kafka/src/java/org/apache/heron/bolts/kafka/KafkaBolt.java
@@ -0,0 +1,106 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.heron.bolts.kafka;
+
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.heron.api.Config;
+import org.apache.heron.api.bolt.BaseRichBolt;
+import org.apache.heron.api.bolt.OutputCollector;
+import org.apache.heron.api.topology.OutputFieldsDeclarer;
+import org.apache.heron.api.topology.TopologyContext;
+import org.apache.heron.api.tuple.Tuple;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.KafkaException;
+
+import static org.apache.heron.api.Config.TopologyReliabilityMode.ATMOST_ONCE;
+
+@SuppressWarnings("unused")
+public class KafkaBolt<K, V> extends BaseRichBolt {
+  private static final long serialVersionUID = -3301619269473733618L;
+  private static final Logger LOG = LoggerFactory.getLogger(KafkaBolt.class);
+  private KafkaProducerFactory<K, V> kafkaProducerFactory;
+  private TupleTransformer<K, V> tupleTransformer;
+  private transient Producer<K, V> producer;
+  private Config.TopologyReliabilityMode topologyReliabilityMode;
+  private transient OutputCollector outputCollector;
+
+  @SuppressWarnings("WeakerAccess")
+  public KafkaBolt(KafkaProducerFactory<K, V> kafkaProducerFactory,
+                   TupleTransformer<K, V> tupleTransformer) {
+    this.kafkaProducerFactory = kafkaProducerFactory;
+    this.tupleTransformer = tupleTransformer;
+  }
+
+  @Override
+  public void prepare(Map<String, Object> heronConf, TopologyContext context,
+                      OutputCollector collector) {
+    topologyReliabilityMode = Config.TopologyReliabilityMode
+        .valueOf(heronConf.getOrDefault(Config.TOPOLOGY_RELIABILITY_MODE, ATMOST_ONCE).toString());
+    producer = kafkaProducerFactory.create();
+    outputCollector = collector;
+  }
+
+  @Override
+  public void cleanup() {
+    super.cleanup();
+    if (producer != null) {
+      producer.close();
+    }
+  }
+
+  @Override
+  public void execute(Tuple input) {
+    ProducerRecord<K, V> producerRecord = new ProducerRecord<>(
+        tupleTransformer.getTopicName(input),
+        tupleTransformer.transformToKey(input),
+        tupleTransformer.transformToValue(input));
+    if (topologyReliabilityMode == Config.TopologyReliabilityMode.EFFECTIVELY_ONCE) {
+      try {
+        producer.send(producerRecord).get();
+      } catch (InterruptedException e) {
+        LOG.error("interrupted while waiting for the record to be sent", e);
+        Thread.currentThread().interrupt();
+      } catch (ExecutionException e) {
+        LOG.error("error has occurred when sending record to Kafka", e);
+        throw new KafkaException(e);
+      }
+    } else {
+      producer.send(producerRecord, (recordMetadata, e) -> {
+        if (e != null) {
+          LOG.error("error has occurred when sending record to Kafka", e);
+          if (topologyReliabilityMode == Config.TopologyReliabilityMode.ATLEAST_ONCE) {
+            outputCollector.fail(input);
+          }
+        } else if (topologyReliabilityMode == Config.TopologyReliabilityMode.ATLEAST_ONCE) {
+          outputCollector.ack(input);
+        }
+      });
+    }
+  }
+
+  @Override
+  public void declareOutputFields(OutputFieldsDeclarer declarer) {
+    //Kafka bolt does not emit anything
+  }
+}
diff --git a/contrib/bolts/kafka/src/java/org/apache/heron/bolts/kafka/KafkaProducerFactory.java b/contrib/bolts/kafka/src/java/org/apache/heron/bolts/kafka/KafkaProducerFactory.java
new file mode 100644
index 0000000..6a7ba65
--- /dev/null
+++ b/contrib/bolts/kafka/src/java/org/apache/heron/bolts/kafka/KafkaProducerFactory.java
@@ -0,0 +1,26 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.heron.bolts.kafka;
+
+import java.io.Serializable;
+
+import org.apache.kafka.clients.producer.Producer;
+
+public interface KafkaProducerFactory<K, V> extends Serializable {
+  Producer<K, V> create();
+}
diff --git a/contrib/bolts/kafka/src/java/org/apache/heron/bolts/kafka/TupleTransformer.java b/contrib/bolts/kafka/src/java/org/apache/heron/bolts/kafka/TupleTransformer.java
new file mode 100644
index 0000000..8611aef
--- /dev/null
+++ b/contrib/bolts/kafka/src/java/org/apache/heron/bolts/kafka/TupleTransformer.java
@@ -0,0 +1,43 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.heron.bolts.kafka;
+
+import java.io.Serializable;
+
+import org.apache.heron.api.tuple.Tuple;
+
+public interface TupleTransformer<K, V> extends Serializable {
+
+  default K transformToKey(Tuple tuple) {
+    if (tuple.contains("key")) {
+      //noinspection unchecked
+      return (K) tuple.getValueByField("key");
+    }
+    return null;
+  }
+
+  default V transformToValue(Tuple tuple) {
+    if (tuple.contains("value")) {
+      //noinspection unchecked
+      return (V) tuple.getValueByField("value");
+    }
+    return null;
+  }
+
+  String getTopicName(Tuple tuple);
+}
diff --git a/contrib/bolts/kafka/test/java/BUILD b/contrib/bolts/kafka/test/java/BUILD
new file mode 100644
index 0000000..e713a6c
--- /dev/null
+++ b/contrib/bolts/kafka/test/java/BUILD
@@ -0,0 +1,23 @@
+heron_kafka_bolts_test_dep = [
+    "//contrib/bolts/kafka/src/java:heron-kafka-bolt-java",
+    "//heron/api/src/java:api-java-low-level",
+    "//heron/common/src/java:basics-java",
+    "//heron/common/src/java:config-java",
+    "//third_party/java:junit4",
+    "@org_apache_kafka_kafka_clients//jar",
+    "@org_mockito_mockito_all//jar",
+]
+
+java_test(
+    name = "KafkaBoltTest",
+    srcs = ["org/apache/heron/bolts/kafka/KafkaBoltTest.java"],
+    test_class = "org.apache.heron.bolts.kafka.KafkaBoltTest",
+    deps = heron_kafka_bolts_test_dep,
+)
+
+java_test(
+    name = "DefaultKafkaProducerFactoryTest",
+    srcs = ["org/apache/heron/bolts/kafka/DefaultKafkaProducerFactoryTest.java"],
+    test_class = "org.apache.heron.bolts.kafka.DefaultKafkaProducerFactoryTest",
+    deps = heron_kafka_bolts_test_dep,
+)
diff --git a/contrib/bolts/kafka/test/java/org/apache/heron/bolts/kafka/DefaultKafkaProducerFactoryTest.java b/contrib/bolts/kafka/test/java/org/apache/heron/bolts/kafka/DefaultKafkaProducerFactoryTest.java
new file mode 100644
index 0000000..18a573b
--- /dev/null
+++ b/contrib/bolts/kafka/test/java/org/apache/heron/bolts/kafka/DefaultKafkaProducerFactoryTest.java
@@ -0,0 +1,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.heron.bolts.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+import static org.junit.Assert.assertNotNull;
+
+public class DefaultKafkaProducerFactoryTest {
+
+  @Test
+  public void create() {
+    Map<String, Object> configs = new HashMap<>();
+    configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+    configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+    configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+    assertNotNull(new DefaultKafkaProducerFactory<>(configs).create());
+  }
+}
diff --git a/contrib/bolts/kafka/test/java/org/apache/heron/bolts/kafka/KafkaBoltTest.java b/contrib/bolts/kafka/test/java/org/apache/heron/bolts/kafka/KafkaBoltTest.java
new file mode 100644
index 0000000..9572b11
--- /dev/null
+++ b/contrib/bolts/kafka/test/java/org/apache/heron/bolts/kafka/KafkaBoltTest.java
@@ -0,0 +1,137 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.heron.bolts.kafka;
+
+import java.util.Collections;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.mockito.stubbing.Answer;
+
+import org.apache.heron.api.Config;
+import org.apache.heron.api.bolt.OutputCollector;
+import org.apache.heron.api.tuple.Tuple;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.KafkaException;
+
+import static org.apache.heron.api.Config.TopologyReliabilityMode.ATLEAST_ONCE;
+import static org.apache.heron.api.Config.TopologyReliabilityMode.ATMOST_ONCE;
+import static org.apache.heron.api.Config.TopologyReliabilityMode.EFFECTIVELY_ONCE;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KafkaBoltTest {
+  @Mock
+  private KafkaProducerFactory<String, byte[]> kafkaProducerFactory;
+  @Mock
+  private Producer<String, byte[]> producer;
+  @Mock
+  private TupleTransformer<String, byte[]> tupleTransformer;
+  @Mock
+  private OutputCollector outputCollector;
+  @Mock
+  private Tuple tuple;
+  @Mock
+  private Future<RecordMetadata> future;
+  private KafkaBolt<String, byte[]> kafkaBolt;
+
+  @Before
+  public void setUp() {
+    when(kafkaProducerFactory.create()).thenReturn(producer);
+    kafkaBolt = new KafkaBolt<>(kafkaProducerFactory, tupleTransformer);
+  }
+
+  @Test
+  public void cleanup() {
+    kafkaBolt.prepare(Collections.emptyMap(), null, outputCollector);
+    kafkaBolt.cleanup();
+    verify(producer).close();
+  }
+
+  @Test
+  public void executeATLEASTONCE() {
+    kafkaBolt.prepare(Collections.singletonMap(Config.TOPOLOGY_RELIABILITY_MODE, ATLEAST_ONCE),
+        null, outputCollector);
+    when(tupleTransformer.transformToKey(tuple)).thenReturn("key");
+    byte[] value = new byte[]{1, 2, 3};
+    when(tupleTransformer.transformToValue(tuple)).thenReturn(value);
+    when(tupleTransformer.getTopicName(tuple)).thenReturn("topic");
+
+    ProducerRecord<String, byte[]> producerRecord = new ProducerRecord<>("topic", "key", value);
+    when(producer.send(eq(producerRecord), any(Callback.class)))
+        .then((Answer<Future<RecordMetadata>>) invocationOnMock -> {
+          invocationOnMock.getArgumentAt(1, Callback.class)
+              .onCompletion(new RecordMetadata(null, 0, 0, 0, null, 0, 0), null);
+          return future;
+        });
+    kafkaBolt.execute(tuple);
+    verify(outputCollector).ack(tuple);
+
+    when(producer.send(eq(producerRecord), any(Callback.class)))
+        .then((Answer<Future<RecordMetadata>>) invocationOnMock -> {
+          invocationOnMock.getArgumentAt(1, Callback.class)
+              .onCompletion(new RecordMetadata(null, 0, 0, 0, null, 0, 0), new Exception());
+          return future;
+        });
+    kafkaBolt.execute(tuple);
+    verify(outputCollector).fail(tuple);
+  }
+
+  @Test(expected = KafkaException.class)
+  public void executeEFFECTIVEONCE() throws ExecutionException, InterruptedException {
+    kafkaBolt.prepare(Collections.singletonMap(Config.TOPOLOGY_RELIABILITY_MODE, EFFECTIVELY_ONCE),
+        null, outputCollector);
+    when(tupleTransformer.transformToKey(tuple)).thenReturn("key");
+    byte[] value = new byte[]{1, 2, 3};
+    when(tupleTransformer.transformToValue(tuple)).thenReturn(value);
+    when(tupleTransformer.getTopicName(tuple)).thenReturn("topic");
+
+    ProducerRecord<String, byte[]> producerRecord = new ProducerRecord<>("topic", "key", value);
+    when(producer.send(producerRecord)).thenReturn(future);
+    kafkaBolt.execute(tuple);
+    verify(future).get();
+
+    when(future.get()).thenThrow(ExecutionException.class);
+    kafkaBolt.execute(tuple);
+  }
+
+  @Test
+  public void executeATMOSTONCE() {
+    kafkaBolt.prepare(Collections.singletonMap(Config.TOPOLOGY_RELIABILITY_MODE, ATMOST_ONCE),
+        null, outputCollector);
+    when(tupleTransformer.transformToKey(tuple)).thenReturn("key");
+    byte[] value = new byte[]{1, 2, 3};
+    when(tupleTransformer.transformToValue(tuple)).thenReturn(value);
+    when(tupleTransformer.getTopicName(tuple)).thenReturn("topic");
+
+    ProducerRecord<String, byte[]> producerRecord = new ProducerRecord<>("topic", "key", value);
+    kafkaBolt.execute(tuple);
+    verify(producer).send(eq(producerRecord), any(Callback.class));
+  }
+}