From babf8f9691890aff5da13276293a833ef64cfa59 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Mr=C2=B7x?= <as2252258@163.com>
Date: Mon, 26 Oct 2020 17:25:24 +0800
Subject: [PATCH] =?UTF-8?q?=E6=94=B9=E5=90=8D?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 Kafka/Kafka.php  | 12 ++++++------
 Kafka/Struct.php |  3 +--
 2 files changed, 7 insertions(+), 8 deletions(-)

diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php
index 56129b91..dd06dde3 100644
--- a/Kafka/Kafka.php
+++ b/Kafka/Kafka.php
@@ -65,7 +65,7 @@ class Kafka extends \Snowflake\Process\Process
 			}
 			switch ($message->err) {
 				case RD_KAFKA_RESP_ERR_NO_ERROR:
-					$this->channel->push([$message->topic_name, $message->partition, $message]);
+					$this->channel->push([$message->topic_name, $message]);
 					break;
 				case RD_KAFKA_RESP_ERR__PARTITION_EOF:
 					$this->application->error('No more messages; will wait for more');
@@ -95,7 +95,7 @@ class Kafka extends \Snowflake\Process\Process
 						$group->done();
 					});
 					while ($messages = $this->channel->pop()) {
-						$this->handlerExecute($messages[0], $messages[1], $messages[2]);
+						$this->handlerExecute($messages[0], $messages[1]);
 					}
 				});
 			}
@@ -109,7 +109,7 @@ class Kafka extends \Snowflake\Process\Process
 	 * @param $part
 	 * @param $message
 	 */
-	protected function handlerExecute($topic, $part, $message)
+	protected function handlerExecute($topic, $message)
 	{
 		try {
 			$namespace = 'App\\Kafka\\' . ucfirst($topic) . 'Consumer';
@@ -120,7 +120,7 @@ class Kafka extends \Snowflake\Process\Process
 			if (!($class instanceof ConsumerInterface)) {
 				return;
 			}
-			$class->onHandler(new Struct($topic, $part, $message));
+			$class->onHandler(new Struct($topic, $message));
 		} catch (\Throwable $exception) {
 			$this->application->error($exception->getMessage());
 		}
@@ -140,8 +140,8 @@ class Kafka extends \Snowflake\Process\Process
 		$conf->set('group.id', uniqid('kafka'));
 		$conf->set('group.id', 'myConsumerGroup');
 
-		$conf->set('metadata.broker.list', '127.0.0.1:9092');
-//		$conf->set('metadata.broker.list', $kafka['brokers']);
+//		$conf->set('metadata.broker.list', '127.0.0.1:9092');
+		$conf->set('metadata.broker.list', $kafka['brokers']);
 		$conf->set('auto.offset.reset', 'earliest');
 
 		$conf->set('socket.timeout.ms', 300000);
diff --git a/Kafka/Struct.php b/Kafka/Struct.php
index 7d549605..d4978db9 100644
--- a/Kafka/Struct.php
+++ b/Kafka/Struct.php
@@ -20,10 +20,9 @@ class Struct
 	/**
 	 * Struct constructor.
 	 * @param $topic
-	 * @param $part
 	 * @param $message
 	 */
-	public function __construct($topic, $part,Message $message)
+	public function __construct($topic, Message $message)
 	{
 		$this->topic = $topic;
 		$this->offset = $message->offset;