apache · wyzssw · Sep 11, 2015 · Sep 11, 2015 · Sep 12, 2015 · Sep 12, 2015
diff --git a/core/src/main/scala/kafka/consumer/ConsumerConfig.scala b/core/src/main/scala/kafka/consumer/ConsumerConfig.scala
@@ -181,6 +181,10 @@ class ConsumerConfig private (val props: VerifiableProperties) extends ZKConfig(
   /** Select a strategy for assigning partitions to consumer streams. Possible values: range, roundrobin */
   val partitionAssignmentStrategy = props.getString("partition.assignment.strategy", DefaultPartitionAssignmentStrategy)
 
+
+  /** commit offset after consumed switch,default false*/
+  val commitAfterConsumed  = props.getBoolean("manual.commit.after.consumed",false);
+
   validate(this)
 }
 
diff --git a/core/src/main/scala/kafka/consumer/ConsumerIterator.scala b/core/src/main/scala/kafka/consumer/ConsumerIterator.scala
@@ -23,6 +23,13 @@ import kafka.serializer.Decoder
 import java.util.concurrent.atomic.AtomicReference
 import kafka.message.{MessageAndOffset, MessageAndMetadata}
 import kafka.common.{KafkaException, MessageSizeTooLargeException}
+import kafka.common.TopicAndPartition
+import kafka.utils.Pool
+import kafka.common.TopicAndPartition
+import kafka.common.TopicAndPartition
+import org.apache.kafka.common.TopicPartition
+import kafka.common.TopicAndPartition
+import kafka.common.TopicAndPartition
 
 
 /**
@@ -34,19 +41,26 @@ class ConsumerIterator[K, V](private val channel: BlockingQueue[FetchedDataChunk
                              consumerTimeoutMs: Int,
                              private val keyDecoder: Decoder[K],
                              private val valueDecoder: Decoder[V],
-                             val clientId: String)
+                             val clientId: String,val commitAfterConsumed: Boolean)
   extends IteratorTemplate[MessageAndMetadata[K, V]] with Logging {
 
   private var current: AtomicReference[Iterator[MessageAndOffset]] = new AtomicReference(null)
   private var currentTopicInfo: PartitionTopicInfo = null
   private var consumedOffset: Long = -1L
   private val consumerTopicStats = ConsumerTopicStatsRegistry.getConsumerTopicStat(clientId)
+  private val topicPartition2Info = new Pool[TopicAndPartition, PartitionTopicInfo]
+	private val iteratorOffsetMap = new Pool[TopicAndPartition, Long]
 
   override def next(): MessageAndMetadata[K, V] = {
     val item = super.next()
     if(consumedOffset < 0)
       throw new KafkaException("Offset returned by the message set is invalid %d".format(consumedOffset))
-    currentTopicInfo.resetConsumeOffset(consumedOffset)
+    if(!commitAfterConsumed){
+      currentTopicInfo.resetConsumeOffset(consumedOffset)
+    }else{
+      iteratorOffsetMap.put(TopicAndPartition(item.topic,item.partition), consumedOffset);
+    }
+    topicPartition2Info.put(TopicAndPartition(currentTopicInfo.topic,currentTopicInfo.partitionId), currentTopicInfo);
     val topic = currentTopicInfo.topic
     trace("Setting %s consumed offset to %d".format(topic, consumedOffset))
     consumerTopicStats.getConsumerTopicStats(topic).messageRate.mark()
@@ -77,9 +91,14 @@ class ConsumerIterator[K, V](private val channel: BlockingQueue[FetchedDataChunk
         val cdcFetchOffset = currentDataChunk.fetchOffset
         val ctiConsumeOffset = currentTopicInfo.getConsumeOffset
         if (ctiConsumeOffset < cdcFetchOffset) {
-          error("consumed offset: %d doesn't match fetch offset: %d for %s;\n Consumer may lose data"
-            .format(ctiConsumeOffset, cdcFetchOffset, currentTopicInfo))
-          currentTopicInfo.resetConsumeOffset(cdcFetchOffset)
+          if (!commitAfterConsumed) {
+            error("consumed offset: %d doesn't match fetch offset: %d for %s;\n Consumer may lose data"
+              .format(ctiConsumeOffset, cdcFetchOffset, currentTopicInfo))
+            currentTopicInfo.resetConsumeOffset(cdcFetchOffset)
+          } else {
+            debug("consumed offset: %d doesn't match fetch offset: %d for %s;\n "
+              .format(ctiConsumeOffset, cdcFetchOffset, currentTopicInfo))
+          }
         }
         localCurrent = currentDataChunk.messages.iterator
 
@@ -93,7 +112,12 @@ class ConsumerIterator[K, V](private val channel: BlockingQueue[FetchedDataChunk
     }
     var item = localCurrent.next()
     // reject the messages that have already been consumed
-    while (item.offset < currentTopicInfo.getConsumeOffset && localCurrent.hasNext) {
+
+     var iteratorOffset = iteratorOffsetMap.get(TopicAndPartition(currentTopicInfo.topic, currentTopicInfo.partitionId));
+     if(!commitAfterConsumed){
+          iteratorOffset  =  currentTopicInfo.getConsumeOffset;
+     }
+    while (item.offset < iteratorOffset && localCurrent.hasNext) {
       item = localCurrent.next()
     }
     consumedOffset = item.nextOffset
@@ -109,6 +133,19 @@ class ConsumerIterator[K, V](private val channel: BlockingQueue[FetchedDataChunk
       current.set(null)
     }
   }
+
+  def resetConsumeOffset(topic:String,partitionId:Int,offset:Long) {
+    if (commitAfterConsumed) {
+       val targetTopicPartitionInfo = topicPartition2Info.get(TopicAndPartition(topic,partitionId));
+       targetTopicPartitionInfo.resetConsumeOffset(offset);
+    } 
+  }
+
+  def resetConsumeOffset() {
+    if (!commitAfterConsumed)
+      currentTopicInfo.resetConsumeOffset(consumedOffset)
+   }
+
 }
 
 class ConsumerTimeoutException() extends RuntimeException()

diff --git a/core/src/main/scala/kafka/consumer/KafkaStream.scala b/core/src/main/scala/kafka/consumer/KafkaStream.scala
@@ -26,11 +26,11 @@ class KafkaStream[K,V](private val queue: BlockingQueue[FetchedDataChunk],
                         consumerTimeoutMs: Int,
                         private val keyDecoder: Decoder[K],
                         private val valueDecoder: Decoder[V],
-                        val clientId: String)
+                        val clientId: String,val commitAfterConsumed: Boolean)
    extends Iterable[MessageAndMetadata[K,V]] with java.lang.Iterable[MessageAndMetadata[K,V]] {
 
   private val iter: ConsumerIterator[K,V] =
-    new ConsumerIterator[K,V](queue, consumerTimeoutMs, keyDecoder, valueDecoder, clientId)
+    new ConsumerIterator[K,V](queue, consumerTimeoutMs, keyDecoder, valueDecoder, clientId,commitAfterConsumed);
 
   /**
    *  Create an iterator over messages in the stream.

diff --git a/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala b/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
@@ -230,7 +230,7 @@ private[kafka] class ZookeeperConsumerConnector(val config: ConsumerConfig,
       threadIdSet.map(_ => {
         val queue =  new LinkedBlockingQueue[FetchedDataChunk](config.queuedMaxMessages)
         val stream = new KafkaStream[K,V](
-          queue, config.consumerTimeoutMs, keyDecoder, valueDecoder, config.clientId)
+          queue, config.consumerTimeoutMs, keyDecoder, valueDecoder, config.clientId,config.commitAfterConsumed)
         (queue, stream)
       })
     ).flatten.toList
@@ -279,6 +279,7 @@ private[kafka] class ZookeeperConsumerConnector(val config: ConsumerConfig,
 
   def commitOffsetToZooKeeper(topicPartition: TopicAndPartition, offset: Long) {
     if (checkpointedZkOffsets.get(topicPartition) != offset) {
+      info("offset commiting " + topicPartition +" offset = "+ offset.toString());
       val topicDirs = new ZKGroupTopicDirs(config.groupId, topicPartition.topic)
       updatePersistentPath(zkClient, topicDirs.consumerOffsetDir + "/" + topicPartition.partition, offset.toString)
       checkpointedZkOffsets.put(topicPartition, offset)
@@ -922,7 +923,7 @@ private[kafka] class ZookeeperConsumerConnector(val config: ConsumerConfig,
                                           config.consumerTimeoutMs,
                                           keyDecoder,
                                           valueDecoder,
-                                          config.clientId)
+                                          config.clientId,config.commitAfterConsumed)
         (queue, stream)
     }).toList
 

diff --git a/core/src/test/scala/unit/kafka/consumer/ConsumerIteratorTest.scala b/core/src/test/scala/unit/kafka/consumer/ConsumerIteratorTest.scala
@@ -76,7 +76,8 @@ class ConsumerIteratorTest extends JUnit3Suite with KafkaServerTestHarness {
                                                     consumerConfig.consumerTimeoutMs,
                                                     new StringDecoder(), 
                                                     new StringDecoder(),
-                                                    clientId = "")
+                                                    clientId = "",
+                                                    false)
     val receivedMessages = (0 until 5).map(i => iter.next.message).toList
 
     assertFalse(iter.hasNext)
@@ -99,7 +100,8 @@ class ConsumerIteratorTest extends JUnit3Suite with KafkaServerTestHarness {
       ConsumerConfig.ConsumerTimeoutMs,
       new FailDecoder(),
       new FailDecoder(),
-      clientId = "")
+      clientId = "",
+      false)
 
     val receivedMessages = (0 until 5).map{ i =>
       assertTrue(iter.hasNext)

diff --git a/pom.xml b/pom.xml
@@ -0,0 +1,91 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <modelVersion>4.0.0</modelVersion>
+  <groupId>com.xiaoju.nova.strategy</groupId>
+  <artifactId>kafka_0.8.2.1_nova</artifactId>
+  <version>1.0.0-SNAPSHOT</version>
+  <description>POM was created from install:install-file</description>
+  <dependencies>
+    <dependency>
+      <groupId>org.objenesis</groupId>
+      <artifactId>objenesis</artifactId>
+      <version>1.2</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-log4j12</artifactId>
+      <version>1.7.6</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>4.1</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.yammer.metrics</groupId>
+      <artifactId>metrics-core</artifactId>
+      <version>2.2.0</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.scala-lang</groupId>
+      <artifactId>scala-library</artifactId>
+      <version>2.10.4</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.scalatest</groupId>
+      <artifactId>scalatest_2.10</artifactId>
+      <version>1.9.1</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.easymock</groupId>
+      <artifactId>easymock</artifactId>
+      <version>3.0</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-clients</artifactId>
+      <version>0.8.2.1</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.zookeeper</groupId>
+      <artifactId>zookeeper</artifactId>
+      <version>3.4.6</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>net.sf.jopt-simple</groupId>
+      <artifactId>jopt-simple</artifactId>
+      <version>3.2</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.101tec</groupId>
+      <artifactId>zkclient</artifactId>
+      <version>0.3</version>
+      <scope>compile</scope>
+    </dependency>
+</dependencies>
+
+   <distributionManagement>
+     <repository>
+      <id>nexus-releases-id</id>
+      <name>Nexus Release Repository</name>
+      <url>http://10.10.65.4:8081/nexus/content/repositories/releases/</url>
+    </repository>
+    <snapshotRepository>
+      <id>nexus-snapshots-id</id>
+      <name>Nexus Snapshot Repository</name>
+      <url>http://10.10.65.4:8081/nexus/content/repositories/snapshots/</url>
+    </snapshotRepository>
+  </distributionManagement>
+
+</project>