as2252258/kiri-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

改名

Browse Source
This commit is contained in:
Mr·x
2020-10-09 11:10:07 +08:00
parent 44f86005a8
commit a73ba003db
1 changed files with 1 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Producer/SyncProcess.php
+1 -4
View File
@@ -113,15 +113,12 @@ class SyncProcess
$broker = \Kafka\Broker::getInstance(); $broker = \Kafka\Broker::getInstance();
foreach ($brokerHost as $host) { foreach ($brokerHost as $host) {
$socket = $broker->getMetaConnect($host, true); $socket = $broker->getMetaConnect($host, true);
var_dump($socket);
if ($socket) { if ($socket) {
$params = array(); $params = array();
$this->debug('Start sync metadata request params:' . json_encode($params)); $this->debug('Start sync metadata request params:' . json_encode($params));
$requestData = \Kafka\Protocol::encode(\Kafka\Protocol::METADATA_REQUEST, $params); $requestData = \Kafka\Protocol::encode(\Kafka\Protocol::METADATA_REQUEST, $params);
$socket->write($requestData); $socket->write($requestData);
$dataLen = \Kafka\Protocol\Protocol::unpack(\Kafka\Protocol\Protocol::BIT_B32, $socket->read(4)); $data = $socket->read(\Kafka\Protocol\Protocol::unpack(\Kafka\Protocol\Protocol::BIT_B32, $socket->read(4)));
$data = $socket->read($dataLen);
$correlationId = \Kafka\Protocol\Protocol::unpack(\Kafka\Protocol\Protocol::BIT_B32, substr($data, 0, 4));
$result = \Kafka\Protocol::decode(\Kafka\Protocol::METADATA_REQUEST, substr($data, 4)); $result = \Kafka\Protocol::decode(\Kafka\Protocol::METADATA_REQUEST, substr($data, 4));
if (!isset($result['brokers']) || !isset($result['topics'])) { if (!isset($result['brokers']) || !isset($result['topics'])) {
throw new \Kafka\Exception('Get metadata is fail, brokers or topics is null.'); throw new \Kafka\Exception('Get metadata is fail, brokers or topics is null.');