Forwarding messages from the UDP server to Kafka

[pbromb]

Hi,
I need to write a high performance UDP server to send messages to a Kafka server and handle acknowledgements to receive 1 million messages per second.

For each incoming message I create a task

$serv->on('Task', function (Swoole\Server $serv, $task_id, $reactor_id, $data) {
echo "#{$serv->worker_id}\tonTask: [PID={$serv->worker_pid}]: task_id=$task_id, data_len=".strlen($data).".".PHP_EOL;

  $conf = new \RdKafka\Conf();
  $conf->set('metadata.broker.list', '127.0.0.1:9092');
  
  $conf->setDrMsgCb(function ($producer, $msg) {
      echo "Send response via UDP..."  . PHP_EOL;
  });
  
  $producer = new \RdKafka\Producer($conf);
  
  $topic = $producer->newTopic('test');
  $topic->producev(RD_KAFKA_PARTITION_UA, 0, $data['payload']);
  $producer->poll(0); // non-blockig
  
  $serv->finish($data);
});

But my callback setDrMsgCb, will not be called because the task will finish faster.

I can use the blocking function
$producer->poll(1000);
However, this is very costly as it requires running thousands of tasker processes.

Any idea how to handle confirmations efficiently so as not to block threads ? Separate process to handle Kafka ?

[leocavalcante]

Have you tried finihsing the task at the delivery callback?

$conf->setDrMsgCb(function ($producer, $msg) {
    echo "Send response via UDP..."  . PHP_EOL;
    $serv->finish($data);
});

[pbromb]

I do not want to wait for confirmation of a single message in a task process. because it is inefficient.

The response from Kafka for 1 message takes 10 ms and for 1000 messages 15 ms for all messages.

This is because librdkafka has its own internal queue and sending a large batch of messages at once is much more efficient than sending 1 message at a time and waiting for an acknowledgement.

It seems to me that I should make a separate process for sending messages to Kafka. But I have never done something like that in swoole