kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jun...@apache.org
Subject [3/3] kafka git commit: kafka-1926; Replace kafka.utils.Utils with o.a.k.common.utils.Utils; patched by Tong Li; reviewed by Jun Rao
Date Mon, 06 Apr 2015 04:46:28 GMT
kafka-1926; Replace kafka.utils.Utils with o.a.k.common.utils.Utils; patched by Tong Li; reviewed by Jun Rao


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/9c23d935
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/9c23d935
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/9c23d935

Branch: refs/heads/trunk
Commit: 9c23d93553a33c5d85231193614d192a9945796e
Parents: 53f3143
Author: Tong Li <litong02@us.ibm.com>
Authored: Sun Apr 5 21:46:11 2015 -0700
Committer: Jun Rao <junrao@gmail.com>
Committed: Sun Apr 5 21:46:11 2015 -0700

----------------------------------------------------------------------
 .../org/apache/kafka/common/utils/Utils.java    | 124 +++-
 .../apache/kafka/common/utils/UtilsTest.java    |  33 +
 core/src/main/scala/kafka/Kafka.scala           |   3 +-
 .../kafka/admin/ConsumerGroupCommand.scala      |   1 +
 .../PreferredReplicaLeaderElectionCommand.scala |   4 +-
 .../kafka/admin/ReassignPartitionsCommand.scala |   9 +-
 .../main/scala/kafka/admin/TopicCommand.scala   |   3 +-
 .../main/scala/kafka/client/ClientUtils.scala   |   4 +-
 core/src/main/scala/kafka/cluster/Broker.scala  |  16 +-
 .../main/scala/kafka/cluster/Partition.scala    |   2 +-
 .../kafka/consumer/ConsumerFetcherManager.scala |   2 +-
 .../scala/kafka/consumer/ConsumerIterator.scala |   2 +-
 .../kafka/consumer/PartitionAssignor.scala      |   4 +-
 .../main/scala/kafka/consumer/TopicCount.scala  |   4 +-
 .../consumer/ZookeeperConsumerConnector.scala   |   2 +-
 .../controller/ControllerChannelManager.scala   |   4 +-
 .../kafka/controller/KafkaController.scala      |   2 +-
 .../controller/PartitionStateMachine.scala      |   3 +-
 .../kafka/controller/ReplicaStateMachine.scala  |   2 +-
 .../kafka/controller/TopicDeletionManager.scala |   2 +-
 .../main/scala/kafka/log/FileMessageSet.scala   |   6 +-
 core/src/main/scala/kafka/log/Log.scala         |   6 +-
 .../scala/kafka/log/LogCleanerManager.scala     |   2 +-
 core/src/main/scala/kafka/log/LogConfig.scala   |   2 +-
 core/src/main/scala/kafka/log/LogManager.scala  |   8 +-
 core/src/main/scala/kafka/log/LogSegment.scala  |   8 +-
 core/src/main/scala/kafka/log/OffsetIndex.scala |   8 +-
 core/src/main/scala/kafka/log/OffsetMap.scala   |   3 +-
 core/src/main/scala/kafka/message/Message.scala |   3 +-
 .../kafka/message/MessageAndMetadata.scala      |   2 +-
 .../scala/kafka/message/MessageWriter.scala     |   2 +-
 .../kafka/metrics/KafkaCSVMetricsReporter.scala |   4 +-
 .../kafka/metrics/KafkaMetricsConfig.scala      |   4 +-
 .../kafka/metrics/KafkaMetricsReporter.scala    |   6 +-
 .../network/BoundedByteBufferReceive.scala      |   4 +-
 .../main/scala/kafka/network/SocketServer.scala |   1 +
 .../kafka/producer/ByteArrayPartitioner.scala   |   1 +
 .../kafka/producer/DefaultPartitioner.scala     |   1 +
 .../main/scala/kafka/producer/Producer.scala    |   6 +-
 .../scala/kafka/producer/ProducerConfig.scala   |   6 +-
 .../producer/async/DefaultEventHandler.scala    |   9 +-
 .../kafka/server/AbstractFetcherManager.scala   |   5 +-
 .../kafka/server/AbstractFetcherThread.scala    |   2 +-
 .../kafka/server/BrokerMetadataCheckpoint.scala |   2 +-
 .../main/scala/kafka/server/KafkaConfig.scala   |  18 +-
 .../kafka/server/KafkaRequestHandler.scala      |   1 +
 .../main/scala/kafka/server/KafkaServer.scala   |  20 +-
 .../main/scala/kafka/server/MetadataCache.scala |   2 +-
 .../main/scala/kafka/server/OffsetManager.scala |   1 +
 .../kafka/server/ZookeeperLeaderElector.scala   |   2 +-
 .../scala/kafka/tools/ConsoleConsumer.scala     |   1 +
 .../scala/kafka/tools/DumpLogSegments.scala     |   5 +-
 .../scala/kafka/tools/KafkaMigrationTool.java   |   2 +-
 .../main/scala/kafka/tools/MirrorMaker.scala    |  14 +-
 .../scala/kafka/tools/SimpleConsumerShell.scala |   1 +
 .../kafka/tools/StateChangeLogMerger.scala      |   4 +-
 .../scala/kafka/tools/TestEndToEndLatency.scala |  92 ---
 .../scala/kafka/tools/TestLogCleaning.scala     | 311 ---------
 .../scala/kafka/tools/UpdateOffsetsInZK.scala   |   4 +-
 core/src/main/scala/kafka/utils/CoreUtils.scala | 347 ++++++++++
 core/src/main/scala/kafka/utils/Crc32.java      | 637 -------------------
 .../main/scala/kafka/utils/KafkaScheduler.scala |   4 +-
 .../scala/kafka/utils/Log4jController.scala     |   2 +-
 core/src/main/scala/kafka/utils/Logging.scala   |  10 +-
 core/src/main/scala/kafka/utils/Utils.scala     | 619 ------------------
 .../kafka/utils/VerifiableProperties.scala      |   2 +-
 .../kafka/api/ProducerCompressionTest.scala     |   4 +-
 .../scala/kafka/tools/TestEndToEndLatency.scala |  91 +++
 .../scala/kafka/tools/TestLogCleaning.scala     | 311 +++++++++
 .../test/scala/other/kafka/DeleteZKPath.scala   |   3 +-
 .../test/scala/other/kafka/StressTestLog.scala  |   2 +-
 .../scala/other/kafka/TestCrcPerformance.scala  |   3 +-
 .../other/kafka/TestLinearWriteSpeed.scala      |   4 +-
 .../unit/kafka/admin/AddPartitionsTest.scala    |   4 +-
 .../test/scala/unit/kafka/admin/AdminTest.scala |   2 +-
 .../kafka/consumer/ConsumerIteratorTest.scala   |   2 +-
 .../integration/KafkaServerTestHarness.scala    |   4 +-
 .../kafka/integration/PrimitiveApiTest.scala    |  12 +-
 .../kafka/integration/RollingBounceTest.scala   |   4 +-
 .../integration/UncleanLeaderElectionTest.scala |   4 +-
 .../unit/kafka/log/BrokerCompressionTest.scala  |   2 +-
 .../test/scala/unit/kafka/log/CleanerTest.scala |   5 +-
 .../kafka/log/LogCleanerIntegrationTest.scala   |   6 +-
 .../scala/unit/kafka/log/LogManagerTest.scala   |   4 +-
 .../src/test/scala/unit/kafka/log/LogTest.scala |   4 +-
 .../kafka/log4j/KafkaLog4jAppenderTest.scala    |   4 +-
 .../scala/unit/kafka/message/MessageTest.scala  |   3 +-
 .../unit/kafka/producer/AsyncProducerTest.scala |   4 +-
 .../unit/kafka/producer/ProducerTest.scala      |   4 +-
 .../unit/kafka/server/AdvertiseBrokerTest.scala |   4 +-
 .../server/HighwatermarkPersistenceTest.scala   |   4 +-
 .../unit/kafka/server/KafkaConfigTest.scala     |  10 +-
 .../unit/kafka/server/LeaderElectionTest.scala  |   4 +-
 .../scala/unit/kafka/server/LogOffsetTest.scala |   2 +-
 .../unit/kafka/server/LogRecoveryTest.scala     |   4 +-
 .../unit/kafka/server/OffsetCommitTest.scala    |   2 +-
 .../server/ServerGenerateBrokerIdTest.scala     |  14 +-
 .../unit/kafka/server/ServerShutdownTest.scala  |  12 +-
 .../unit/kafka/server/ServerStartupTest.scala   |   6 +-
 .../test/scala/unit/kafka/utils/TestUtils.scala |  14 +-
 .../test/scala/unit/kafka/utils/UtilsTest.scala |  33 +-
 .../scala/unit/kafka/zk/EmbeddedZookeeper.scala |  10 +-
 .../unit/kafka/zk/ZooKeeperTestHarness.scala    |   6 +-
 103 files changed, 1154 insertions(+), 1883 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
old mode 100644
new mode 100755
index 39e8d7c..f73eedb
--- a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
@@ -14,15 +14,26 @@ package org.apache.kafka.common.utils;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.File;
+import java.io.FileInputStream;
 import java.io.OutputStream;
 import java.io.UnsupportedEncodingException;
+import java.io.FileNotFoundException;
+import java.io.StringWriter;
+import java.io.PrintWriter;
 import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.Properties;
+import java.nio.channels.FileChannel;
+import java.nio.charset.Charset;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.apache.kafka.common.KafkaException;
 
 public class Utils {
@@ -33,6 +44,8 @@ public class Utils {
 
     public static final String NL = System.getProperty("line.separator");
 
+    private static final Logger log = LoggerFactory.getLogger(Utils.class);
+
     /**
      * Turn the given UTF8 byte array into a string
      * 
@@ -330,7 +343,7 @@ public class Utils {
                 ? "[" + host + "]:" + port // IPv6
                 : host + ":" + port;
     }
-    
+
     /**
      * Create a string representation of an array joined by the given separator
      * @param strs The array of items
@@ -357,4 +370,113 @@ public class Utils {
         }
         return sb.toString();
     }
+
+    /**
+     * Read a properties file from the given path
+     * @param filename The path of the file to read
+     */
+    public static Properties loadProps(String filename) throws IOException, FileNotFoundException {
+        Properties props = new Properties();
+        InputStream propStream = null;
+        try {
+            propStream = new FileInputStream(filename);
+            props.load(propStream);
+        } finally {
+            if (propStream != null)
+                propStream.close();
+        }
+        return props;
+    }
+
+    /**
+     * Get the stack trace from an exception as a string
+     */
+    public static String stackTrace(Throwable e) {
+        StringWriter sw = new StringWriter();
+        PrintWriter pw = new PrintWriter(sw);
+        e.printStackTrace(pw);
+        return sw.toString();
+    }
+
+    /**
+     * Create a new thread
+     * @param name The name of the thread
+     * @param runnable The work for the thread to do
+     * @param daemon Should the thread block JVM shutdown?
+     * @return The unstarted thread
+     */
+    public static Thread newThread(String name, Runnable runnable, Boolean daemon) {
+        Thread thread = new Thread(runnable, name);
+        thread.setDaemon(daemon);
+        thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
+            public void uncaughtException(Thread t, Throwable e) {
+                log.error("Uncaught exception in thread '" + t.getName() + "':", e);
+            }
+        });
+        return thread;
+    }
+
+    /**
+     * Create a daemon thread
+     * @param name The name of the thread
+     * @param runnable The runnable to execute in the background
+     * @return The unstarted thread
+     */
+    public static Thread daemonThread(String name, Runnable runnable) {
+        return newThread(name, runnable, true);
+    }
+
+    /**
+     * Print an error message and shutdown the JVM
+     * @param message The error message
+     */
+    public static void croak(String message) {
+        System.err.println(message);
+        System.exit(1);
+    }
+
+    /**
+     * Read a buffer into a Byte array for the given offset and length
+     */
+    public static byte[] readBytes(ByteBuffer buffer, int offset, int length) {
+        byte[] dest = new byte[length];
+        if (buffer.hasArray()) {
+            System.arraycopy(buffer.array(), buffer.arrayOffset() + offset, dest, 0, length);
+        } else {
+            buffer.mark();
+            buffer.position(offset);
+            buffer.get(dest, 0, length);
+            buffer.reset();
+        }
+        return dest;
+    }
+
+    /**
+     * Read the given byte buffer into a Byte array
+     */
+    public static byte[] readBytes(ByteBuffer buffer) {
+        return Utils.readBytes(buffer, 0, buffer.limit());
+    }
+
+    /**
+     * Attempt to read a file as a string
+     * @throws IOException 
+     */
+    public static String readFileAsString(String path, Charset charset) throws IOException {
+        if (charset == null) charset = Charset.defaultCharset();
+        FileInputStream stream = new FileInputStream(new File(path));
+        String result = new String();
+        try {
+            FileChannel fc = stream.getChannel();
+            MappedByteBuffer bb = fc.map(FileChannel.MapMode.READ_ONLY, 0, fc.size());
+            result = charset.decode(bb).toString();
+        } finally {
+            stream.close();
+        }
+        return result;
+    }
+
+    public static String readFileAsString(String path) throws IOException {
+        return Utils.readFileAsString(path, Charset.defaultCharset());
+    }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java b/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
old mode 100644
new mode 100755
index 4b706d7..2ebe3c2
--- a/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
@@ -18,6 +18,7 @@ package org.apache.kafka.common.utils;
 
 import java.util.Arrays;
 import java.util.Collections;
+import java.nio.ByteBuffer;
 
 import org.junit.Test;
 
@@ -67,4 +68,36 @@ public class UtilsTest {
         assertEquals(0, Utils.abs(0));
         assertEquals(1, Utils.abs(-1));
     }
+
+    private void subTest(ByteBuffer buffer) {
+        // The first byte should be 'A'
+        assertEquals('A', (Utils.readBytes(buffer, 0, 1))[0]);
+
+        // The offset is 2, so the first 2 bytes should be skipped.
+        byte[] results = Utils.readBytes(buffer, 2, 3);
+        assertEquals('y', results[0]);
+        assertEquals(' ', results[1]);
+        assertEquals('S', results[2]);
+        assertEquals(3, results.length);
+
+        // test readBytes without offset and length specified.
+        results = Utils.readBytes(buffer);
+        assertEquals('A', results[0]);
+        assertEquals('t', results[buffer.limit() - 1]);
+        assertEquals(buffer.limit(), results.length);
+    }
+
+    @Test
+    public void testReadBytes() {
+        byte[] myvar = "Any String you want".getBytes();
+        ByteBuffer buffer = ByteBuffer.allocate(myvar.length);
+        buffer.put(myvar);
+        buffer.rewind();
+
+        this.subTest(buffer);
+
+        // test readonly buffer, different path
+        buffer = ByteBuffer.wrap(myvar).asReadOnlyBuffer();
+        this.subTest(buffer);
+    }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/Kafka.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/Kafka.scala b/core/src/main/scala/kafka/Kafka.scala
old mode 100644
new mode 100755
index 37de7df..fb860e7
--- a/core/src/main/scala/kafka/Kafka.scala
+++ b/core/src/main/scala/kafka/Kafka.scala
@@ -21,7 +21,8 @@ import scala.collection.JavaConversions._
 import joptsimple.OptionParser
 import metrics.KafkaMetricsReporter
 import server.{KafkaConfig, KafkaServerStartable, KafkaServer}
-import kafka.utils.{VerifiableProperties, CommandLineUtils, Utils, Logging}
+import kafka.utils.{VerifiableProperties, CommandLineUtils, Logging}
+import org.apache.kafka.common.utils.Utils
 
 object Kafka extends Logging {
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
old mode 100644
new mode 100755
index 89fa29a..1c3b380
--- a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
@@ -30,6 +30,7 @@ import joptsimple.{OptionSpec, OptionParser}
 import scala.collection.{Set, mutable}
 import kafka.consumer.SimpleConsumer
 import collection.JavaConversions._
+import org.apache.kafka.common.utils.Utils
 
 
 object ConsumerGroupCommand {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala b/core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala
old mode 100644
new mode 100755
index 79b5e0a..3b3cd67
--- a/core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala
+++ b/core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala
@@ -22,7 +22,7 @@ import org.I0Itec.zkclient.ZkClient
 import org.I0Itec.zkclient.exception.ZkNodeExistsException
 import kafka.common.{TopicAndPartition, AdminCommandFailedException}
 import collection._
-import mutable.ListBuffer
+import org.apache.kafka.common.utils.Utils
 
 object PreferredReplicaLeaderElectionCommand extends Logging {
 
@@ -84,7 +84,7 @@ object PreferredReplicaLeaderElectionCommand extends Logging {
               val partition = p.get("partition").get.asInstanceOf[Int]
               TopicAndPartition(topic, partition)
             }
-            val duplicatePartitions = Utils.duplicates(partitions)
+            val duplicatePartitions = CoreUtils.duplicates(partitions)
             val partitionsSet = partitions.toSet
             if (duplicatePartitions.nonEmpty)
               throw new AdminOperationException("Preferred replica election data contains duplicate partitions: %s".format(duplicatePartitions.mkString(",")))

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala b/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
old mode 100644
new mode 100755
index 979992b..bbe3362
--- a/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
+++ b/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
@@ -22,6 +22,7 @@ import collection._
 import org.I0Itec.zkclient.ZkClient
 import org.I0Itec.zkclient.exception.ZkNodeExistsException
 import kafka.common.{TopicAndPartition, AdminCommandFailedException}
+import org.apache.kafka.common.utils.Utils
 
 object ReassignPartitionsCommand extends Logging {
 
@@ -81,12 +82,12 @@ object ReassignPartitionsCommand extends Logging {
       CommandLineUtils.printUsageAndDie(opts.parser, "If --generate option is used, command must include both --topics-to-move-json-file and --broker-list options")
     val topicsToMoveJsonFile = opts.options.valueOf(opts.topicsToMoveJsonFileOpt)
     val brokerListToReassign = opts.options.valueOf(opts.brokerListOpt).split(',').map(_.toInt)
-    val duplicateReassignments = Utils.duplicates(brokerListToReassign)
+    val duplicateReassignments = CoreUtils.duplicates(brokerListToReassign)
     if (duplicateReassignments.nonEmpty)
       throw new AdminCommandFailedException("Broker list contains duplicate entries: %s".format(duplicateReassignments.mkString(",")))
     val topicsToMoveJsonString = Utils.readFileAsString(topicsToMoveJsonFile)
     val topicsToReassign = ZkUtils.parseTopicsData(topicsToMoveJsonString)
-    val duplicateTopicsToReassign = Utils.duplicates(topicsToReassign)
+    val duplicateTopicsToReassign = CoreUtils.duplicates(topicsToReassign)
     if (duplicateTopicsToReassign.nonEmpty)
       throw new AdminCommandFailedException("List of topics to reassign contains duplicate entries: %s".format(duplicateTopicsToReassign.mkString(",")))
     val topicPartitionsToReassign = ZkUtils.getReplicaAssignmentForTopics(zkClient, topicsToReassign)
@@ -112,11 +113,11 @@ object ReassignPartitionsCommand extends Logging {
     val partitionsToBeReassigned = ZkUtils.parsePartitionReassignmentDataWithoutDedup(reassignmentJsonString)
     if (partitionsToBeReassigned.isEmpty)
       throw new AdminCommandFailedException("Partition reassignment data file %s is empty".format(reassignmentJsonFile))
-    val duplicateReassignedPartitions = Utils.duplicates(partitionsToBeReassigned.map{ case(tp,replicas) => tp})
+    val duplicateReassignedPartitions = CoreUtils.duplicates(partitionsToBeReassigned.map{ case(tp,replicas) => tp})
     if (duplicateReassignedPartitions.nonEmpty)
       throw new AdminCommandFailedException("Partition reassignment contains duplicate topic partitions: %s".format(duplicateReassignedPartitions.mkString(",")))
     val duplicateEntries= partitionsToBeReassigned
-      .map{ case(tp,replicas) => (tp, Utils.duplicates(replicas))}
+      .map{ case(tp,replicas) => (tp, CoreUtils.duplicates(replicas))}
       .filter{ case (tp,duplicatedReplicas) => duplicatedReplicas.nonEmpty }
     if (duplicateEntries.nonEmpty) {
       val duplicatesMsg = duplicateEntries

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/admin/TopicCommand.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala
old mode 100644
new mode 100755
index e36a9d1..60f0228
--- a/core/src/main/scala/kafka/admin/TopicCommand.scala
+++ b/core/src/main/scala/kafka/admin/TopicCommand.scala
@@ -28,6 +28,7 @@ import scala.collection.JavaConversions._
 import kafka.log.LogConfig
 import kafka.consumer.Whitelist
 import kafka.server.OffsetManager
+import org.apache.kafka.common.utils.Utils
 
 
 object TopicCommand {
@@ -228,7 +229,7 @@ object TopicCommand {
     val ret = new mutable.HashMap[Int, List[Int]]()
     for (i <- 0 until partitionList.size) {
       val brokerList = partitionList(i).split(":").map(s => s.trim().toInt)
-      val duplicateBrokers = Utils.duplicates(brokerList)
+      val duplicateBrokers = CoreUtils.duplicates(brokerList)
       if (duplicateBrokers.nonEmpty)
         throw new AdminCommandFailedException("Partition replica lists may not contain duplicate entries: %s".format(duplicateBrokers.mkString(",")))
       ret.put(i, brokerList.toList)

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/client/ClientUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/client/ClientUtils.scala b/core/src/main/scala/kafka/client/ClientUtils.scala
old mode 100644
new mode 100755
index ad4c9d2..f08aaf2
--- a/core/src/main/scala/kafka/client/ClientUtils.scala
+++ b/core/src/main/scala/kafka/client/ClientUtils.scala
@@ -23,7 +23,7 @@ import kafka.cluster._
 import kafka.api._
 import kafka.producer._
 import kafka.common.{ErrorMapping, KafkaException}
-import kafka.utils.{Utils, Logging}
+import kafka.utils.{CoreUtils, Logging}
 import java.util.Properties
 import util.Random
 import kafka.network.BlockingChannel
@@ -98,7 +98,7 @@ object ClientUtils extends Logging{
    * Parse a list of broker urls in the form host1:port1, host2:port2, ... 
    */
   def parseBrokerList(brokerListStr: String): Seq[BrokerEndpoint] = {
-    val brokersStr = Utils.parseCsvList(brokerListStr)
+    val brokersStr = CoreUtils.parseCsvList(brokerListStr)
 
     brokersStr.zipWithIndex.map { case (address, brokerId) =>
       BrokerEndpoint.createBrokerEndPoint(brokerId, address)

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/cluster/Broker.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/cluster/Broker.scala b/core/src/main/scala/kafka/cluster/Broker.scala
old mode 100644
new mode 100755
index 3933bb3..8e603b6
--- a/core/src/main/scala/kafka/cluster/Broker.scala
+++ b/core/src/main/scala/kafka/cluster/Broker.scala
@@ -17,11 +17,13 @@
 
 package kafka.cluster
 
+import kafka.utils.CoreUtils._
+import kafka.utils.Json
+import kafka.api.ApiUtils._
 import java.nio.ByteBuffer
 
 import kafka.common.{BrokerEndPointNotAvailableException, BrokerNotAvailableException, KafkaException}
 import kafka.utils.Json
-import kafka.utils.Utils._
 import org.apache.kafka.common.protocol.SecurityProtocol
 
 /**
@@ -139,14 +141,4 @@ case class Broker(id: Int, endPoints: Map[SecurityProtocol, EndPoint]) {
     }
   }
 
-  override def equals(obj: Any): Boolean = {
-    obj match {
-      case null => false
-      // Yes, Scala compares lists element by element
-      case n: Broker => id == n.id && endPoints == n.endPoints
-      case _ => false
-    }
-  }
-
-  override def hashCode(): Int = hashcode(id, endPoints)
-}
\ No newline at end of file
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/cluster/Partition.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala
old mode 100644
new mode 100755
index 6d142d6..3fb549c
--- a/core/src/main/scala/kafka/cluster/Partition.scala
+++ b/core/src/main/scala/kafka/cluster/Partition.scala
@@ -18,7 +18,7 @@ package kafka.cluster
 
 import kafka.common._
 import kafka.utils._
-import kafka.utils.Utils.{inReadLock,inWriteLock}
+import kafka.utils.CoreUtils.{inReadLock,inWriteLock}
 import kafka.admin.AdminUtils
 import kafka.api.{PartitionStateInfo, LeaderAndIsr}
 import kafka.log.LogConfig

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala b/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala
old mode 100644
new mode 100755
index 484a57f..6bb0d56
--- a/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala
+++ b/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala
@@ -25,7 +25,7 @@ import scala.collection.immutable
 import collection.mutable.HashMap
 import scala.collection.mutable
 import java.util.concurrent.locks.ReentrantLock
-import kafka.utils.Utils.inLock
+import kafka.utils.CoreUtils.inLock
 import kafka.utils.ZkUtils._
 import kafka.utils.{ShutdownableThread, SystemTime}
 import kafka.common.TopicAndPartition

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/consumer/ConsumerIterator.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/consumer/ConsumerIterator.scala b/core/src/main/scala/kafka/consumer/ConsumerIterator.scala
old mode 100644
new mode 100755
index b00a4dc..0c5c451
--- a/core/src/main/scala/kafka/consumer/ConsumerIterator.scala
+++ b/core/src/main/scala/kafka/consumer/ConsumerIterator.scala
@@ -17,7 +17,7 @@
 
 package kafka.consumer
 
-import kafka.utils.{IteratorTemplate, Logging, Utils}
+import kafka.utils.{IteratorTemplate, Logging, CoreUtils}
 import java.util.concurrent.{TimeUnit, BlockingQueue}
 import kafka.serializer.Decoder
 import java.util.concurrent.atomic.AtomicReference

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/consumer/PartitionAssignor.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/consumer/PartitionAssignor.scala b/core/src/main/scala/kafka/consumer/PartitionAssignor.scala
old mode 100644
new mode 100755
index bc2e5b4..4afda8b
--- a/core/src/main/scala/kafka/consumer/PartitionAssignor.scala
+++ b/core/src/main/scala/kafka/consumer/PartitionAssignor.scala
@@ -19,7 +19,7 @@ package kafka.consumer
 
 import org.I0Itec.zkclient.ZkClient
 import kafka.common.TopicAndPartition
-import kafka.utils.{Pool, Utils, ZkUtils, Logging}
+import kafka.utils.{Pool, CoreUtils, ZkUtils, Logging}
 
 import scala.collection.mutable
 
@@ -88,7 +88,7 @@ class RoundRobinAssignor() extends PartitionAssignor with Logging {
             "Topic %s has the following available consumer streams: %s\n".format(headTopic, headThreadIdSet))
       }
 
-      val threadAssignor = Utils.circularIterator(headThreadIdSet.toSeq.sorted)
+      val threadAssignor = CoreUtils.circularIterator(headThreadIdSet.toSeq.sorted)
 
       info("Starting round-robin assignment with consumers " + ctx.consumers)
       val allTopicPartitions = ctx.partitionsForTopic.flatMap { case (topic, partitions) =>

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/consumer/TopicCount.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/consumer/TopicCount.scala b/core/src/main/scala/kafka/consumer/TopicCount.scala
old mode 100644
new mode 100755
index 0954b3c..6994c8e
--- a/core/src/main/scala/kafka/consumer/TopicCount.scala
+++ b/core/src/main/scala/kafka/consumer/TopicCount.scala
@@ -19,7 +19,7 @@ package kafka.consumer
 
 import scala.collection._
 import org.I0Itec.zkclient.ZkClient
-import kafka.utils.{Json, ZKGroupDirs, ZkUtils, Logging, Utils}
+import kafka.utils.{Json, ZKGroupDirs, ZkUtils, Logging, CoreUtils}
 import kafka.common.KafkaException
 
 private[kafka] trait TopicCount {
@@ -136,7 +136,7 @@ private[kafka] class WildcardTopicCount(zkClient: ZkClient,
     TopicCount.makeConsumerThreadIdsPerTopic(consumerIdString, Map(wildcardTopics.map((_, numStreams)): _*))
   }
 
-  def getTopicCountMap = Map(Utils.JSONEscapeString(topicFilter.regex) -> numStreams)
+  def getTopicCountMap = Map(CoreUtils.JSONEscapeString(topicFilter.regex) -> numStreams)
 
   def pattern: String = {
     topicFilter match {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala b/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
old mode 100644
new mode 100755
index 3e8a75b..e250b94
--- a/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
+++ b/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
@@ -32,7 +32,7 @@ import kafka.javaapi.consumer.ConsumerRebalanceListener
 import kafka.metrics._
 import kafka.network.BlockingChannel
 import kafka.serializer._
-import kafka.utils.Utils.inLock
+import kafka.utils.CoreUtils.inLock
 import kafka.utils.ZkUtils._
 import kafka.utils._
 import org.I0Itec.zkclient.exception.ZkNodeExistsException

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
old mode 100644
new mode 100755
index fb596ba..97acdb2
--- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
+++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
@@ -17,7 +17,7 @@
 package kafka.controller
 
 import kafka.network.{Receive, BlockingChannel}
-import kafka.utils.{Utils, Logging, ShutdownableThread}
+import kafka.utils.{CoreUtils, Logging, ShutdownableThread}
 import collection.mutable.HashMap
 import kafka.cluster.Broker
 import java.util.concurrent.{LinkedBlockingQueue, BlockingQueue}
@@ -140,7 +140,7 @@ class RequestSendThread(val controllerId: Int,
               connectToBroker(toBroker, channel)
               isSendSuccessful = false
               // backoff before retrying the connection and send
-              Utils.swallowTrace(Thread.sleep(300))
+              CoreUtils.swallowTrace(Thread.sleep(300))
           }
         }
         if (receive != null) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala
old mode 100644
new mode 100755
index f2e62f5..3a09377
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -29,7 +29,7 @@ import kafka.log.LogConfig
 import kafka.metrics.{KafkaTimer, KafkaMetricsGroup}
 import kafka.utils.ZkUtils._
 import kafka.utils._
-import kafka.utils.Utils._
+import kafka.utils.CoreUtils._
 import org.apache.zookeeper.Watcher.Event.KeeperState
 import org.I0Itec.zkclient.{IZkDataListener, IZkStateListener, ZkClient}
 import org.I0Itec.zkclient.exception.{ZkNodeExistsException, ZkNoNodeException}

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/controller/PartitionStateMachine.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/PartitionStateMachine.scala b/core/src/main/scala/kafka/controller/PartitionStateMachine.scala
old mode 100644
new mode 100755
index 2f0694b..92fd92d
--- a/core/src/main/scala/kafka/controller/PartitionStateMachine.scala
+++ b/core/src/main/scala/kafka/controller/PartitionStateMachine.scala
@@ -25,9 +25,8 @@ import kafka.common.{LeaderElectionNotNeededException, TopicAndPartition, StateC
 import kafka.utils.{Logging, ZkUtils, ReplicationUtils}
 import org.I0Itec.zkclient.{IZkDataListener, IZkChildListener}
 import org.I0Itec.zkclient.exception.ZkNodeExistsException
-import org.apache.log4j.Logger
 import kafka.controller.Callbacks.CallbackBuilder
-import kafka.utils.Utils._
+import kafka.utils.CoreUtils._
 
 /**
  * This class represents the state machine for partitions. It defines the states that a partition can be in, and

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/controller/ReplicaStateMachine.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/ReplicaStateMachine.scala b/core/src/main/scala/kafka/controller/ReplicaStateMachine.scala
old mode 100644
new mode 100755
index 3e87e1d..e5c56e0
--- a/core/src/main/scala/kafka/controller/ReplicaStateMachine.scala
+++ b/core/src/main/scala/kafka/controller/ReplicaStateMachine.scala
@@ -24,7 +24,7 @@ import kafka.utils.{ZkUtils, ReplicationUtils, Logging}
 import org.I0Itec.zkclient.IZkChildListener
 import org.apache.log4j.Logger
 import kafka.controller.Callbacks._
-import kafka.utils.Utils._
+import kafka.utils.CoreUtils._
 
 /**
  * This class represents the state machine for replicas. It defines the states that a replica can be in, and

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/TopicDeletionManager.scala b/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
old mode 100644
new mode 100755
index e56f22d..64ecb49
--- a/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
+++ b/core/src/main/scala/kafka/controller/TopicDeletionManager.scala
@@ -18,7 +18,7 @@ package kafka.controller
 
 import collection.mutable
 import kafka.utils.{ShutdownableThread, Logging, ZkUtils}
-import kafka.utils.Utils._
+import kafka.utils.CoreUtils._
 import collection.Set
 import kafka.common.{ErrorMapping, TopicAndPartition}
 import kafka.api.{StopReplicaResponse, RequestOrResponse}

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/log/FileMessageSet.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/FileMessageSet.scala b/core/src/main/scala/kafka/log/FileMessageSet.scala
old mode 100644
new mode 100755
index b2652dd..2522604
--- a/core/src/main/scala/kafka/log/FileMessageSet.scala
+++ b/core/src/main/scala/kafka/log/FileMessageSet.scala
@@ -66,12 +66,12 @@ class FileMessageSet private[kafka](@volatile var file: File,
    * Create a file message set with no slicing
    */
   def this(file: File) = 
-    this(file, Utils.openChannel(file, mutable = true))
+    this(file, CoreUtils.openChannel(file, mutable = true))
 
   /**
    * Create a file message set with mutable option
    */
-  def this(file: File, mutable: Boolean) = this(file, Utils.openChannel(file, mutable))
+  def this(file: File, mutable: Boolean) = this(file, CoreUtils.openChannel(file, mutable))
   
   /**
    * Create a slice view of the file message set that begins and ends at the given byte offsets
@@ -231,7 +231,7 @@ class FileMessageSet private[kafka](@volatile var file: File,
    * @return True iff this message set was deleted.
    */
   def delete(): Boolean = {
-    Utils.swallow(channel.close())
+    CoreUtils.swallow(channel.close())
     file.delete()
   }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/log/Log.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala
old mode 100644
new mode 100755
index a0745be..5563f2d
--- a/core/src/main/scala/kafka/log/Log.scala
+++ b/core/src/main/scala/kafka/log/Log.scala
@@ -136,12 +136,12 @@ class Log(val dir: File,
         // we crashed in the middle of a swap operation, to recover:
         // if a log, swap it in and delete the .index file
         // if an index just delete it, it will be rebuilt
-        val baseName = new File(Utils.replaceSuffix(file.getPath, SwapFileSuffix, ""))
+        val baseName = new File(CoreUtils.replaceSuffix(file.getPath, SwapFileSuffix, ""))
         if(baseName.getPath.endsWith(IndexFileSuffix)) {
           file.delete()
         } else if(baseName.getPath.endsWith(LogFileSuffix)){
           // delete the index
-          val index = new File(Utils.replaceSuffix(baseName.getPath, LogFileSuffix, IndexFileSuffix))
+          val index = new File(CoreUtils.replaceSuffix(baseName.getPath, LogFileSuffix, IndexFileSuffix))
           index.delete()
           // complete the swap operation
           val renamed = file.renameTo(baseName)
@@ -627,7 +627,7 @@ class Log(val dir: File,
       removeLogMetrics()
       logSegments.foreach(_.delete())
       segments.clear()
-      Utils.rm(dir)
+      CoreUtils.rm(dir)
     }
   }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/log/LogCleanerManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/LogCleanerManager.scala b/core/src/main/scala/kafka/log/LogCleanerManager.scala
old mode 100644
new mode 100755
index 351824b..f6795d3
--- a/core/src/main/scala/kafka/log/LogCleanerManager.scala
+++ b/core/src/main/scala/kafka/log/LogCleanerManager.scala
@@ -24,7 +24,7 @@ import kafka.utils.{Logging, Pool}
 import kafka.server.OffsetCheckpoint
 import collection.mutable
 import java.util.concurrent.locks.ReentrantLock
-import kafka.utils.Utils._
+import kafka.utils.CoreUtils._
 import java.util.concurrent.TimeUnit
 import kafka.common.{LogCleaningAbortedException, TopicAndPartition}
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/log/LogConfig.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/LogConfig.scala b/core/src/main/scala/kafka/log/LogConfig.scala
old mode 100644
new mode 100755
index 8b67aee..558c703
--- a/core/src/main/scala/kafka/log/LogConfig.scala
+++ b/core/src/main/scala/kafka/log/LogConfig.scala
@@ -197,7 +197,7 @@ object LogConfig {
    * Parse the given properties instance into a LogConfig object
    */
   def fromProps(props: Properties): LogConfig = {
-    import kafka.utils.Utils.evaluateDefaults
+    import kafka.utils.CoreUtils.evaluateDefaults
     val parsed = configDef.parse(evaluateDefaults(props))
     new LogConfig(segmentSize = parsed.get(SegmentBytesProp).asInstanceOf[Int],
                   segmentMs = parsed.get(SegmentMsProp).asInstanceOf[Long],

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/log/LogManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala
old mode 100644
new mode 100755
index 47d250a..a7a9b85
--- a/core/src/main/scala/kafka/log/LogManager.scala
+++ b/core/src/main/scala/kafka/log/LogManager.scala
@@ -132,7 +132,7 @@ class LogManager(val logDirs: Array[File],
         dirContent <- Option(dir.listFiles).toList
         logDir <- dirContent if logDir.isDirectory
       } yield {
-        Utils.runnable {
+        CoreUtils.runnable {
           debug("Loading log '" + logDir.getName + "'")
 
           val topicPartition = Log.parseTopicPartitionName(logDir)
@@ -210,7 +210,7 @@ class LogManager(val logDirs: Array[File],
 
     // stop the cleaner first
     if (cleaner != null) {
-      Utils.swallow(cleaner.shutdown())
+      CoreUtils.swallow(cleaner.shutdown())
     }
 
     // close logs in each dir
@@ -223,7 +223,7 @@ class LogManager(val logDirs: Array[File],
       val logsInDir = logsByDir.getOrElse(dir.toString, Map()).values
 
       val jobsForDir = logsInDir map { log =>
-        Utils.runnable {
+        CoreUtils.runnable {
           // flush the log to ensure latest possible recovery point
           log.flush()
           log.close()
@@ -244,7 +244,7 @@ class LogManager(val logDirs: Array[File],
 
         // mark that the shutdown was clean by creating marker file
         debug("Writing clean shutdown marker at " + dir)
-        Utils.swallow(new File(dir, Log.CleanShutdownFile).createNewFile())
+        CoreUtils.swallow(new File(dir, Log.CleanShutdownFile).createNewFile())
       }
     } catch {
       case e: ExecutionException => {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/log/LogSegment.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/LogSegment.scala b/core/src/main/scala/kafka/log/LogSegment.scala
old mode 100644
new mode 100755
index 0256764..ed03953
--- a/core/src/main/scala/kafka/log/LogSegment.scala
+++ b/core/src/main/scala/kafka/log/LogSegment.scala
@@ -254,10 +254,10 @@ class LogSegment(val log: FileMessageSet,
    * Change the suffix for the index and log file for this log segment
    */
   def changeFileSuffixes(oldSuffix: String, newSuffix: String) {
-    val logRenamed = log.renameTo(new File(Utils.replaceSuffix(log.file.getPath, oldSuffix, newSuffix)))
+    val logRenamed = log.renameTo(new File(CoreUtils.replaceSuffix(log.file.getPath, oldSuffix, newSuffix)))
     if(!logRenamed)
       throw new KafkaStorageException("Failed to change the log file suffix from %s to %s for log segment %d".format(oldSuffix, newSuffix, baseOffset))
-    val indexRenamed = index.renameTo(new File(Utils.replaceSuffix(index.file.getPath, oldSuffix, newSuffix)))
+    val indexRenamed = index.renameTo(new File(CoreUtils.replaceSuffix(index.file.getPath, oldSuffix, newSuffix)))
     if(!indexRenamed)
       throw new KafkaStorageException("Failed to change the index file suffix from %s to %s for log segment %d".format(oldSuffix, newSuffix, baseOffset))
   }
@@ -266,8 +266,8 @@ class LogSegment(val log: FileMessageSet,
    * Close this log segment
    */
   def close() {
-    Utils.swallow(index.close)
-    Utils.swallow(log.close)
+    CoreUtils.swallow(index.close)
+    CoreUtils.swallow(log.close)
   }
   
   /**

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/log/OffsetIndex.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/OffsetIndex.scala b/core/src/main/scala/kafka/log/OffsetIndex.scala
old mode 100644
new mode 100755
index ca82c04..4ab22de
--- a/core/src/main/scala/kafka/log/OffsetIndex.scala
+++ b/core/src/main/scala/kafka/log/OffsetIndex.scala
@@ -24,7 +24,7 @@ import java.nio.channels._
 import java.util.concurrent.locks._
 import java.util.concurrent.atomic._
 import kafka.utils._
-import kafka.utils.Utils.inLock
+import kafka.utils.CoreUtils.inLock
 import kafka.common.InvalidOffsetException
 
 /**
@@ -81,7 +81,7 @@ class OffsetIndex(@volatile var file: File, val baseOffset: Long, val maxIndexSi
           idx.position(roundToExactMultiple(idx.limit, 8))
         idx
       } finally {
-        Utils.swallow(raf.close())
+        CoreUtils.swallow(raf.close())
       }
     }
   
@@ -287,7 +287,7 @@ class OffsetIndex(@volatile var file: File, val baseOffset: Long, val maxIndexSi
         this.maxEntries = this.mmap.limit / 8
         this.mmap.position(position)
       } finally {
-        Utils.swallow(raf.close())
+        CoreUtils.swallow(raf.close())
       }
     }
   }
@@ -319,7 +319,7 @@ class OffsetIndex(@volatile var file: File, val baseOffset: Long, val maxIndexSi
   def delete(): Boolean = {
     info("Deleting index " + this.file.getAbsolutePath)
     if(Os.isWindows)
-      Utils.swallow(forceUnmap(this.mmap))
+      CoreUtils.swallow(forceUnmap(this.mmap))
     this.file.delete()
   }
   

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/log/OffsetMap.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/OffsetMap.scala b/core/src/main/scala/kafka/log/OffsetMap.scala
old mode 100644
new mode 100755
index 42cdfbb..2940e47
--- a/core/src/main/scala/kafka/log/OffsetMap.scala
+++ b/core/src/main/scala/kafka/log/OffsetMap.scala
@@ -21,6 +21,7 @@ import java.util.Arrays
 import java.security.MessageDigest
 import java.nio.ByteBuffer
 import kafka.utils._
+import org.apache.kafka.common.utils.Utils
 
 trait OffsetMap {
   def slots: Int
@@ -158,7 +159,7 @@ class SkimpyOffsetMap(val memory: Int, val hashAlgorithm: String = "MD5") extend
    * @return The byte offset in the buffer at which the ith probing for the given hash would reside
    */
   private def positionOf(hash: Array[Byte], attempt: Int): Int = {
-    val probe = Utils.readInt(hash, math.min(attempt, hashSize - 4)) + math.max(0, attempt - hashSize + 4)
+    val probe = CoreUtils.readInt(hash, math.min(attempt, hashSize - 4)) + math.max(0, attempt - hashSize + 4)
     val slot = Utils.abs(probe) % slots
     this.probes += 1
     slot * bytesPerEntry

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/message/Message.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/message/Message.scala b/core/src/main/scala/kafka/message/Message.scala
old mode 100644
new mode 100755
index 7ba280f..999b115
--- a/core/src/main/scala/kafka/message/Message.scala
+++ b/core/src/main/scala/kafka/message/Message.scala
@@ -20,6 +20,7 @@ package kafka.message
 import java.nio._
 import scala.math._
 import kafka.utils._
+import org.apache.kafka.common.utils.Utils
 
 /**
  * Constants related to messages
@@ -146,7 +147,7 @@ class Message(val buffer: ByteBuffer) {
    * Compute the checksum of the message from the message contents
    */
   def computeChecksum(): Long = 
-    Utils.crc32(buffer.array, buffer.arrayOffset + MagicOffset,  buffer.limit - MagicOffset)
+    CoreUtils.crc32(buffer.array, buffer.arrayOffset + MagicOffset,  buffer.limit - MagicOffset)
   
   /**
    * Retrieve the previously computed CRC for this message

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/message/MessageAndMetadata.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/message/MessageAndMetadata.scala b/core/src/main/scala/kafka/message/MessageAndMetadata.scala
old mode 100644
new mode 100755
index d693abc..26b75c8
--- a/core/src/main/scala/kafka/message/MessageAndMetadata.scala
+++ b/core/src/main/scala/kafka/message/MessageAndMetadata.scala
@@ -18,7 +18,7 @@
 package kafka.message
 
 import kafka.serializer.Decoder
-import kafka.utils.Utils
+import org.apache.kafka.common.utils.Utils
 
 case class MessageAndMetadata[K, V](topic: String, partition: Int,
                                     private val rawMessage: Message, offset: Long,

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/message/MessageWriter.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/message/MessageWriter.scala b/core/src/main/scala/kafka/message/MessageWriter.scala
old mode 100644
new mode 100755
index 7eb72cb..0c6040e
--- a/core/src/main/scala/kafka/message/MessageWriter.scala
+++ b/core/src/main/scala/kafka/message/MessageWriter.scala
@@ -20,7 +20,7 @@ package kafka.message
 import java.io.{InputStream, OutputStream}
 import java.nio.ByteBuffer
 
-import kafka.utils.Crc32
+import org.apache.kafka.common.utils.Crc32
 
 class MessageWriter(segmentSize: Int) extends BufferingOutputStream(segmentSize) {
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/metrics/KafkaCSVMetricsReporter.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/metrics/KafkaCSVMetricsReporter.scala b/core/src/main/scala/kafka/metrics/KafkaCSVMetricsReporter.scala
old mode 100644
new mode 100755
index ea9559f..cc0da9f
--- a/core/src/main/scala/kafka/metrics/KafkaCSVMetricsReporter.scala
+++ b/core/src/main/scala/kafka/metrics/KafkaCSVMetricsReporter.scala
@@ -24,7 +24,7 @@ import com.yammer.metrics.Metrics
 import java.io.File
 import com.yammer.metrics.reporting.CsvReporter
 import java.util.concurrent.TimeUnit
-import kafka.utils.{Utils, VerifiableProperties, Logging}
+import kafka.utils.{CoreUtils, VerifiableProperties, Logging}
 
 
 private trait KafkaCSVMetricsReporterMBean extends KafkaMetricsReporterMBean
@@ -48,7 +48,7 @@ private class KafkaCSVMetricsReporter extends KafkaMetricsReporter
       if (!initialized) {
         val metricsConfig = new KafkaMetricsConfig(props)
         csvDir = new File(props.getString("kafka.csv.metrics.dir", "kafka_metrics"))
-        Utils.rm(csvDir)
+        CoreUtils.rm(csvDir)
         csvDir.mkdirs()
         underlying = new CsvReporter(Metrics.defaultRegistry(), csvDir)
         if (props.getBoolean("kafka.csv.metrics.reporter.enabled", default = false)) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/metrics/KafkaMetricsConfig.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/metrics/KafkaMetricsConfig.scala b/core/src/main/scala/kafka/metrics/KafkaMetricsConfig.scala
old mode 100644
new mode 100755
index 84f6208..ad9eb20
--- a/core/src/main/scala/kafka/metrics/KafkaMetricsConfig.scala
+++ b/core/src/main/scala/kafka/metrics/KafkaMetricsConfig.scala
@@ -20,7 +20,7 @@
 
 package kafka.metrics
 
-import kafka.utils.{VerifiableProperties, Utils}
+import kafka.utils.{VerifiableProperties, CoreUtils}
 
 class KafkaMetricsConfig(props: VerifiableProperties) {
 
@@ -28,7 +28,7 @@ class KafkaMetricsConfig(props: VerifiableProperties) {
    * Comma-separated list of reporter types. These classes should be on the
    * classpath and will be instantiated at run-time.
    */
-  val reporters = Utils.parseCsvList(props.getString("kafka.metrics.reporters", ""))
+  val reporters = CoreUtils.parseCsvList(props.getString("kafka.metrics.reporters", ""))
 
   /**
    * The metrics polling interval (in seconds).

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/metrics/KafkaMetricsReporter.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/metrics/KafkaMetricsReporter.scala b/core/src/main/scala/kafka/metrics/KafkaMetricsReporter.scala
old mode 100644
new mode 100755
index 14e4624..30fd0ea
--- a/core/src/main/scala/kafka/metrics/KafkaMetricsReporter.scala
+++ b/core/src/main/scala/kafka/metrics/KafkaMetricsReporter.scala
@@ -20,7 +20,7 @@
 
 package kafka.metrics
 
-import kafka.utils.{Utils, VerifiableProperties}
+import kafka.utils.{CoreUtils, VerifiableProperties}
 import java.util.concurrent.atomic.AtomicBoolean
 
 
@@ -56,10 +56,10 @@ object KafkaMetricsReporter {
         val metricsConfig = new KafkaMetricsConfig(verifiableProps)
         if(metricsConfig.reporters.size > 0) {
           metricsConfig.reporters.foreach(reporterType => {
-            val reporter = Utils.createObject[KafkaMetricsReporter](reporterType)
+            val reporter = CoreUtils.createObject[KafkaMetricsReporter](reporterType)
             reporter.init(verifiableProps)
             if (reporter.isInstanceOf[KafkaMetricsReporterMBean])
-              Utils.registerMBean(reporter, reporter.asInstanceOf[KafkaMetricsReporterMBean].getMBeanName)
+              CoreUtils.registerMBean(reporter, reporter.asInstanceOf[KafkaMetricsReporterMBean].getMBeanName)
           })
           ReporterStarted.set(true)
         }

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/network/BoundedByteBufferReceive.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/BoundedByteBufferReceive.scala b/core/src/main/scala/kafka/network/BoundedByteBufferReceive.scala
old mode 100644
new mode 100755
index a442545..c0d7726
--- a/core/src/main/scala/kafka/network/BoundedByteBufferReceive.scala
+++ b/core/src/main/scala/kafka/network/BoundedByteBufferReceive.scala
@@ -51,7 +51,7 @@ private[kafka] class BoundedByteBufferReceive(val maxSize: Int) extends Receive
     var read = 0
     // have we read the request size yet?
     if(sizeBuffer.remaining > 0)
-      read += Utils.read(channel, sizeBuffer)
+      read += CoreUtils.read(channel, sizeBuffer)
     // have we allocated the request buffer yet?
     if(contentBuffer == null && !sizeBuffer.hasRemaining) {
       sizeBuffer.rewind()
@@ -64,7 +64,7 @@ private[kafka] class BoundedByteBufferReceive(val maxSize: Int) extends Receive
     }
     // if we have a buffer read some stuff into it
     if(contentBuffer != null) {
-      read = Utils.read(channel, contentBuffer)
+      read = CoreUtils.read(channel, contentBuffer)
       // did we get everything?
       if(!contentBuffer.hasRemaining) {
         contentBuffer.rewind()

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/network/SocketServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala
old mode 100644
new mode 100755
index 0ad9057..8fbea7b
--- a/core/src/main/scala/kafka/network/SocketServer.scala
+++ b/core/src/main/scala/kafka/network/SocketServer.scala
@@ -33,6 +33,7 @@ import kafka.common.KafkaException
 import kafka.metrics.KafkaMetricsGroup
 import kafka.utils._
 import com.yammer.metrics.core.{Gauge, Meter}
+import org.apache.kafka.common.utils.Utils
 
 /**
  * An NIO socket server. The threading model is

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/producer/ByteArrayPartitioner.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/producer/ByteArrayPartitioner.scala b/core/src/main/scala/kafka/producer/ByteArrayPartitioner.scala
old mode 100644
new mode 100755
index 6a3b02e..e6b100e
--- a/core/src/main/scala/kafka/producer/ByteArrayPartitioner.scala
+++ b/core/src/main/scala/kafka/producer/ByteArrayPartitioner.scala
@@ -19,6 +19,7 @@ package kafka.producer
 
 
 import kafka.utils._
+import org.apache.kafka.common.utils.Utils
 
 class ByteArrayPartitioner(props: VerifiableProperties = null) extends Partitioner {
   def partition(key: Any, numPartitions: Int): Int = {

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/producer/DefaultPartitioner.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/producer/DefaultPartitioner.scala b/core/src/main/scala/kafka/producer/DefaultPartitioner.scala
old mode 100644
new mode 100755
index 3afb22e..1141ed1
--- a/core/src/main/scala/kafka/producer/DefaultPartitioner.scala
+++ b/core/src/main/scala/kafka/producer/DefaultPartitioner.scala
@@ -19,6 +19,7 @@ package kafka.producer
 
 
 import kafka.utils._
+import org.apache.kafka.common.utils.Utils
 
 class DefaultPartitioner(props: VerifiableProperties = null) extends Partitioner {
   private val random = new java.util.Random

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/producer/Producer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/producer/Producer.scala b/core/src/main/scala/kafka/producer/Producer.scala
old mode 100644
new mode 100755
index e38d2fa..4be06c8
--- a/core/src/main/scala/kafka/producer/Producer.scala
+++ b/core/src/main/scala/kafka/producer/Producer.scala
@@ -58,9 +58,9 @@ class Producer[K,V](val config: ProducerConfig,
   def this(config: ProducerConfig) =
     this(config,
          new DefaultEventHandler[K,V](config,
-                                      Utils.createObject[Partitioner](config.partitionerClass, config.props),
-                                      Utils.createObject[Encoder[V]](config.serializerClass, config.props),
-                                      Utils.createObject[Encoder[K]](config.keySerializerClass, config.props),
+                                      CoreUtils.createObject[Partitioner](config.partitionerClass, config.props),
+                                      CoreUtils.createObject[Encoder[V]](config.serializerClass, config.props),
+                                      CoreUtils.createObject[Encoder[K]](config.keySerializerClass, config.props),
                                       new ProducerPool(config)))
 
   /**

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/producer/ProducerConfig.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/producer/ProducerConfig.scala b/core/src/main/scala/kafka/producer/ProducerConfig.scala
old mode 100644
new mode 100755
index 3cdf23d..08a4e51
--- a/core/src/main/scala/kafka/producer/ProducerConfig.scala
+++ b/core/src/main/scala/kafka/producer/ProducerConfig.scala
@@ -19,8 +19,8 @@ package kafka.producer
 
 import async.AsyncProducerConfig
 import java.util.Properties
-import kafka.utils.{Utils, VerifiableProperties}
-import kafka.message.{CompressionCodec, NoCompressionCodec}
+import kafka.utils.{CoreUtils, VerifiableProperties}
+import kafka.message.NoCompressionCodec
 import kafka.common.{InvalidConfigException, Config}
 
 object ProducerConfig extends Config {
@@ -90,7 +90,7 @@ class ProducerConfig private (val props: VerifiableProperties)
    *
    *  If the compression codec is NoCompressionCodec, compression is disabled for all topics
    */
-  val compressedTopics = Utils.parseCsvList(props.getString("compressed.topics", null))
+  val compressedTopics = CoreUtils.parseCsvList(props.getString("compressed.topics", null))
 
   /** The leader may be unavailable transiently, which can fail the sending of a message.
     *  This property specifies the number of retries when such failures occur.

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala b/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala
old mode 100644
new mode 100755
index 821901e..a6179a9
--- a/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala
+++ b/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala
@@ -21,12 +21,13 @@ import kafka.common._
 import kafka.message.{NoCompressionCodec, Message, ByteBufferMessageSet}
 import kafka.producer._
 import kafka.serializer.Encoder
-import kafka.utils.{Utils, Logging, SystemTime}
+import kafka.utils.{CoreUtils, Logging, SystemTime}
 import scala.util.Random
 import scala.collection.{Seq, Map}
 import scala.collection.mutable.{ArrayBuffer, HashMap, Set}
 import java.util.concurrent.atomic._
 import kafka.api.{TopicMetadata, ProducerRequest}
+import org.apache.kafka.common.utils.Utils
 
 class DefaultEventHandler[K,V](config: ProducerConfig,
                                private val partitioner: Partitioner,
@@ -64,7 +65,7 @@ class DefaultEventHandler[K,V](config: ProducerConfig,
       topicMetadataToRefresh ++= outstandingProduceRequests.map(_.topic)
       if (topicMetadataRefreshInterval >= 0 &&
           SystemTime.milliseconds - lastTopicMetadataRefreshTime > topicMetadataRefreshInterval) {
-        Utils.swallowError(brokerPartitionInfo.updateInfo(topicMetadataToRefresh.toSet, correlationId.getAndIncrement))
+        CoreUtils.swallowError(brokerPartitionInfo.updateInfo(topicMetadataToRefresh.toSet, correlationId.getAndIncrement))
         sendPartitionPerTopicCache.clear()
         topicMetadataToRefresh.clear
         lastTopicMetadataRefreshTime = SystemTime.milliseconds
@@ -75,7 +76,7 @@ class DefaultEventHandler[K,V](config: ProducerConfig,
         // back off and update the topic metadata cache before attempting another send operation
         Thread.sleep(config.retryBackoffMs)
         // get topics of the outstanding produce requests and refresh metadata for those
-        Utils.swallowError(brokerPartitionInfo.updateInfo(outstandingProduceRequests.map(_.topic).toSet, correlationId.getAndIncrement))
+        CoreUtils.swallowError(brokerPartitionInfo.updateInfo(outstandingProduceRequests.map(_.topic).toSet, correlationId.getAndIncrement))
         sendPartitionPerTopicCache.clear()
         remainingRetries -= 1
         producerStats.resendRate.mark()
@@ -262,7 +263,7 @@ class DefaultEventHandler[K,V](config: ProducerConfig,
           if (logger.isTraceEnabled) {
             val successfullySentData = response.status.filter(_._2.error == ErrorMapping.NoError)
             successfullySentData.foreach(m => messagesPerTopic(m._1).foreach(message =>
-              trace("Successfully sent message: %s".format(if(message.message.isNull) null else Utils.readString(message.message.payload)))))
+              trace("Successfully sent message: %s".format(if(message.message.isNull) null else message.message.toString()))))
           }
           val failedPartitionsAndStatus = response.status.filter(_._2.error != ErrorMapping.NoError).toSeq
           failedTopicPartitions = failedPartitionsAndStatus.map(partitionStatus => partitionStatus._1)

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/server/AbstractFetcherManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/AbstractFetcherManager.scala b/core/src/main/scala/kafka/server/AbstractFetcherManager.scala
old mode 100644
new mode 100755
index 94aa952..f8f9331
--- a/core/src/main/scala/kafka/server/AbstractFetcherManager.scala
+++ b/core/src/main/scala/kafka/server/AbstractFetcherManager.scala
@@ -20,11 +20,12 @@ package kafka.server
 import scala.collection.mutable
 import scala.collection.Set
 import scala.collection.Map
-import kafka.utils.{Utils, Logging}
+import kafka.utils.Logging
 import kafka.cluster.BrokerEndpoint
 import kafka.metrics.KafkaMetricsGroup
 import kafka.common.TopicAndPartition
 import com.yammer.metrics.core.Gauge
+import org.apache.kafka.common.utils.Utils
 
 abstract class AbstractFetcherManager(protected val name: String, clientId: String, numFetchers: Int = 1)
   extends Logging with KafkaMetricsGroup {
@@ -128,4 +129,4 @@ abstract class AbstractFetcherManager(protected val name: String, clientId: Stri
 
 case class BrokerAndFetcherId(broker: BrokerEndpoint, fetcherId: Int)
 
-case class BrokerAndInitialOffset(broker: BrokerEndpoint, initOffset: Long)
\ No newline at end of file
+case class BrokerAndInitialOffset(broker: BrokerEndpoint, initOffset: Long)

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
old mode 100644
new mode 100755
index 93f67d5..1e26de2
--- a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
+++ b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
@@ -22,7 +22,7 @@ import kafka.utils.{Pool, ShutdownableThread}
 import kafka.consumer.{PartitionTopicInfo, SimpleConsumer}
 import kafka.api.{FetchRequest, FetchResponse, FetchResponsePartitionData, FetchRequestBuilder}
 import kafka.common.{KafkaException, ClientIdAndBroker, TopicAndPartition, ErrorMapping}
-import kafka.utils.Utils.inLock
+import kafka.utils.CoreUtils.inLock
 import kafka.message.{InvalidMessageException, ByteBufferMessageSet, MessageAndOffset}
 import kafka.metrics.KafkaMetricsGroup
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/server/BrokerMetadataCheckpoint.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/BrokerMetadataCheckpoint.scala b/core/src/main/scala/kafka/server/BrokerMetadataCheckpoint.scala
old mode 100644
new mode 100755
index 0e542ff..01e8f72
--- a/core/src/main/scala/kafka/server/BrokerMetadataCheckpoint.scala
+++ b/core/src/main/scala/kafka/server/BrokerMetadataCheckpoint.scala
@@ -20,7 +20,7 @@ package kafka.server
 import java.io._
 import java.util.Properties
 import kafka.utils._
-
+import org.apache.kafka.common.utils.Utils
 
 case class BrokerMetadata(brokerId: Int)
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/server/KafkaConfig.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
old mode 100644
new mode 100755
index cf1a5a6..69b772c
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -23,7 +23,7 @@ import kafka.api.ApiVersion
 import kafka.cluster.EndPoint
 import kafka.consumer.ConsumerConfig
 import kafka.message.{BrokerCompressionCodec, CompressionCodec, Message, MessageSet}
-import kafka.utils.Utils
+import kafka.utils.CoreUtils
 import org.apache.kafka.common.config.ConfigDef
 import org.apache.kafka.common.protocol.SecurityProtocol
 import scala.collection.{immutable, JavaConversions, Map}
@@ -492,7 +492,7 @@ object KafkaConfig {
    * Parse the given properties instance into a KafkaConfig object
    */
   def fromProps(props: Properties): KafkaConfig = {
-    import kafka.utils.Utils.evaluateDefaults
+    import kafka.utils.CoreUtils.evaluateDefaults
     val parsed = configDef.parse(evaluateDefaults(props))
     new KafkaConfig(
       /** ********* Zookeeper Configuration ***********/
@@ -755,7 +755,7 @@ class KafkaConfig(/** ********* Zookeeper Configuration ***********/
   val advertisedHostName: String = _advertisedHostName.getOrElse(hostName)
   val advertisedPort: Int = _advertisedPort.getOrElse(port)
   val advertisedListeners = getAdvertisedListeners()
-  val logDirs = Utils.parseCsvList(_logDirs.getOrElse(_logDir))
+  val logDirs = CoreUtils.parseCsvList(_logDirs.getOrElse(_logDir))
 
   val logRollTimeMillis = _logRollTimeMillis.getOrElse(60 * 60 * 1000L * logRollTimeHours)
   val logRollTimeJitterMillis = _logRollTimeJitterMillis.getOrElse(60 * 60 * 1000L * logRollTimeJitterHours)
@@ -780,7 +780,7 @@ class KafkaConfig(/** ********* Zookeeper Configuration ***********/
 
   private def getMap(propName: String, propValue: String): Map[String, String] = {
     try {
-      Utils.parseCsvMap(propValue)
+      CoreUtils.parseCsvMap(propValue)
     } catch {
       case e: Exception => throw new IllegalArgumentException("Error parsing configuration property '%s': %s".format(propName, e.getMessage))
     }
@@ -789,7 +789,7 @@ class KafkaConfig(/** ********* Zookeeper Configuration ***********/
   private def validateUniquePortAndProtocol(listeners: String) {
 
     val endpoints = try {
-      val listenerList = Utils.parseCsvList(listeners)
+      val listenerList = CoreUtils.parseCsvList(listeners)
       listenerList.map(listener => EndPoint.createEndPoint(listener))
     } catch {
       case e: Exception => throw new IllegalArgumentException("Error creating broker listeners from '%s': %s".format(listeners, e.getMessage))
@@ -806,9 +806,9 @@ class KafkaConfig(/** ********* Zookeeper Configuration ***********/
   private def getListeners(): immutable.Map[SecurityProtocol, EndPoint] = {
     if (_listeners.isDefined) {
       validateUniquePortAndProtocol(_listeners.get)
-      Utils.listenerListToEndPoints(_listeners.get)
+      CoreUtils.listenerListToEndPoints(_listeners.get)
     } else {
-      Utils.listenerListToEndPoints("PLAINTEXT://" + hostName + ":" + port)
+      CoreUtils.listenerListToEndPoints("PLAINTEXT://" + hostName + ":" + port)
     }
   }
 
@@ -818,9 +818,9 @@ class KafkaConfig(/** ********* Zookeeper Configuration ***********/
   private def getAdvertisedListeners(): immutable.Map[SecurityProtocol, EndPoint] = {
     if (_advertisedListeners.isDefined) {
       validateUniquePortAndProtocol(_advertisedListeners.get)
-      Utils.listenerListToEndPoints(_advertisedListeners.get)
+      CoreUtils.listenerListToEndPoints(_advertisedListeners.get)
     } else if (_advertisedHostName.isDefined || _advertisedPort.isDefined ) {
-      Utils.listenerListToEndPoints("PLAINTEXT://" + advertisedHostName  + ":" + advertisedPort)
+      CoreUtils.listenerListToEndPoints("PLAINTEXT://" + advertisedHostName  + ":" + advertisedPort)
     } else {
       getListeners()
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
old mode 100644
new mode 100755
index 4d86bdf..a1558af
--- a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
+++ b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
@@ -22,6 +22,7 @@ import kafka.utils._
 import kafka.metrics.KafkaMetricsGroup
 import java.util.concurrent.TimeUnit
 import com.yammer.metrics.core.Meter
+import org.apache.kafka.common.utils.Utils
 
 /**
  * A thread that answers kafka requests.

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/server/KafkaServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
old mode 100644
new mode 100755
index 9df2cf4..c63f4ba
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -319,27 +319,27 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime) extends Logg
 
       val canShutdown = isShuttingDown.compareAndSet(false, true)
       if (canShutdown && shutdownLatch.getCount > 0) {
-        Utils.swallow(controlledShutdown())
+        CoreUtils.swallow(controlledShutdown())
         brokerState.newState(BrokerShuttingDown)
         if(socketServer != null)
-          Utils.swallow(socketServer.shutdown())
+          CoreUtils.swallow(socketServer.shutdown())
         if(requestHandlerPool != null)
-          Utils.swallow(requestHandlerPool.shutdown())
+          CoreUtils.swallow(requestHandlerPool.shutdown())
         if(offsetManager != null)
           offsetManager.shutdown()
-        Utils.swallow(kafkaScheduler.shutdown())
+        CoreUtils.swallow(kafkaScheduler.shutdown())
         if(apis != null)
-          Utils.swallow(apis.close())
+          CoreUtils.swallow(apis.close())
         if(replicaManager != null)
-          Utils.swallow(replicaManager.shutdown())
+          CoreUtils.swallow(replicaManager.shutdown())
         if(logManager != null)
-          Utils.swallow(logManager.shutdown())
+          CoreUtils.swallow(logManager.shutdown())
         if(consumerCoordinator != null)
-          Utils.swallow(consumerCoordinator.shutdown())
+          CoreUtils.swallow(consumerCoordinator.shutdown())
         if(kafkaController != null)
-          Utils.swallow(kafkaController.shutdown())
+          CoreUtils.swallow(kafkaController.shutdown())
         if(zkClient != null)
-          Utils.swallow(zkClient.close())
+          CoreUtils.swallow(zkClient.close())
 
         brokerState.newState(NotRunning)
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/server/MetadataCache.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/MetadataCache.scala b/core/src/main/scala/kafka/server/MetadataCache.scala
old mode 100644
new mode 100755
index 008f02b..4460b42
--- a/core/src/main/scala/kafka/server/MetadataCache.scala
+++ b/core/src/main/scala/kafka/server/MetadataCache.scala
@@ -26,7 +26,7 @@ import kafka.controller.KafkaController.StateChangeLogger
 import org.apache.kafka.common.protocol.SecurityProtocol
 import scala.collection.{Seq, Set, mutable}
 import kafka.utils.Logging
-import kafka.utils.Utils._
+import kafka.utils.CoreUtils._
 
 import java.util.concurrent.locks.ReentrantReadWriteLock
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/server/OffsetManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/OffsetManager.scala b/core/src/main/scala/kafka/server/OffsetManager.scala
old mode 100644
new mode 100755
index 395b1db..420e2c3
--- a/core/src/main/scala/kafka/server/OffsetManager.scala
+++ b/core/src/main/scala/kafka/server/OffsetManager.scala
@@ -21,6 +21,7 @@ import org.apache.kafka.common.protocol.types.{Struct, Schema, Field}
 import org.apache.kafka.common.protocol.types.Type.STRING
 import org.apache.kafka.common.protocol.types.Type.INT32
 import org.apache.kafka.common.protocol.types.Type.INT64
+import org.apache.kafka.common.utils.Utils
 
 import kafka.utils._
 import kafka.common._

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/server/ZookeeperLeaderElector.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/ZookeeperLeaderElector.scala b/core/src/main/scala/kafka/server/ZookeeperLeaderElector.scala
old mode 100644
new mode 100755
index a75818a..a5c5fb3
--- a/core/src/main/scala/kafka/server/ZookeeperLeaderElector.scala
+++ b/core/src/main/scala/kafka/server/ZookeeperLeaderElector.scala
@@ -17,7 +17,7 @@
 package kafka.server
 
 import kafka.utils.ZkUtils._
-import kafka.utils.Utils._
+import kafka.utils.CoreUtils._
 import kafka.utils.{Json, SystemTime, Logging}
 import org.I0Itec.zkclient.exception.ZkNodeExistsException
 import org.I0Itec.zkclient.IZkDataListener

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/tools/ConsoleConsumer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/tools/ConsoleConsumer.scala b/core/src/main/scala/kafka/tools/ConsoleConsumer.scala
old mode 100644
new mode 100755
index 910691e..9d9b781
--- a/core/src/main/scala/kafka/tools/ConsoleConsumer.scala
+++ b/core/src/main/scala/kafka/tools/ConsoleConsumer.scala
@@ -28,6 +28,7 @@ import kafka.serializer._
 import kafka.utils._
 import kafka.metrics.KafkaMetricsReporter
 import kafka.consumer.{Blacklist,Whitelist,ConsumerConfig,Consumer}
+import org.apache.kafka.common.utils.Utils
 
 /**
  * Consumer that dumps messages out to standard out.

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/tools/DumpLogSegments.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/tools/DumpLogSegments.scala b/core/src/main/scala/kafka/tools/DumpLogSegments.scala
old mode 100644
new mode 100755
index b7a3630..fc11a2a
--- a/core/src/main/scala/kafka/tools/DumpLogSegments.scala
+++ b/core/src/main/scala/kafka/tools/DumpLogSegments.scala
@@ -25,6 +25,7 @@ import collection.mutable
 import joptsimple.OptionParser
 import kafka.serializer.Decoder
 import kafka.utils.VerifiableProperties
+import org.apache.kafka.common.utils.Utils
 
 object DumpLogSegments {
 
@@ -64,8 +65,8 @@ object DumpLogSegments {
     val maxMessageSize = options.valueOf(maxMessageSizeOpt).intValue()
     val isDeepIteration = if(options.has(deepIterationOpt)) true else false
   
-    val valueDecoder: Decoder[_] = Utils.createObject[Decoder[_]](options.valueOf(valueDecoderOpt), new VerifiableProperties)
-    val keyDecoder: Decoder[_] = Utils.createObject[Decoder[_]](options.valueOf(keyDecoderOpt), new VerifiableProperties)
+    val valueDecoder: Decoder[_] = CoreUtils.createObject[Decoder[_]](options.valueOf(valueDecoderOpt), new VerifiableProperties)
+    val keyDecoder: Decoder[_] = CoreUtils.createObject[Decoder[_]](options.valueOf(keyDecoderOpt), new VerifiableProperties)
 
     val misMatchesForIndexFilesMap = new mutable.HashMap[String, List[(Long, Long)]]
     val nonConsecutivePairsForLogFilesMap = new mutable.HashMap[String, List[(Long, Long)]]

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/tools/KafkaMigrationTool.java
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/tools/KafkaMigrationTool.java b/core/src/main/scala/kafka/tools/KafkaMigrationTool.java
old mode 100644
new mode 100755
index 026d819..f19df0c
--- a/core/src/main/scala/kafka/tools/KafkaMigrationTool.java
+++ b/core/src/main/scala/kafka/tools/KafkaMigrationTool.java
@@ -21,7 +21,6 @@ import joptsimple.*;
 import kafka.javaapi.producer.Producer;
 import kafka.producer.KeyedMessage;
 import kafka.producer.ProducerConfig;
-import kafka.utils.Utils;
 
 import java.io.File;
 import java.io.FileInputStream;
@@ -39,6 +38,7 @@ import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.kafka.common.utils.Utils;
 
 
 /**

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/tools/MirrorMaker.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/tools/MirrorMaker.scala b/core/src/main/scala/kafka/tools/MirrorMaker.scala
old mode 100644
new mode 100755
index ec07743..9548521
--- a/core/src/main/scala/kafka/tools/MirrorMaker.scala
+++ b/core/src/main/scala/kafka/tools/MirrorMaker.scala
@@ -24,17 +24,19 @@ import java.util.{Collections, Properties}
 
 import com.yammer.metrics.core.Gauge
 import joptsimple.OptionParser
-import kafka.consumer.{KafkaStream, Blacklist, ConsumerConfig, ConsumerThreadId, ConsumerTimeoutException, TopicFilter, Whitelist, ZookeeperConsumerConnector}
+import kafka.consumer.{Blacklist, ConsumerConfig, ConsumerThreadId, ConsumerTimeoutException, TopicFilter, Whitelist, ZookeeperConsumerConnector}
 import kafka.javaapi.consumer.ConsumerRebalanceListener
 import kafka.message.MessageAndMetadata
 import kafka.metrics.KafkaMetricsGroup
 import kafka.serializer.DefaultDecoder
-import kafka.utils.{CommandLineUtils, Logging, Utils}
+import kafka.utils.{CommandLineUtils, Logging, CoreUtils}
 import org.apache.kafka.clients.producer.internals.ErrorLoggingCallback
 import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord, RecordMetadata}
+import org.apache.kafka.common.utils.Utils
 
 import scala.collection.JavaConversions._
 
+
 /**
  * The mirror maker has the following architecture:
  * - There are N mirror maker thread shares one ZookeeperConsumerConnector and each owns a Kafka stream.
@@ -207,9 +209,9 @@ object MirrorMaker extends Logging with KafkaMetricsGroup {
     val customRebalanceListener = {
       if (customRebalanceListenerClass != null) {
         if (rebalanceListenerArgs != null)
-          Some(Utils.createObject[ConsumerRebalanceListener](customRebalanceListenerClass, rebalanceListenerArgs))
+          Some(CoreUtils.createObject[ConsumerRebalanceListener](customRebalanceListenerClass, rebalanceListenerArgs))
         else
-          Some(Utils.createObject[ConsumerRebalanceListener](customRebalanceListenerClass))
+          Some(CoreUtils.createObject[ConsumerRebalanceListener](customRebalanceListenerClass))
       } else {
         None
       }
@@ -237,9 +239,9 @@ object MirrorMaker extends Logging with KafkaMetricsGroup {
     messageHandler = {
       if (customMessageHandlerClass != null) {
         if (messageHandlerArgs != null)
-          Utils.createObject[MirrorMakerMessageHandler](customMessageHandlerClass, messageHandlerArgs)
+          CoreUtils.createObject[MirrorMakerMessageHandler](customMessageHandlerClass, messageHandlerArgs)
         else
-          Utils.createObject[MirrorMakerMessageHandler](customMessageHandlerClass)
+          CoreUtils.createObject[MirrorMakerMessageHandler](customMessageHandlerClass)
       } else {
         defaultMirrorMakerMessageHandler
       }

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/tools/SimpleConsumerShell.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/tools/SimpleConsumerShell.scala b/core/src/main/scala/kafka/tools/SimpleConsumerShell.scala
old mode 100644
new mode 100755
index 7379fe3..9a6804c
--- a/core/src/main/scala/kafka/tools/SimpleConsumerShell.scala
+++ b/core/src/main/scala/kafka/tools/SimpleConsumerShell.scala
@@ -25,6 +25,7 @@ import kafka.api.{OffsetRequest, FetchRequestBuilder, Request}
 import kafka.cluster.BrokerEndpoint
 import scala.collection.JavaConversions._
 import kafka.common.TopicAndPartition
+import org.apache.kafka.common.utils.Utils
 
 /**
  * Command line program to dump out messages to standard out using the simple consumer

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/tools/StateChangeLogMerger.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/tools/StateChangeLogMerger.scala b/core/src/main/scala/kafka/tools/StateChangeLogMerger.scala
old mode 100644
new mode 100755
index b34b8c7..8b523e7
--- a/core/src/main/scala/kafka/tools/StateChangeLogMerger.scala
+++ b/core/src/main/scala/kafka/tools/StateChangeLogMerger.scala
@@ -22,7 +22,7 @@ import scala.util.matching.Regex
 import collection.mutable
 import java.util.Date
 import java.text.SimpleDateFormat
-import kafka.utils.{Utils, Logging, CommandLineUtils}
+import kafka.utils.{CoreUtils, Logging, CommandLineUtils}
 import kafka.common.Topic
 import java.io.{BufferedOutputStream, OutputStream}
 
@@ -115,7 +115,7 @@ object StateChangeLogMerger extends Logging {
     }
     if (options.has(partitionsOpt)) {
       partitions = options.valueOf(partitionsOpt).split(",").toList.map(_.toInt)
-      val duplicatePartitions = Utils.duplicates(partitions)
+      val duplicatePartitions = CoreUtils.duplicates(partitions)
       if (duplicatePartitions.nonEmpty) {
         System.err.println("The list of partitions contains repeated entries: %s".format(duplicatePartitions.mkString(",")))
         System.exit(1)

http://git-wip-us.apache.org/repos/asf/kafka/blob/9c23d935/core/src/main/scala/kafka/tools/TestEndToEndLatency.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/tools/TestEndToEndLatency.scala b/core/src/main/scala/kafka/tools/TestEndToEndLatency.scala
deleted file mode 100644
index 48cff20..0000000
--- a/core/src/main/scala/kafka/tools/TestEndToEndLatency.scala
+++ /dev/null
@@ -1,92 +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 kafka.tools
-
-import org.apache.kafka.clients.producer.{ProducerConfig, ProducerRecord, KafkaProducer}
-
-import kafka.consumer._
-
-import java.util.Properties
-import java.util.Arrays
-
-object TestEndToEndLatency {
-  def main(args: Array[String]) {
-    if (args.length != 6) {
-      System.err.println("USAGE: java " + getClass().getName + " broker_list zookeeper_connect topic num_messages consumer_fetch_max_wait producer_acks")
-      System.exit(1)
-    }
-
-    val brokerList = args(0)
-    val zkConnect = args(1)
-    val topic = args(2)
-    val numMessages = args(3).toInt
-    val consumerFetchMaxWait = args(4).toInt
-    val producerAcks = args(5).toInt
-
-    val consumerProps = new Properties()
-    consumerProps.put("group.id", topic)
-    consumerProps.put("auto.commit.enable", "false")
-    consumerProps.put("auto.offset.reset", "largest")
-    consumerProps.put("zookeeper.connect", zkConnect)
-    consumerProps.put("fetch.wait.max.ms", consumerFetchMaxWait.toString)
-    consumerProps.put("socket.timeout.ms", 1201000.toString)
-
-    val config = new ConsumerConfig(consumerProps)
-    val connector = Consumer.create(config)
-    val stream = connector.createMessageStreams(Map(topic -> 1)).get(topic).head.head
-    val iter = stream.iterator
-
-    val producerProps = new Properties()
-    producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList)
-    producerProps.put(ProducerConfig.LINGER_MS_CONFIG, "0")
-    producerProps.put(ProducerConfig.BLOCK_ON_BUFFER_FULL_CONFIG, "true")
-    producerProps.put(ProducerConfig.ACKS_CONFIG, producerAcks.toString)
-    producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer")
-    producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer")
-    val producer = new KafkaProducer[Array[Byte],Array[Byte]](producerProps)
-
-    // make sure the consumer fetcher has started before sending data since otherwise
-    // the consumption from the tail will skip the first message and hence be blocked
-    Thread.sleep(5000)
-
-    val message = "hello there beautiful".getBytes
-    var totalTime = 0.0
-    val latencies = new Array[Long](numMessages)
-    for (i <- 0 until numMessages) {
-      val begin = System.nanoTime
-      producer.send(new ProducerRecord[Array[Byte],Array[Byte]](topic, message))
-      val received = iter.next
-      val elapsed = System.nanoTime - begin
-      // poor man's progress bar
-      if (i % 1000 == 0)
-        println(i + "\t" + elapsed / 1000.0 / 1000.0)
-      totalTime += elapsed
-      latencies(i) = (elapsed / 1000 / 1000)
-    }
-    println("Avg latency: %.4f ms\n".format(totalTime / numMessages / 1000.0 / 1000.0))
-    Arrays.sort(latencies)
-    val p50 = latencies((latencies.length * 0.5).toInt)
-    val p99 = latencies((latencies.length * 0.99).toInt) 
-    val p999 = latencies((latencies.length * 0.999).toInt)
-    println("Percentiles: 50th = %d, 99th = %d, 99.9th = %d".format(p50, p99, p999))
-    producer.close()
-    connector.commitOffsets(true)
-    connector.shutdown()
-    System.exit(0)
-  }
-}
\ No newline at end of file


Mime
View raw message