backport #188 to v0.2

weiboad · Jan 31, 2018 · bbe9329 · bbe9329
1 parent e14311b
commit bbe9329
Showing 1 changed file with 7 additions and 5 deletions.
diff --git a/src/Consumer/Process.php b/src/Consumer/Process.php
@@ -557,21 +557,23 @@ public function succFetch($result, $fd)
                     continue;
                 }
 
-                $offset = $assign->getConsumerOffset($topic['topicName'], $part['partition']);
-                if ($offset === false) {
+				$consumerOffset = $assign->getConsumerOffset($topic['topicName'], $part['partition']);
+                if ($consumerOffset === false) {
                     return; // current is rejoin....
                 }
                 foreach ($part['messages'] as $message) {
                     $this->messages[$topic['topicName']][$part['partition']][] = $message;
                     //if ($this->consumer != null) {
                     //    call_user_func($this->consumer, $topic['topicName'], $part['partition'], $message);
                     //}
-                    $offset = $message['offset'];
+                    $commitOffset = $message['offset'];
                 }
 
-                $consumerOffset = ($part['highwaterMarkOffset'] > $offset) ? ($offset + 1) : $offset;
+				$commitOffset = isset($commitOffset) ? $commitOffset : $consumerOffset - 1;
+				$consumerOffset = $commitOffset + 1;
+
                 $assign->setConsumerOffset($topic['topicName'], $part['partition'], $consumerOffset);
-                $assign->setCommitOffset($topic['topicName'], $part['partition'], $offset);
+				$assign->setCommitOffset($topic['topicName'], $part['partition'], $commitOffset);
             }
         }
         $this->state->succRun(\Kafka\Consumer\State::REQUEST_FETCH, $fd);