Making flush optional, in order to be able to add it only when the laravel application shuts down

[sash]

Basis for the idea: Flushing should be done only before the php application finishes to ensure any unfinished async work is actually done before killing the php process. Currently the library does that after each send or batchSend no matter what - that limits the performance that could be achieved for project relying heavily on publishing events on different topics.

Solution: Allow for sending messages w/out flushing

Here is my idea step by step

1. Have producer->produce and produceBatch NOT do flush implicidly (or perhaps have it as an optional argument for backward compat.
2. Create a laravel service provider that will make a singleton for a generic producer for the default configured brokers

$this->app->singleton(\Junges\Kafka\Contracts\MessageProducer::class, function() {
    return Kafka::publish();
});

3. Have a shutdown listener that will always flush before the application terminates

public function boot()
{
    $this->app->terminating(function () {
       $this->app->make(\Junges\Kafka\Contracts\MessageProducer::class)->flush();
    });
}

4. Then in your regular code you can dependency inject the singleton and then use a new method (or a parammed old one) that will send a message w/out flushing

public function requestHandler(Request $request, MessageProducer $producer)
{
    $producer->onTopic('example')->withKafkaKey('foo')->withBodyKey('foo', 'bar')->asyncSend();
}

NB: In order for this to work we will also have to persist in the MessageProducer the built instance of the Junges\Kafka\Producers\Producer class that is currently being built on every call of send (as it calls ->build), but that I think is a general performance improvement not having to create an instance for Junges\Kafka\Producers\Producer for each send.

What do you guys think?

P.S. We are working on incorporating your wanderful kafka library into our laravel stack (replacing another kafka library), so I also want to thank you for your work thus far!

[sash]

Another side-effect of this approach is that probably by the time shutdown happens, because laravel has to do other still there too, there will be nothing left to flush anyways and this way there will be virtually no time penalty to the request being handled.

The ultimate goal here is to minimize the miliseconds that publishing to kafka will cost to the request being handled. Using the queue to publish the events would cause lag and we want to be as close to real-time as possible with out kafka production!

[sash]

I did a test implementation of my idea in order to benchmark it:

$time = Benchmark::measure(function() use (&$logs, $producer){ $producer->withKafkaKey(UuidV4::uuid2(1))
            ->withBodyKey('foo', 'bar')
            ->withBodyKey('time', now())

            ->send();
        }, 10);

time: 3.5404458 ms (average for the 10)

$time = Benchmark::measure(function() use (&$logs, $producer){ $producer->withKafkaKey(UuidV4::uuid2(1))
            ->withBodyKey('foo', 'bar')
            ->withBodyKey('time', now())

            ->asyncSend();
        }, 10);

time: 1.0999582 ms

I think there is merit to this idea! Let me know If you want a MR

[sash]

If we do 1000 its even more noticable (average in ms):
->send(): 2.241173689 ms, time for test command to complete: 4.237s
->asyncSend(): 0.276011574 ms, time for test command to complete: 2.103s

[sash]

One more thought I had. This change practically makes batch send obsolete since the difference b/n send and sendBatch was the flush was done at the end of the batch send, and in asyncSend its not done at all, so asyncSend and batchSend should be equally fast!