diff --git a/pom.xml b/pom.xml
index b76b1b3..401d97c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -165,7 +165,7 @@
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <surefire.redirectTestOutputToFile>true</surefire.redirectTestOutputToFile>
-        <slf4j.version>1.7.5</slf4j.version>
+        <slf4j.version>1.7.30</slf4j.version>
         <logback.version>1.0.9</logback.version>
         <guava.version>13.0.1</guava.version>
         <gson.version>2.2.4</gson.version>
@@ -174,7 +174,7 @@
         <snappy-java.version>1.0.5</snappy-java.version>
         <jcl-over-slf4j.version>1.7.2</jcl-over-slf4j.version>
         <asm.version>7.1</asm.version>
-        <kafka.version>0.8.0</kafka.version>
+        <kafka.version>0.8.2.2</kafka.version>
         <zkclient.version>0.10</zkclient.version>
         <zookeeper.version>3.4.5</zookeeper.version>
         <junit.version>4.11</junit.version>
@@ -754,8 +754,17 @@
                         <groupId>com.sun.jmx</groupId>
                         <artifactId>jmxri</artifactId>
                     </exclusion>
+                    <exclusion>
+                        <groupId>net.jpountz.lz4</groupId>
+                        <artifactId>lz4</artifactId>
+                    </exclusion>
                 </exclusions>
             </dependency>
+            <dependency>
+                <groupId>org.slf4j</groupId>
+                <artifactId>log4j-over-slf4j</artifactId>
+                <version>${slf4j.version}</version>
+            </dependency>
 	    <dependency>
 	      <groupId>net.sf.jopt-simple</groupId>
 	      <artifactId>jopt-simple</artifactId>
diff --git a/twill-core/pom.xml b/twill-core/pom.xml
index 249c061..603e809 100644
--- a/twill-core/pom.xml
+++ b/twill-core/pom.xml
@@ -89,6 +89,10 @@
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka_2.10</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>log4j-over-slf4j</artifactId>
+        </dependency>
 	<!-- https://mvnrepository.com/artifact/net.sf.jopt-simple/jopt-simple -->
 	<dependency>
 	  <groupId>net.sf.jopt-simple</groupId>
diff --git a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/IntegerPartitioner.java b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/IntegerPartitioner.java
deleted file mode 100644
index 4aa7940..0000000
--- a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/IntegerPartitioner.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * 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.twill.internal.kafka.client;
-
-import kafka.producer.Partitioner;
-import kafka.utils.VerifiableProperties;
-
-/**
- * A kafka {@link kafka.producer.Partitioner} using integer key to compute partition id.
- */
-public final class IntegerPartitioner implements Partitioner<Integer> {
-
-  public IntegerPartitioner(VerifiableProperties properties) {
-  }
-
-  public int partition(Integer key, int numPartitions) {
-    return key % numPartitions;
-  }
-}
diff --git a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaPublisher.java b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaPublisher.java
index e5d0f8d..832faa8 100644
--- a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaPublisher.java
+++ b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaPublisher.java
@@ -47,6 +47,7 @@
 final class SimpleKafkaPublisher implements KafkaPublisher {
 
   private static final Logger LOG = LoggerFactory.getLogger(SimpleKafkaPublisher.class);
+  private static final int MAX_MESSAGE_BYTES = 1024 * 1024 * 10;
 
   private final BrokerService brokerService;
   private final Ack ack;
@@ -171,9 +172,9 @@
         props.put("metadata.broker.list", newBrokerList);
         props.put("serializer.class", ByteBufferEncoder.class.getName());
         props.put("key.serializer.class", IntegerEncoder.class.getName());
-        props.put("partitioner.class", IntegerPartitioner.class.getName());
         props.put("request.required.acks", Integer.toString(ack.getAck()));
         props.put("compression.codec", compression.getCodec());
+        props.put("message.max.bytes", Integer.toString(MAX_MESSAGE_BYTES));
 
         ProducerConfig config = new ProducerConfig(props);
         newProducer = new Producer<>(config);
diff --git a/twill-core/src/test/java/org/apache/twill/kafka/client/KafkaTest.java b/twill-core/src/test/java/org/apache/twill/kafka/client/KafkaTest.java
index 958925c..e7e7e7c 100644
--- a/twill-core/src/test/java/org/apache/twill/kafka/client/KafkaTest.java
+++ b/twill-core/src/test/java/org/apache/twill/kafka/client/KafkaTest.java
@@ -135,6 +135,9 @@
           server = new EmbeddedKafkaServer(kafkaServerConfig);
           server.startAndWait();
 
+          // Wait a little while to make sure changes is reflected in broker service
+          TimeUnit.SECONDS.sleep(3);
+
           // Publish another message
           createPublishThread(kafkaClient, topic, Compression.NONE, "Second message", 1).start();
 
@@ -308,16 +311,14 @@
 
   private Thread createPublishThread(final KafkaClient kafkaClient, final String topic, final Compression compression,
                                      final String message, final int count, final int base) {
-    return new Thread() {
-      public void run() {
-        KafkaPublisher publisher = kafkaClient.getPublisher(KafkaPublisher.Ack.ALL_RECEIVED, compression);
-        KafkaPublisher.Preparer preparer = publisher.prepare(topic);
-        for (int i = 0; i < count; i++) {
-          preparer.add(Charsets.UTF_8.encode((base + i) + " " + message), 0);
-        }
-        Futures.getUnchecked(preparer.send());
+    return new Thread(() -> {
+      KafkaPublisher publisher = kafkaClient.getPublisher(KafkaPublisher.Ack.ALL_RECEIVED, compression);
+      KafkaPublisher.Preparer preparer = publisher.prepare(topic);
+      for (int i = 0; i < count; i++) {
+        preparer.add(Charsets.UTF_8.encode((base + i) + " " + message), 0);
       }
-    };
+      Futures.getUnchecked(preparer.send());
+    });
   }
 
 
diff --git a/twill-yarn/src/main/java/org/apache/twill/internal/ServiceMain.java b/twill-yarn/src/main/java/org/apache/twill/internal/ServiceMain.java
index b27dcd8..ca0bc08 100644
--- a/twill-yarn/src/main/java/org/apache/twill/internal/ServiceMain.java
+++ b/twill-yarn/src/main/java/org/apache/twill/internal/ServiceMain.java
@@ -18,8 +18,11 @@
 package org.apache.twill.internal;
 
 import ch.qos.logback.classic.LoggerContext;
+import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.classic.util.ContextInitializer;
+import ch.qos.logback.core.filter.Filter;
 import ch.qos.logback.core.joran.spi.JoranException;
+import ch.qos.logback.core.spi.FilterReply;
 import com.google.common.base.Throwables;
 import com.google.common.util.concurrent.AbstractIdleService;
 import com.google.common.util.concurrent.Futures;
@@ -211,7 +214,14 @@
       }
     }
 
+    KafkaAppender kafkaAppender = getKafkaAppender(context);
+    kafkaAppender.start();
+
     // Attach the KafkaAppender to the root logger
+    context.getLogger(ch.qos.logback.classic.Logger.ROOT_LOGGER_NAME).addAppender(kafkaAppender);
+  }
+
+  private KafkaAppender getKafkaAppender(LoggerContext context) {
     KafkaAppender kafkaAppender = new KafkaAppender();
     kafkaAppender.setName("KAFKA");
     kafkaAppender.setTopic(Constants.LOG_TOPIC);
@@ -223,10 +233,15 @@
       kafkaAppender.setRunnableName(runnableName);
     }
 
+    kafkaAppender.addFilter(new Filter<ILoggingEvent>() {
+      @Override
+      public FilterReply decide(ILoggingEvent event) {
+        return event.getLoggerName().startsWith("kafka.") ? FilterReply.DENY : FilterReply.ACCEPT;
+      }
+    });
     kafkaAppender.setContext(context);
-    kafkaAppender.start();
 
-    context.getLogger(ch.qos.logback.classic.Logger.ROOT_LOGGER_NAME).addAppender(kafkaAppender);
+    return kafkaAppender;
   }
 
   /**