Kafka has become a TLP

git-svn-id: https://svn.apache.org/repos/asf/kafka/branches/0.7.1@1414576 13f79535-47bb-0310-9956-ffa450edef68

NOTICE

@@ -1,5 +1,5 @@
 Apache Kafka
-Copyright 2011 The Apache Software Foundation.
+Copyright 2012 The Apache Software Foundation.
 
 This product includes software developed at
 The Apache Software Foundation (http://www.apache.org/).

core/src/main/scala/kafka/consumer/ConsumerIterator.scala

@@ -47,6 +47,7 @@ class ConsumerIterator[T](private val channel: BlockingQueue[FetchedDataChunk],
     val topic = currentTopicInfo.topic
     trace("Setting %s consumed offset to %d".format(topic, consumedOffset))
     ConsumerTopicStat.getConsumerTopicStat(topic).recordMessagesPerTopic(1)
+    ConsumerTopicStat.getConsumerAllTopicStat().recordMessagesPerTopic(1)
     item
   }
 

core/src/main/scala/kafka/consumer/ConsumerTopicStat.scala

@@ -22,19 +22,29 @@ import kafka.utils.{Pool, Utils, threadsafe, Logging}
 
 trait ConsumerTopicStatMBean {
   def getMessagesPerTopic: Long
+  def getBytesPerTopic: Long
 }
 
 @threadsafe
 class ConsumerTopicStat extends ConsumerTopicStatMBean {
   private val numCumulatedMessagesPerTopic = new AtomicLong(0)
+  private val numCumulatedBytesPerTopic = new AtomicLong(0)
 
   def getMessagesPerTopic: Long = numCumulatedMessagesPerTopic.get
 
   def recordMessagesPerTopic(nMessages: Int) = numCumulatedMessagesPerTopic.getAndAdd(nMessages)
+
+  def getBytesPerTopic: Long = numCumulatedBytesPerTopic.get
+
+  def recordBytesPerTopic(nBytes: Long) = numCumulatedBytesPerTopic.getAndAdd(nBytes)
 }
 
 object ConsumerTopicStat extends Logging {
   private val stats = new Pool[String, ConsumerTopicStat]
+  private val allTopicStat = new ConsumerTopicStat
+  Utils.registerMBean(allTopicStat, "kafka:type=kafka.ConsumerAllTopicStat")
+
+  def getConsumerAllTopicStat(): ConsumerTopicStat = allTopicStat
 
   def getConsumerTopicStat(topic: String): ConsumerTopicStat = {
     var stat = stats.get(topic)

core/src/main/scala/kafka/consumer/PartitionTopicInfo.scala

@@ -61,6 +61,8 @@ private[consumer] class PartitionTopicInfo(val topic: String,
       chunkQueue.put(new FetchedDataChunk(messages, this, fetchOffset))
       val newOffset = fetchedOffset.addAndGet(size)
       debug("updated fetch offset of ( %s ) to %d".format(this, newOffset))
+      ConsumerTopicStat.getConsumerTopicStat(topic).recordBytesPerTopic(size)
+      ConsumerTopicStat.getConsumerAllTopicStat().recordBytesPerTopic(size)
     }
     size
   }

project/build.properties

@@ -16,7 +16,7 @@
 #Mon Feb 28 11:55:49 PST 2011
 project.name=Kafka
 sbt.version=0.7.5
-project.version=0.7.0
+project.version=0.7.1
 build.scala.versions=2.8.0
 contrib.root.dir=contrib
 lib.dir=lib

project/build/KafkaProject.scala

@@ -191,6 +191,9 @@ class KafkaProject(info: ProjectInfo) extends ParentProject(info) with IdeaProje
          <dependency org="org.apache.hadoop" name="hadoop-core" rev="0.20.2">
            <exclude module="junit"/>
          </dependency>
+         <dependency org="org.apache.pig" name="pig" rev="0.8.0">
+           <exclude module="junit"/>
+         </dependency>
        </dependencies>
 
     }
@@ -211,6 +214,9 @@ class KafkaProject(info: ProjectInfo) extends ParentProject(info) with IdeaProje
          <dependency org="org.apache.hadoop" name="hadoop-core" rev="0.20.2">
            <exclude module="junit"/>
          </dependency>
+         <dependency org="org.apache.pig" name="pig" rev="0.8.0">
+           <exclude module="junit"/>
+         </dependency>
        </dependencies>
 
       val jodaTime = "joda-time" % "joda-time" % "1.6"