Allowed memory exhausted after consuming for a period of time

[itzJustKranker]

Not sure if this is a bug or if im doing something wrong, so i'll just describe what im trying to do and maybe someone will know whether or not im doing something blatantly wrong with the approach.

i setup a pretty simple consumer inside an artisan command

$consumer = Kafka::consumer(['topic'])
    ->withSasl(
        config('kafka.sasl.username'),
        config('kafka.sasl.password'),
        config('kafka.sasl.mechanisms'),
        config('kafka.securityProtocol')
    )
    ->withHandler(function (ConsumedMessage $message) {
        $body = $message->getBody();
        event(new Event($body));
        $this->info('Received message: '.json_encode($body));
    })->build();

$consumer->consume();

and this seems to work just fine for a period of time, my consumer reads from the topic, pushes the events, continues consuming. so on and so forth.

After a while i get a memory exhaustion like the following

PHP Fatal error: Allowed memory size of 134217728 bytes exhausted (tried to allocate 20480 bytes) in /local-path/vendor/laravel/framework/src/Illuminate/Collections/Arr.php on line 213

I don't believe this is related to the messages themselves as the data is pretty slim, i.e. a couple numeric/string fields. I think this may be related to the sheer volume of data im trying to consume.

For reference, the topic im consuming from receives maybe 10K messages every minute or so. Probably 1m+ messages a day. Is there a better way to consume from a topic such as this, or is there some configuration issue i missed?

Can this library handle the scale described?

Thanks in advance, its been very easy to work with the library so far.

[mateusjunges]

hey @itzJustKranker 👋 thanks for reporting. I'll take a look and get back to you.

Do you have an example of messages you are consuming? I'd like to reproduce this as close to your use case as possible. No need to share real data, just the structure would be fine

[mateusjunges]

Also, do you know how long it takes from starting the consumer until it crashes?

[itzJustKranker]

@mateusjunges This would be an example of the data

{
  "ts": 1732121246028,
  "vin": "6L3BPCFG839TXXXXX",
  "logicalevent_id": 1,
  "event_id": 1,
  "device_id": "UOOIN3WHXXXXX",
  "data": {
    "value": "5800"
  },
  "state": null,
  "status": null,
  "value": "58"
}

I tried running the consumer again 3 times for 5 mins each time and I did not experience the memory overflow. I'm not sure what caused it the first couple of times. If I see it again I will take note of what was going on and report back. Thanks for your time!