Skip to content

Commit 6501059

Browse files
YurunsoftYurunsoft
authored
Fix problems with multiple brokers (#24)
* Fix problems with multiple brokers rejoin when return MEMBER_ID_REQUIRED * Add muilt broker test * Fix * Fix test * Fix SwooleClient * Fix test * Add kafka 2.7 test * Fix test
1 parent e4c1d81 commit 6501059

File tree

14 files changed

+209
-61
lines changed

14 files changed

+209
-61
lines changed

‎.github/docker-compose.yml‎

Lines changed: 24 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -6,41 +6,43 @@ services:
66
ports:
77
- "2181:2181"
88

9-
kafka-1.0.0:
10-
container_name: kafka
11-
image: wurstmeister/kafka:1.0.0
9+
kafka1:
10+
container_name: kafka1
11+
image: wurstmeister/kafka:${KAFKA_VERSION}
1212
depends_on:
1313
- zookeeper
1414
environment:
15+
KAFKA_BROKER_ID: 1
16+
KAFKA_ADVERTISED_HOST_NAME: kafka1
1517
KAFKA_ADVERTISED_PORT: 9092
18+
KAFKA_HOST_NAME: kafka1
19+
KAFKA_PORT: 9092
1620
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
1721
HOSTNAME_COMMAND: hostname -i
22+
KAFKA_LISTENERS: PLAINTEXT://kafka1:9092
23+
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092
24+
KAFKA_NUM_PARTITIONS: 3
1825
ports:
1926
- "9092:9092"
2027

21-
kafka-1.1.1:
22-
container_name: kafka
23-
image: wurstmeister/kafka:2.11-1.1.1
28+
kafka2:
29+
container_name: kafka2
30+
image: wurstmeister/kafka:${KAFKA_VERSION}
2431
depends_on:
2532
- zookeeper
2633
environment:
27-
KAFKA_ADVERTISED_PORT: 9092
34+
KAFKA_BROKER_ID: 2
35+
KAFKA_ADVERTISED_HOST_NAME: kafka2
36+
KAFKA_ADVERTISED_PORT: 9093
37+
KAFKA_HOST_NAME: kafka2
38+
KAFKA_PORT: 9093
2839
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
2940
HOSTNAME_COMMAND: hostname -i
41+
KAFKA_LISTENERS: PLAINTEXT://kafka2:9093
42+
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka2:9093
43+
KAFKA_NUM_PARTITIONS: 3
3044
ports:
31-
- "9092:9092"
32-
33-
kafka-2.6.0:
34-
container_name: kafka
35-
image: wurstmeister/kafka:2.13-2.6.0
36-
depends_on:
37-
- zookeeper
38-
environment:
39-
KAFKA_ADVERTISED_PORT: 9092
40-
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
41-
HOSTNAME_COMMAND: hostname -i
42-
ports:
43-
- "9092:9092"
45+
- "9093:9093"
4446

4547
swoole:
4648
container_name: "swoole"
@@ -52,6 +54,8 @@ services:
5254
PHP_VERSION: ${PHP_VERSION}
5355
depends_on:
5456
- zookeeper
57+
- kafka1
58+
- kafka2
5559
volumes:
5660
- "${GITHUB_WORKSPACE}:/kafka-client:rw"
5761
working_dir: /kafka-client

‎.github/workflows/ci.yml‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -10,13 +10,13 @@ jobs:
1010
fail-fast: false
1111
matrix:
1212
php: [7.1, 7.2, 7.3, 7.4, "8.0"]
13-
kafka: [1.0.0, 1.1.1, 2.6.0]
13+
kafka: [1.0.0, 2.11-1.1.1, 2.13-2.6.0, 2.13-2.7.0]
1414

1515
env:
1616
PHP_VERSION: ${{ matrix.php }}
1717
KAFKA_VERSION: ${{ matrix.kafka }}
18-
SWOOLE_VERSION: 4.5.9
19-
KAFKA_HOST: kafka-${{ matrix.kafka }}
18+
SWOOLE_VERSION: 4.5
19+
KAFKA_HOST: kafka1
2020
KAFKA_PORT: 9092
2121

2222
steps:
@@ -25,16 +25,16 @@ jobs:
2525
- name: start docker
2626
run: |
2727
cd .github
28-
# kafka
29-
docker-compose up -d kafka-$KAFKA_VERSION
28+
# run
29+
docker-compose up -d
3030
# swoole
31-
docker-compose up -d swoole && docker exec swoole php -v && docker exec swoole php --ri swoole && docker exec swoole composer -V
31+
docker exec swoole php -v && docker exec swoole php --ri swoole && docker exec swoole composer -V
3232
docker ps -a && docker ps
3333
3434
- name: prepare
3535
run: |
3636
docker exec swoole composer update
37-
docker exec kafka /opt/kafka/bin/kafka-topics.sh --zookeeper zookeeper:2181 --create --partitions 3 --replication-factor 1 --topic test
37+
docker exec kafka1 /opt/kafka/bin/kafka-topics.sh --zookeeper zookeeper:2181 --create --partitions 3 --replication-factor 1 --topic test
3838
3939
- name: php-test
4040
run: docker exec -e KAFKA_HOST="$KAFKA_HOST" -e KAFKA_PORT="$KAFKA_PORT" -e KAFKA_VERSION="$KAFKA_VERSION" swoole composer test

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,3 +9,4 @@ vendor/
99
*.log
1010
/coverage.txt
1111
/.php_cs.cache
12+
*.cache

‎src/Broker.php‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,11 @@ class Broker
3636
*/
3737
protected $topicsMeta;
3838

39+
/**
40+
* @var string[]
41+
*/
42+
protected $metaUpdatedTopics = [];
43+
3944
public function __construct($config)
4045
{
4146
$this->config = $config;
@@ -95,7 +100,17 @@ public function updateMetadata(array $topics = [], ?ClientInterface $client = nu
95100
$request->setAllowAutoTopicCreation($config->getAutoCreateTopic());
96101
/** @var MetadataResponse $response */
97102
$response = $client->sendRecv($request);
98-
$this->topicsMeta = $response->getTopics();
103+
$topicsMeta = $response->getTopics();
104+
if ($this->topicsMeta) {
105+
$this->topicsMeta = array_values(array_merge($this->topicsMeta, $topicsMeta));
106+
} else {
107+
$this->topicsMeta = $topicsMeta;
108+
}
109+
if ($this->metaUpdatedTopics) {
110+
$this->metaUpdatedTopics = array_values(array_merge($this->metaUpdatedTopics, $topics));
111+
} else {
112+
$this->metaUpdatedTopics = $topics;
113+
}
99114

100115
return $response;
101116
}
@@ -171,6 +186,9 @@ public function getTopicsMeta(): array
171186

172187
public function getBrokerIdByTopic(string $topic, int $partition): ?int
173188
{
189+
if (!\in_array($topic, $this->metaUpdatedTopics)) {
190+
$this->updateMetadata([$topic]);
191+
}
174192
foreach ($this->topicsMeta as $topicMeta) {
175193
if ($topicMeta->getName() === $topic) {
176194
foreach ($topicMeta->getPartitions() as $topicPartition) {

‎src/Client/SwooleClient.php‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
use Exception;
88
use InvalidArgumentException;
99
use longlang\phpkafka\Config\CommonConfig;
10+
use longlang\phpkafka\Exception\SocketException;
1011
use longlang\phpkafka\Protocol\AbstractRequest;
1112
use longlang\phpkafka\Protocol\AbstractResponse;
1213
use longlang\phpkafka\Protocol\ApiKeys;
@@ -49,14 +50,10 @@ public function connect(): void
4950

5051
public function close(): bool
5152
{
52-
if ($this->socket->close()) {
53-
$this->connected = false;
54-
$this->recvChannels = [];
53+
$this->connected = false;
54+
$this->recvChannels = [];
5555

56-
return true;
57-
} else {
58-
return false;
59-
}
56+
return $this->socket->close();
6057
}
6158

6259
/**
@@ -141,6 +138,9 @@ private function startRecvCo()
141138
$this->recvChannels[$correlationId]->push($data);
142139
}
143140
} catch (Exception $e) {
141+
if ($e instanceof SocketException && !$this->connected) {
142+
return;
143+
}
144144
$callback = $this->getConfig()->getExceptionCallback();
145145
if ($callback) {
146146
$callback($e);

‎src/Consumer/Consumer.php‎

Lines changed: 55 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@
66

77
use InvalidArgumentException;
88
use longlang\phpkafka\Broker;
9-
use longlang\phpkafka\Client\ClientInterface;
109
use longlang\phpkafka\Consumer\Assignor\PartitionAssignorInterface;
1110
use longlang\phpkafka\Consumer\Struct\ConsumerGroupMemberMetadata;
1211
use longlang\phpkafka\Exception\KafkaErrorException;
@@ -19,6 +18,7 @@
1918
use longlang\phpkafka\Protocol\Fetch\FetchPartition;
2019
use longlang\phpkafka\Protocol\Fetch\FetchRequest;
2120
use longlang\phpkafka\Protocol\Fetch\FetchResponse;
21+
use longlang\phpkafka\Protocol\FindCoordinator\FindCoordinatorResponse;
2222
use longlang\phpkafka\Protocol\JoinGroup\JoinGroupRequestProtocol;
2323
use longlang\phpkafka\Util\KafkaUtil;
2424
use Swoole\Timer;
@@ -50,11 +50,6 @@ class Consumer
5050
*/
5151
protected $offsetManagers = [];
5252

53-
/**
54-
* @var ClientInterface
55-
*/
56-
protected $client;
57-
5853
/**
5954
* @var string
6055
*/
@@ -100,6 +95,13 @@ class Consumer
10095
*/
10196
private $assignor;
10297

98+
/**
99+
* @var FindCoordinatorResponse
100+
*/
101+
private $coordinator;
102+
103+
protected $fetchOptions = [];
104+
103105
public function __construct(ConsumerConfig $config, ?callable $consumeCallback = null)
104106
{
105107
$this->config = $config;
@@ -112,14 +114,13 @@ public function __construct(ConsumerConfig $config, ?callable $consumeCallback =
112114
$broker->setBrokers($config->getBroker());
113115
}
114116

115-
$this->client = $broker->getClient();
116117
$this->groupManager = $groupManager = new GroupManager($broker);
117118
$groupId = $config->getGroupId();
118119

119120
$this->broker->updateMetadata($config->getTopic());
120121

121122
// findCoordinator
122-
$groupManager->findCoordinator($groupId, CoordinatorType::GROUP, $config->getGroupRetry(), $config->getGroupRetrySleep());
123+
$this->coordinator = $groupManager->findCoordinator($groupId, CoordinatorType::GROUP, $config->getGroupRetry(), $config->getGroupRetrySleep());
123124

124125
$this->rejoin();
125126
}
@@ -133,7 +134,6 @@ public function rejoin()
133134
$groupManager = $this->groupManager;
134135
$groupId = $config->getGroupId();
135136
$topics = $config->getTopic();
136-
$client = $this->broker->getClient();
137137

138138
$metadata = new ConsumerGroupMemberMetadata();
139139
$metadata->setTopics($config->getTopic());
@@ -165,6 +165,9 @@ public function rejoin()
165165
$consumerGroupMemberAssignment->unpack($data);
166166
}
167167

168+
$this->initFetchOptions();
169+
170+
$client = $this->broker->getClient($this->coordinator->getNodeId());
168171
foreach ($topics as $topic) {
169172
$this->offsetManagers[$topic] = $offsetManager = new OffsetManager($client, $topic, $this->getPartitions($topic), $groupId, $config->getGroupInstanceId(), $this->memberId, $this->generationId);
170173
$offsetManager->updateOffsets($config->getOffsetRetry());
@@ -234,6 +237,37 @@ public function ack(ConsumeMessage $message)
234237
$offsetManager->saveOffsets($partition, $this->config->getOffsetRetry());
235238
}
236239

240+
protected function initFetchOptions()
241+
{
242+
$fetchOptions = [];
243+
$config = $this->config;
244+
$broker = $this->broker;
245+
$topicsMeta = $broker->getTopicsMeta();
246+
foreach ($config->getTopic() as $topic) {
247+
$currentTopicMetaItem = null;
248+
foreach ($topicsMeta as $topicMetaItem) {
249+
if ($topicMetaItem->getName() === $topic) {
250+
$currentTopicMetaItem = $topicMetaItem;
251+
break;
252+
}
253+
}
254+
if (!$currentTopicMetaItem) {
255+
continue;
256+
}
257+
foreach ($this->getFetchPartitions($topic) as $partition) {
258+
foreach ($currentTopicMetaItem->getPartitions() as $topicsMetaItemPartition) {
259+
if ($partition === $topicsMetaItemPartition->getPartitionIndex()) {
260+
foreach ($topicsMetaItemPartition->getReplicaNodes() as $nodeId) {
261+
$fetchOptions[$nodeId][$topic][] = $partition;
262+
}
263+
break;
264+
}
265+
}
266+
}
267+
}
268+
$this->fetchOptions = $fetchOptions;
269+
}
270+
237271
protected function fetchMessages()
238272
{
239273
if (!$this->swooleHeartbeat) {
@@ -250,17 +284,23 @@ protected function fetchMessages()
250284
}
251285
$request->setRackId($config->getRackId());
252286
$topics = [];
253-
foreach ($config->getTopic() as $topic) {
287+
$currentList = current($this->fetchOptions);
288+
if (false === $currentList) {
289+
$currentList = reset($this->fetchOptions);
290+
}
291+
$nodeId = key($this->fetchOptions);
292+
next($this->fetchOptions);
293+
foreach ($currentList as $topic => $partitions) {
254294
$fetchPartitions = [];
255-
foreach ($this->getFetchPartitions($topic) as $partition) {
295+
foreach ($partitions as $partition) {
256296
$fetchPartitions[] = (new FetchPartition())->setPartitionIndex($partition)->setFetchOffset($this->getOffsetManager($topic)->getFetchOffset($partition));
257297
}
258298
$topics[] = (new FetchableTopic())->setName($topic)->setFetchPartitions($fetchPartitions);
259299
}
260300
$request->setTopics($topics);
261301

262302
/** @var FetchResponse $response */
263-
$response = $this->client->sendRecv($request);
303+
$response = $this->broker->getClient($nodeId)->sendRecv($request);
264304
$errorCode = $response->getErrorCode();
265305
switch ($errorCode) {
266306
case ErrorCode::REBALANCE_IN_PROGRESS:
@@ -348,6 +388,9 @@ protected function getPartitions(string $topic): array
348388
return $partitions;
349389
}
350390

391+
/**
392+
* @return int[]
393+
*/
351394
protected function getFetchPartitions(string $topic): array
352395
{
353396
$partitions = [];

‎src/Group/GroupManager.php‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66

77
use longlang\phpkafka\Broker;
88
use longlang\phpkafka\Client\ClientInterface;
9+
use longlang\phpkafka\Protocol\ErrorCode;
910
use longlang\phpkafka\Protocol\FindCoordinator\FindCoordinatorRequest;
1011
use longlang\phpkafka\Protocol\FindCoordinator\FindCoordinatorResponse;
1112
use longlang\phpkafka\Protocol\Heartbeat\HeartbeatRequest;
@@ -52,10 +53,7 @@ public function findCoordinator(string $key, int $keyType = CoordinatorType::GRO
5253
$request->setKey($key);
5354
$request->setKeyType($keyType);
5455

55-
/** @var FindCoordinatorResponse $response */
56-
$this->findCoordinatorResponse = KafkaUtil::retry($this->broker->getClient(), $request, $retry, $sleep);
57-
58-
return $this->findCoordinatorResponse;
56+
return $this->findCoordinatorResponse = KafkaUtil::retry($this->broker->getClient(), $request, $retry, $sleep);
5957
}
6058

6159
public function joinGroup(string $groupId, string $memberId, string $protocolType, ?string $groupInstanceId = null, array $protocols = [], int $sessionTimeoutMs = 60000, int $rebalanceTimeoutMs = -1, int $retry = 0, float $sleep = 0.01): JoinGroupResponse
@@ -70,7 +68,13 @@ public function joinGroup(string $groupId, string $memberId, string $protocolTyp
7068
$request->setRebalanceTimeoutMs($rebalanceTimeoutMs);
7169

7270
/** @var JoinGroupResponse $response */
73-
$response = $this->joinGroupResponse = KafkaUtil::retry($this->broker->getClient($this->findCoordinatorResponse->getNodeId()), $request, $retry, $sleep);
71+
$response = $this->joinGroupResponse = KafkaUtil::retry($this->broker->getClient($this->findCoordinatorResponse->getNodeId()), $request, $retry, $sleep, [
72+
ErrorCode::MEMBER_ID_REQUIRED => function (JoinGroupResponse $response) use ($request, $retry, $sleep) {
73+
$request->setMemberId($response->getMemberId());
74+
75+
return KafkaUtil::retry($this->broker->getClient($this->findCoordinatorResponse->getNodeId()), $request, $retry, $sleep);
76+
},
77+
]);
7478

7579
$this->isLeader = $response->getLeader() === $response->getMemberId();
7680

‎src/Producer/Producer.php‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -123,9 +123,7 @@ public function sendBatch(array $messages, ?int $brokerId = null)
123123
$value = $message->getValue();
124124
$key = $message->getKey();
125125
$partitionIndex = $message->getPartitionIndex() ?? $this->partitioner->partition($topicName, $value, $key, $topicsMeta);
126-
if (null === $brokerId) {
127-
$brokerId = $broker->getBrokerIdByTopic($topicName, $partitionIndex);
128-
}
126+
$brokerId = $broker->getBrokerIdByTopic($topicName, $partitionIndex);
129127
if (isset($topicsMap[$brokerId][$topicName])) {
130128
/** @var TopicProduceData $topicData */
131129
$topicData = $topicsMap[$brokerId][$topicName];

‎src/Socket/SwooleSocket.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,7 @@ public function recv(int $length, ?float $timeout = null): string
128128
$leftTime = $timeout;
129129
while ($this->socket && !isset($this->receivedBuffer[$length - 1]) && (-1 == $timeout || $leftTime > 0)) {
130130
$buffer = $this->socket->recv($timeout);
131-
if ($buffer === '' || $buffer === false) {
131+
if ('' === $buffer || false === $buffer) {
132132
throw new SocketException(sprintf('Could not recv data from stream, %s [%d]', $this->socket->errMsg, $this->socket->errCode));
133133
}
134134
$this->receivedBuffer .= $buffer;

0 commit comments

Comments
 (0)