kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jun...@apache.org
Subject git commit: kafka-882; Enhance 0.7 ProducerPerformance to send sequential MessageID as in 0.8; patched by John Fung; reviewed by Jun Rao
Date Tue, 23 Jul 2013 17:20:30 GMT
Updated Branches:
  refs/heads/0.7 68c8434f6 -> a984f2fe5


kafka-882; Enhance 0.7 ProducerPerformance to send sequential MessageID as in 0.8; patched
by John Fung; 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/a984f2fe
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/a984f2fe
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/a984f2fe

Branch: refs/heads/0.7
Commit: a984f2fe5e3370163fb53586a04611ebd878c5ae
Parents: 68c8434
Author: John Fung <fung.john@gmail.com>
Authored: Tue Jul 23 10:23:11 2013 -0700
Committer: Jun Rao <junrao@gmail.com>
Committed: Tue Jul 23 10:23:11 2013 -0700

----------------------------------------------------------------------
 .../scala/kafka/perf/ProducerPerformance.scala  | 52 ++++++++++++++++++--
 1 file changed, 48 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/a984f2fe/perf/src/main/scala/kafka/perf/ProducerPerformance.scala
----------------------------------------------------------------------
diff --git a/perf/src/main/scala/kafka/perf/ProducerPerformance.scala b/perf/src/main/scala/kafka/perf/ProducerPerformance.scala
index 5888f1e..a8bd47f 100644
--- a/perf/src/main/scala/kafka/perf/ProducerPerformance.scala
+++ b/perf/src/main/scala/kafka/perf/ProducerPerformance.scala
@@ -97,6 +97,13 @@ object ProducerPerformance extends Logging {
       .describedAs("compression codec ")
       .ofType(classOf[java.lang.Integer])
       .defaultsTo(0)
+    val initialMessageIdOpt = parser.accepts("initial-message-id", "The is used for generating
test data, If set, messages will be tagged with an " +
+      "ID and sent by producer starting from this ID sequentially. Message content will be
String type and " +
+      "in the form of 'Message:000...1:xxx...'")
+      .withRequiredArg()
+      .describedAs("initial message id")
+      .ofType(classOf[java.lang.Integer])
+      .defaultsTo(-1)
 
     val options = parser.parse(args : _*)
     for(arg <- List(topicOpt, brokerInfoOpt, numMessagesOpt)) {
@@ -119,6 +126,7 @@ object ProducerPerformance extends Logging {
     var batchSize = options.valueOf(batchSizeOpt).intValue
     val numThreads = options.valueOf(numThreadsOpt).intValue
     val compressionCodec = CompressionCodec.getCompressionCodec(options.valueOf(compressionCodecOption).intValue)
+    var initialMessageId = options.valueOf(initialMessageIdOpt).intValue()
   }
 
   private def getStringOfLength(len: Int) : String = {
@@ -157,6 +165,35 @@ object ProducerPerformance extends Logging {
     }
     val producerConfig = new ProducerConfig(props)
     val producer = new Producer[Message, Message](producerConfig)
+    val seqIdNumDigit = 10  // no. of digits for max int value
+    // generate the sequential message ID
+    private val SEP            = ":"              // message field separator
+    private val messageIdLabel = "MessageID"
+    private val threadIdLabel  = "ThreadID"
+    private val topicLabel     = "Topic"
+    private var leftPaddedSeqId : String = ""
+
+    private def generateMessageWithSeqId(topic: String, msgId: Long, msgSize: Int): Array[Byte]
= {
+      // Each thread gets a unique range of sequential no. for its ids.
+      // Eg. 1000 msg in 10 threads => 100 msg per thread
+      // thread 0 IDs :   0 ~  99
+      // thread 1 IDs : 100 ~ 199
+      // thread 2 IDs : 200 ~ 299
+      // . . .
+      leftPaddedSeqId = String.format("%0"+seqIdNumDigit+"d", long2Long(msgId))
+
+      val msgHeader = topicLabel      + SEP +
+              topic           + SEP +
+              threadIdLabel   + SEP +
+              threadId        + SEP +
+              messageIdLabel  + SEP +
+              leftPaddedSeqId + SEP
+
+      // pad the rest of the message with 'x'
+      val seqMsgString = String.format("%1$-"+msgSize+"s", msgHeader).replace(' ', 'x')
+      debug(seqMsgString)
+      return seqMsgString.getBytes()
+    }
 
     override def run {
       var bytesSent = 0L
@@ -195,10 +232,17 @@ object ProducerPerformance extends Logging {
             nSends += config.batchSize
           }else {
             if(!config.isFixSize) {
-              strLength = rand.nextInt(config.messageSize)
-              val messageBytes = getByteArrayOfLength(strLength)
-              rand.nextBytes(messageBytes)
-              val message = new Message(messageBytes)
+              strLength = rand.nextInt(config.messageSize) + 1
+              var message : Message = null
+              if (config.initialMessageId > -1) {
+                val seqId = config.initialMessageId + (messagesPerThread * threadId) + j
+                message = new Message(generateMessageWithSeqId(config.topic, seqId, strLength))
+              }
+              else {
+                val messageBytes = getByteArrayOfLength(strLength)
+                rand.nextBytes(messageBytes)
+                message = new Message(messageBytes)
+              }
               producer.send(new ProducerData[Message,Message](config.topic, message))
               debug(config.topic + "-checksum:" + message.checksum)
               bytesSent += message.payloadSize


Mime
View raw message