Hi, I really miss such a function, so I decided to try to implement it myself. I am waiting for an answer to how much this option is correct and whether it will be added to the package.

Hello,

I am getting error " Call to undefined method RdKafka\ProducerTopic::producev() " on creating producer. i have also installed librdkafka using command "$ apt install librdkafka-dev"

I am using below environment. PHP : 8.0.12 OS : Linux rdkafka : 5.0.0

Thanks, Dilip

@mateusjunges Hi i am trying to save data in database but not save in database.

<?php

namespace App\Handlers;

use Illuminate\Support\Facades\Log;
use Junges\Kafka\Contracts\KafkaConsumerMessage;
use App\Http\Controllers\TermsController;
use App\Model\Aquib;
use DB;

class TestHandler
{
    public function __invoke(KafkaConsumerMessage $message)
    {
        $values = array('ab' => 1);
       DB::table('test')->insert($values);

        
        $var = $message->getBody();
        Log::debug('Message received!', [
            'body' => $var,
            'headers' => $message->getHeaders(),
            'partition' => $message->getPartition(),
            'key' => $message->getKey(),
            'topic' => $message->getTopicName()
        ]);
        
        
    }
}

I'm calling the method stopAfterLastMessage() on the consumer building but the consumption don't stop.

I'm on ubuntu 22.04, running php 8.1 with rdkafka 6.0.3, laravel-kafka:1.8.1.

I check the code and in line 349 we have this: if ($this->config->shouldStopAfterLastMessage() && RD_KAFKA_RESP_ERR__PARTITION_EOF === $message->err) { RD_KAFKA_RESP_ERR__PARTITION_EOF equals -191 but $message->err is returning -185, i try to send a pull request but i get access denied, so i am sending the code that would be like this:

    private const CONSUMER_STOP_ERRORS = [
        RD_KAFKA_RESP_ERR__PARTITION_EOF,
        RD_KAFKA_RESP_ERR__TIMED_OUT
    ];
    
   ....
   
    if ($this->config->shouldStopAfterLastMessage() && in_array($message->err, self::CONSUMER_STOP_ERRORS)) {
        $this->stopConsume();
    }

Hi @mateusjunges , i have the same error as #105 but when creating a consumer :

$builder = Kafka::createConsumer($this->topics, $this->group, $this->host.':'.$this->port);
$builder->withHandler(function(KafkaConsumerMessage $message) {
	// Handle your message here
	echo $message->getBody();
	echo $message->getTopicName();
})
->stopAfterLastMessage()
->withMaxMessages(100);

$consumer = $builder->build();

$consumer->consume();

I'm using the following versions (on windows 11) :

laravel-kafka : 1.8.1 rdKafka : 6.0.1 librdkafka version (runtime) : 1.6.2 librdkafka version (build) :| 1.6.2.255

I tried to change the KAFKA_COMPRESSION_TYPE option to "none" or "gzip" without any success.

Note that by following the high level consumer exemple from rdkafka i am receiving message just fine.

Any ideas what am i missing ?

Thanks

@mateusjunges can you plz help me on this error is coming when am trying to send message to kafka. Here is my code,

Kafka::publishOn('broker', 'topic')
    ->withBodyKey('foo', 'bar')
    ->send();

return response('ok');

Hi, i'm trying to publish a message to an AWS MSK broker using SASL authentication with SCRAM mechanism, but i'm getting the message "Sent messages may not be completed yet." as response. Below you can see my code:

      $message = new Message(
            body: $request->get('message'),
        );

        $sasl = new Sasl(
            username: 'myuser',
            password: 'mypass',
            mechanisms: 'SCRAM-SHA-256'
        );

        try {
            Kafka::publishOn('test')
                ->withSasl($sasl)
                ->withMessage($message)
                ->send();
        } catch (CouldNotPublishMessage $exception) {
            return back()->with('error', $exception->getMessage());
        }

Am I missing something or there is some new feature needed to publish in AWS MSK?

Hi, I noticed a little problem with batch consuming messages.

Let's say I have a topic full of hundreds of messages, and I want to consume by batches of 20 messages. What happens : the consumer reads 20 messages, and then 1, then 20, then 1, then 20, 1, 20, 1, etc It would be better to try to have 20, 20, 20, 20, 20 etc (assuming we can get 20 before timeout of the batch).

The problem is here :

As there is a return, the timer isn't reset if we reach batch max size and the timer continue, so maybe the next time the function will be called the timer will have timeout already ...

Removing the return and add a elseif solves the problem : (Yes, it's the same code in both if... so it could be better, but you got the idea)

Something to fix for next release ?

Hi, I'm using laravel-opcache and laravel-octane-dockerfile with Laravel Octane. Before starting the server and after running php artisan compile command to pre-compile application code (includes vendor directory), the following error will be thrown:

 Symfony\Component\ErrorHandler\Error\FatalError 

  Cannot declare class App\Console\Kernel, because the name is already in use

  at vendor/mateusjunges/laravel-kafka/dev/laravel-console-kernel.php:0
      1▕ <?php
      2▕ 
      3▕ namespace App\Console;
      4▕ 
      5▕ use Illuminate\Console\Scheduling\Schedule;
      6▕ use Illuminate\Foundation\Console\Kernel as ConsoleKernel;
      7▕ 
      8▕ class Kernel extends ConsoleKernel
      9▕ {


   Whoops\Exception\ErrorException 

  Cannot declare class App\Console\Kernel, because the name is already in use

  at vendor/mateusjunges/laravel-kafka/dev/laravel-console-kernel.php:0
      1▕ <?php
      2▕ 
      3▕ namespace App\Console;
      4▕ 
      5▕ use Illuminate\Console\Scheduling\Schedule;
      6▕ use Illuminate\Foundation\Console\Kernel as ConsoleKernel;
      7▕ 
      8▕ class Kernel extends ConsoleKernel
      9▕ {

      +1 vendor frames 
  2   [internal]:0
      Whoops\Run::handleShutdown()

I tried to exclude vendor/mateusjunges/laravel-kafka/dev directory from pre-compiling in laravel-opcache package config, but the problem still exists.

This PR originally was to fix incorrect testing code in src/Support/Testing/Fakes/KafkaFake.php

Previously this code, the published method was running the callback against the $message variable that was actually passed in during the test definition. This made no sense as the filter ran on the same object everytime so wasn't actually performing a truthy check on the messages that were actually dispatched.

private function published(KafkaProducerMessage $message = null, $callback = null): Collection
{
    if (! $this->hasPublished()) {
        return collect();
    }

    $callback = $callback ?: function () {
        return true;
    };


    return collect($this->getPublishedMessages())->filter(function ($_, $topic) use ($message, $callback) {
        return $callback($message, $topic);
    });
}

I reworked this to perform the truthy test against the published messages and reworked the $message variable so that it can be used as an expected message. I also wrote a default callback to compare the data in the message. This allows you to write tests like

Kafka::fake();
Kafka::publishOn('topic')->withHeaders(['custom' => 'header'])->send();
Kafka::assertPublished((new Message)->withHeaders(['custom' => 'header']);

Without this change all tests pass regardless of what you pass in. I refactored all test assert methods to use this new style.

One thing i cannot figure out is why the publishedMessages is stored on the ProducerFake object. This object is recreated everytime a new message is published.

This means that if you run the following test always fails because it can only ever be as high as 1.

Kafka::fake();
Kafka::publishOn('topic')->withHeaders(['custom' => 'header'])->send();
Kafka::publishOn('topic')->withHeaders(['custom' => 'header2'])->send();
Kafka::assertPublishedOnTimes('topic', 2)

Hello,

First of all, thanks for the hard work with a very simple-to-use package! I've had a bit of an issue where some Kafka messages were getting lost. I was using the send method in your ProducerBuilder class:

public function send(): bool
{
    $producer = $this->build();

    return $producer->produce($this->message);
}

Since I was producing lots of messages, I changed my code to use the sendBatch method, which fixed my issue and everything worked smoothly after.

After that, I dug into the problem a bit more and I realized the flush method in the Producer class is using the Laravel retry logic:

private function flush(): mixed
{
    return retry(10, function () {
        $result = $this->producer->flush(1000);

        if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
            return true;
        }

        $message = rd_kafka_err2str($result);

        throw CouldNotPublishMessage::withMessage($message, $result);
    });
}

I was surprised since it seemed to me not as many messages should've been lost for me, even by sending them individually (with the send method I shared above) to Kafka with this nice retry logic when flushing. Then I realized, for the flush retry, the sleepMilliseconds parameter is not being set. So the code doesn't seem to be sleeping for any amount of time before retrying. I was just curious about why you took that decision. I was thinking it could be better to sleep for some milliseconds and maybe you could even retry less than 10 times. Being that this method is used for both, the send and sendBatch methods, I think it would be worth it.

Also, curious about why you are using the retry Larvel helper here, but not the retry method in the Retryable class you created. Wondering if the same retry method should be used everywhere in the package.

Thanks again for your work!

Describe the bug On throwing custom exceptions, the report() method of the exception is not called.

To Reproduce Steps to reproduce the behavior: 1- Have a fresh Larvel installation. 2- Make a custom exception using php artisan make:exception TestException, then do something in your report() method of the exception, for example, Log::info('some info'). 3- Create a consumer, then within the handler() of the consumer, throw the TestException.

Expected behavior By default, if a custom exception is thrown in Laravel, if it has a report() method, it's gonna get called. this works just fine outside of the consumer handle() method, and the report() method gets called.

Desktop (please complete the following information):

• OS: Ubuntu 21.04
• mateusjunges/laravel-kafka v1.10
• PHP 8.1

New major version

We now follows semantic versioning

Changes

• The \Junges\Kafka\Contracts\CanProduceMessages contract was renamed to \Junges\Kafka\Contracts\MessageProducer

• The \Junges\Kafka\Contracts\CanPublishMessagesToKafka contract was renamed to \Junges\Kafka\Contracts\MessagePublisher

• The \Junges\Kafka\Contracts\KafkaProducerMessage contract was renamed to \Junges\Kafka\Contracts\ProducerMessage

• The \Junges\Kafka\Contracts\CanConsumeMessages was renamed to \Junges\Kafka\Contracts\MessageConsumer

• The \Junges\Kafka\Contracts\KafkaConsumerMessage was renamed to \Junges\Kafka\Contracts\ConsumerMessage

• The \Junges\Kafka\Contracts\CanConsumeMessagesFromKafka was renamed to \Junges\Kafka\Contracts\ConsumeMessagesFromKafka

• The \Junges\Kafka\Contracts\CanConsumeBatchMessages contract was renamed to \Junges\Kafka\Contracts\BatchMessageConsumer

• The \Junges\Kafka\Contracts\CanConsumeMessages contract was renamed to \Junges\Kafka\Contracts\MessageConsumer

• Using strict types across the codebase

• Added a upgrade guide

• Upgrade the code to use PHP 8.1 features

• The withSasl method now accepts all SASL parameters instead of a Sasl object.

public function withSasl(string $username, string $password, string $mechanisms, string $securityProtocol = 'SASL_PLAINTEXT');

WIP

From what I could find whenever the broker is down, consumer will log the following messages:

%3|1665746426.651|FAIL|rdkafka#producer-2| [thrd:ssl://broker:9092/bootstrap]: ssl://broker:9092/bootstrap: Failed to resolve 'broker:9092': No address associated with hostname (after 3ms in state CONNECT) %3|1665746426.651|ERROR|rdkafka#producer-2| [thrd:ssl://broker:9092/bootstrap]: 1/1 brokers are down

Is there currently, somehow, a way to report or handle this error?

Thanks.

Hello!

I am trying to figure out if a healthcheck can be configured or added.

When I start the consumer in our kubernetes cluster it takes a solid 30 seconds for the consumer to start consuming and I want to see if I can add a healthcheck so I can keep the old consumers running until the new one can actually start consuming. This would prevent having message lag when updating our consumers

Is there an easy way to implement a healthcheck for this?

I have this code in a Laravel Command:

Kafka::createConsumer()
    ->subscribe($this->topic)
    ->withConsumerGroupId(config('kafka.consumer_group_id'))
    ->withHandler(fn (KafkaConsumerMessage $message) => $this->handler($handler, $message))
    ->stopAfterLastMessage()
    ->build()
    ->consume();

This command is executed in a cronjob every minute.

The problem is that never read any message from kafka, the pending queue offset is always in same position.

If I remove the stopAfterLastMessage option, the command will be executed without stop and then they can read the messages without problem.

Kafka::createConsumer()
    ->subscribe($this->topic)
    ->withConsumerGroupId(config('kafka.consumer_group_id'))
    ->withHandler(fn (KafkaConsumerMessage $message) => $this->handler($handler, $message))
    ->build()
    ->consume();

My config is:

<?php

return [
    /*
     | Your kafka brokers url.
     */
    'brokers' => env('KAFKA_BROKERS', 'localhost:9092'),

    /*
     | Kafka consumers belonging to the same consumer group share a group id.
     | The consumers in a group then divides the topic partitions as fairly amongst themselves as possible by
     | establishing that each partition is only consumed by a single consumer from the group.
     | This config defines the consumer group id you want to use for your project.
     */
    'consumer_group_id' => env('KAFKA_CONSUMER_GROUP_ID', 'group'),

    /*
     | After the consumer receives its assignment from the coordinator,
     | it must determine the initial position for each assigned partition.
     | When the group is first created, before any messages have been consumed, the position is set according to a configurable
     | offset reset policy (auto.offset.reset). Typically, consumption starts either at the earliest offset or the latest offset.
     | You can choose between "latest", "earliest" or "none".
     */
    'offset_reset' => env('KAFKA_OFFSET_RESET', 'earliest'),

    /*
     | If you set enable.auto.commit (which is the default), then the consumer will automatically commit offsets periodically at the
     | interval set by auto.commit.interval.ms.
     */
    'auto_commit' => env('KAFKA_AUTO_COMMIT', true),

    'sleep_on_error' => env('KAFKA_ERROR_SLEEP', 5),

    'partition' => env('KAFKA_PARTITION', -1),

    /*
     | Kafka supports 4 compression codecs: none , gzip , lz4 and snappy
     */
    'compression' => env('KAFKA_COMPRESSION_TYPE', 'snappy'),

    /*
     | Choose if debug is enabled or not.
     */
    'debug' => env('KAFKA_DEBUG', false),

    /*
     | Repository for batching messages together
     | Implement BatchRepositoryInterface to save batches in different storage
     */
    'batch_repository' => env('KAFKA_BATCH_REPOSITORY', \Junges\Kafka\BatchRepositories\InMemoryBatchRepository::class),
];

Any idea why can not read the pending queue messages when stopAfterLastMessage is enabled?

I can read the pending queue messages using the kafka command ./bin/kafka-console-consumer.sh --topic topic --bootstrap-server localhost:9092

Thanks!