Message loss when using asyncPublish()

[et-hoangdv]

Hi, I'm new to Kafka so I might be missing something fundamental — any help would be appreciated!
I’m currently testing message publishing and I’ve set up a basic command that sends 100 messages in a loop like this:

public function handle()
{
    for ($i = 0; $i < 100; $i++) {
        $log = [
            'message'   => "message #{$i}",
            'created_at' => now()->toIso8601String()
        ];

        $message = new Message(
            body: $log,
        );

        Kafka::asyncPublish()
            ->onTopic('laravel-logs')
            ->withMessage($message)
            ->send();
    }
}

Then I created two consumers:

• One for the main topic
• One for the dead letter topic

Main topic consumer:

public function handle()
{
    $consumer = Kafka::consumer(['laravel-logs'])
        ->withDlq()
        ->withHandler(function(\Junges\Kafka\Contracts\ConsumerMessage $message, \Junges\Kafka\Contracts\MessageConsumer $consumer) {
            $this->info(json_encode($message->getBody()));
        })
        ->build();

    try {
        $consumer->consume();
    } catch (Exception|ConsumerException $e) {
        $this->error($e->getMessage());
    }
}

Dead-letter topic consumer:

public function handle()
{
    $consumer = Kafka::consumer(['laravel-logs-dlq'])
        ->withHandler(function(\Junges\Kafka\Contracts\ConsumerMessage $message, \Junges\Kafka\Contracts\MessageConsumer $consumer) {
            $this->info("dead letter: " . json_encode($message->getBody()));
        })
        ->build();

    try {
        $consumer->consume();
    } catch (Exception|ConsumerException $e) {
        $this->error($e->getMessage());
    }
}

Problem:

When I run the producer command, the consumer sometimes receives all 100 messages, but often only receives around 80 or so.
Meanwhile, the dead-letter consumer does not show any message.

Questions:

1. Is this expected behavior when using asyncPublish()?
2. Is there a recommended way to detect or log failed deliveries when using asyncPublish()?
3. Could message loss be caused by Kafka config (e.g., acks, retries, etc.) that I’m not setting?

Thanks a lot for any guidance!

[et-hoangdv]

After some testing, I realized that the issue might be related to the use of dd('done') right after the asyncPublish() loop.
I think dd() stops PHP execution immediately, which might interrupt the async dispatch before the messages are actually sent.
In real-world cases, would using asyncPublish() risk losing messages if the script ends too quickly?

public function handle()
{
    for ($i = 0; $i < 100; $i++) {
        $log = [
            'message'   => "message #{$i}",
            'created_at' => now()->toIso8601String()
        ];

        $message = new Message(
            body: $log,
        );

        Kafka::asyncPublish()
            ->onTopic('laravel-logs')
            ->withMessage($message)
            ->send();
    }
    dd('done'); // I added this in my code
}