Read from kafka very slow!

[magnum-ferox]

More than 1 second for each message!
arnaud-lb/php-rdkafka read over 1000 messager per second!
Ping to server ~ 50ms

[makasim]

• What command do you use to consume messages?
• Do you consume from a single queue or several ones at once?
• What wait time do you use?

[makasim]

arnaud-lb/php-rdkafka read over 1000 messager per second!

@magnum-ferox Do you use low level or high level consumer?

[makasim]

Default consumer:

• Produce 1000 messages 0.057 sec
• Consume 1000 messages 60 sec

Consumer async commit true:

• Produce 1000 messages 0.037 sec
• Consume 1000 messages 2 sec

To enabled async commit do:

<?php
$context = (new RdKafkaConnectionFactory($config))->createContext();

$topic = $context->createTopic('a_topic');

$consumer = $context->createConsumer($topic);
$consumer->setCommitAsync(true);

[makasim]

Script to test:

<?php

use Enqueue\RdKafka\RdKafkaConnectionFactory;

include_once __DIR__.'/vendor/autoload.php';

$config = [
    'global' => [
        'group.id' => uniqid('', true),
        'metadata.broker.list' => getenv('RDKAFKA_HOST').':'.getenv('RDKAFKA_PORT'),
        'enable.auto.commit' => 'false',
    ],
    'topic' => [
        'auto.offset.reset' => 'beginning',
    ],
];

$context = (new RdKafkaConnectionFactory($config))->createContext();

sleep(3);


$topic = $context->createTopic(uniqid('', true));

$consumer = $context->createConsumer($topic);
$consumer->setCommitAsync(true);

$expectedBody = __CLASS__.time();

$time = microtime(true);
foreach (range(1, 1010) as $index) {
    $context->createProducer()->send($topic, $context->createMessage($index));
}
var_dump(microtime(true) - $time);

$index = 0;
$time = microtime(true);
while ($index < 1000) {
    if ($message = $consumer->receive(2000)) {
        $consumer->acknowledge($message);

        $index++;
    }
}

var_dump(microtime(true) - $time);

$context->close();

[magnum-ferox]

No. I try your script:
consume 20 messages without setCommitAsync(true) - double(16.388971090317)
consume 20 messages with setCommitAsync(true) - double(21.263849020004) !!!

and without $consumer->acknowledge($message);
20 messages without setCommitAsync(true) - double(15.642835140228)
20 messages with setCommitAsync(true) - double(15.63699388504)

librdkafka-0.11.1-1.el7.x86_64
php-pecl-rdkafka-3.0.5-1.el7.remi.7.1.x86_64
kafka_2.11-0.11.0.2
PHP 7.1.12

queue-interop/queue-interop (0.6.1)
enqueue/rdkafka (0.8.11)

ping to kafka server:
16 packets transmitted, 16 received, 0% packet loss, time 15014ms
rtt min/avg/max/mdev = 53.163/53.248/53.321/0.149 ms