客户端连接 阿里云kafka服务 报错（It was not possible to establish a connection for metadata with the brokers "kafka-cn-internet.aliyun.com:8080"）

[mgckid]

//连接代码 public function producer() { try { // Create the logger $logger = new Logger('my_logger'); // Now add some handlers $logger->pushHandler(new StdoutHandler());

        $config = \Kafka\ProducerConfig::getInstance();
        $config->setMetadataRefreshIntervalMs(10000);
        $config->setMetadataBrokerList($this->kafka_setting['bootstrap_servers']);
        $config->setBrokerVersion('1.0.0');
        $config->setRequiredAck(1);
        $config->setIsAsyn(false);
        $config->setProduceInterval(500);
        $config->setSecurityProtocol(Config::SECURITY_PROTOCOL_SASL_SSL);
        $config->setSaslMechanism(Config::SASL_MECHANISMS_PLAIN);
        $config->setSaslUsername($this->kafka_setting['sasl_plain_username']);
        $config->setSaslPassword($this->kafka_setting['sasl_plain_password']);

// $config->setSaslUsername($this->kafka_setting['sasl_plain_username']); // $config->setSaslPassword($this->kafka_setting['sasl_plain_password']); // $config->setSaslKeytab('/etc/security/keytabs/kafkaclient.keytab'); // $config->setSaslPrincipal('kafka/node1@NMREDKAFKA.COM'); // if use ssl connect $config->setSslLocalCert($this->kafka_setting['ca_file']); // $config->setSslLocalPk('/home/vagrant/code/kafka-php/ca-key'); // $config->setSslPassphrase($this->kafka_setting['sasl_plain_password']); // $config->setSslPeerName($this->kafka_setting['sasl_plain_username']);

        $producer = new \Kafka\Producer(function () {
            return [
                [
                    'topic' => $this->kafka_setting['topic_name'],
                    'value' => 'test....message.',
                    'key' => '',
                ],
            ];
        });
        $producer->setLogger($logger);
        $producer->success(function ($result) {
            var_dump($result);
        });
        $producer->error(function ($errorCode) {
            var_dump($errorCode);
        });
        $r = $producer->send(true);
        var_dump($r);

    } catch (Exception $e) {
        dump($e->getMessage());
    }
}

[nmred]

Please provide a detailed kafka server log

[mgckid]

@nmred 需要我提供哪些详细的日志，请列举下

[Veitor]

I also use Aliyun kafka and the error occurred is "Fatal error: Uncaught Kafka\Exception: Could not read 352518912 bytes from stream, length too longer. in D:\workspace\test\kafka\vendor\nmred\kafka-php\src\Kafka\SocketSync.php:292"

Does it means the metadata read from kafka server is too large?

[mgckid]

@Veitor 为什么你代码不需要配置ssal和ssl就能连接阿里云

[Veitor]

@mgckid 我当时也在想，为啥我什么认证机制都没配置，另外我是刚开始接触kafka，正巧阿里云最近在公测所以想试试，好像阿里云官方之前提供过php客户端demo但现在又没了。。比较蛋疼，kafka的php客户端实现貌似比其他语言繁琐啊

[mgckid]

@Veitor 我今天打算参照node的阿里云kafka demo 再试下连接阿里云看看，已经快绝望了，sasl认证老是过不去

const Kafka = require('node-rdkafka'); const config = require('./setting'); console.log("features:" + Kafka.features); console.log(Kafka.librdkafkaVersion);

var producer = new Kafka.Producer({ /'debug': 'all', / 'api.version.request': 'true', 'bootstrap.servers': config['bootstrap_servers'], 'dr_cb': true, 'dr_msg_cb': true, 'security.protocol' : 'sasl_ssl', 'ssl.ca.location' : './ca-cert', 'sasl.mechanisms' : 'PLAIN', 'sasl.username' : config['sasl_plain_username'], 'sasl.password' : config['sasl_plain_password'] });

[Veitor]

兄弟，你研究的怎么样了。。我已经搞了很久了。。阿里官方demo中都是用的第三方库啊，直接传传配置进去的。。要按照他们那样搞我还得去研究库。。不知道他们怎么想的。。。

我本地用docker搭建了个kafka，通过该代码运行producer似乎是成功的样子的

通过consumer返回的结果也与之前不一样了。。不过看不懂。貌似没取出消息来？

我很好奇这个php库是不是只能用于常规搭建的kafka，而用不了阿里云的？

[mgckid]

我使用官方demo里的python代码去连接阿里云kafka成功了

我上午在阿里云上提了一个gong工单，他们给我反馈了，说php不适合连接队列服务

[sharljimhtsin]

@mgckid 真相了。。。原来是阿里云那边限制了数量。。