Support sending messages in the background

[hubcio]

At 25.02.2025 there was interesting discussion on discord.

The discussion revolved around performance optimizations and feature enhancements in iggy compared to Kafka. Participants debated strategies such as sticky partitioning, batching, and compression, noting that intelligent client implementations have a significant impact on overall throughput and latency. They compared Kafka’s round-robin partitioning with sticky partitioning, emphasizing that filling one partition’s batch before moving to the next can greatly improve performance and cost efficiency by reducing network load.

The conversation also touched on API for sending messages in background and error handling during background sends. Participants discussed the merits of different failure handling strategies, such as blocking until a send succeeds, blocking with a timeout, or failing immediately. To address these options, they proposed introducing a new enum, IggyBackgroundSendFailureMode, which would allow clients to choose the desired behavior when a background send fails.

As a summary of this discussion, I propose task to support this API:

Create an extension for the IggyProducerBuilder that introduces configurable parameters for background sending.

Changes in existing fields:

• batch_size - should be changed to batch_length.
• send_interval - should be changed to linger_time, determines how long messages should remain in the queue before sending

Add new fields (optional?):

• background_send - determines if send in background functionality should be enabled
• batch_size - value (in bytes) of how much data should be buffered before sending
• maximum_buffer_size - maximum value (in bytes) of buffered data, if exceeded error will be reported (see IggyBackgroundSendFailureMode)
• failure_mode - determines behavior of send_messages when buffer is full (or in general on failure)?

Ensure that the new API integrates seamlessly with the current send_messages method, providing optional background sending.
Handle all cases with conflicting parameters set in builder, e.g. maximum_buffer_size should not be set when background_send is set to false.
Is is not certain how errors on send should be handled when buffer is not full. Perhaps error channel to communicate errors would be sufficient.

Proposal of enum:

/// Determines how the `send_messages` API should behave when problem is encountered
#[derive(Debug, Clone)]
pub enum IggyBackgroundSendFailureMode {
    /// Block until the send succeeds
    Block,
    /// Block with a timeout, after which the send fails
    BlockWithTimeout(IggyDuration),
    /// Fail immediately without retrying
    FailImmediately,
}

@spetz @numinnex what's your opinion on this?

[numinnex]

We need to make sure to put "sensible" defaults in the newly added fields in order to make the change backward compatible and not mutually exclusive with our existing batching strategy. I am not sure what does the Block and BlockTimeout in the enum mean ? would it retry sending the batch ? but what if the retries are disabled, just some questions.

[hubcio]

great idea

[haze518]

I like the in-flight mode, but it seems that it complements IggyBackgroundSendFailureMode rather than replacing it, since in-flight is more about the actual sending of messages. Please correct me if I’m wrong.
It looks like we still need to control the behavior when the max_in_flight limit is exceeded.
For example, if we implement this using a semaphore, the logic might look like this:

// read from channel
match failure_mode {
    Block           => permit = sem.acquire().await,
    Timeout(t)      => permit = timeout(t, sem.acquire()).await?,
    FailImmediately => permit = sem.try_acquire()?,
}
spawn(send_batch(permit, batch, delivery_timeout))

[hubcio]

I think you are correct. @spetz what's your opinion?

At this point, I think you can start the implementation and rest of the details should be discussed on discord :)

[spetz]

Looks good to me

[haze518]

ok :)

[spetz]

I think that background_send flag is not needed - instead, we could add a new method e.g. send_in_background (or something like this), which would push the messages to the background sender task via channel. By doing so, we would let the user to explicitly decide, which method to use for the given scenario.

[hubcio]

How will the underlying single producer take care of it? I will need an option to create multiple producers in this case.
you are free to create as many producers as you like :)

background_send is an additional flag in send method or, producer initialization?
producer initialization

IMHO method send_messages should stay the same, only producer configuration should be changed. @spetz do you still insist on having multiple methods?

[haze518]

@spetz, could you please check the thread. I just think that, as @hubcio said, the settings for a single producer should indeed be enough. If the user needs to send messages without using the background mode, they can simply create another producer

[spetz]

Looks like we need to decide once again, which option will be better :D

[haze518]

It seems to me that having a single producer, configured either for background or synchronous sending, is sufficient. The user is expected to decide on the intended usage at the moment of creation — and if both modes are needed, they can simply create an additional producer.

This approach is actually quite close to what I originally proposed here: #1593 (comment) — just without introducing separate methods. Personally, I still believe it would be more convenient if we explicitly split them into async/sync producers (e.g. via separate constructors or builder modes), as that would reduce cognitive load. We already have a fair number of configuration options, and this kind of clarity might help — though I understand we previously decided against that approach.

But since opinions vary, let's vote:
👍 — Keep a single send_messages method, use config flags (e.g. background_send)
🎉 — Add a second method like send_in_background
👀 — I don't have a strong opinion

[spetz]

Ok, sounds good, let's stick to the single producer, if we'd ever decide to add another method, it will be easy, as the underlying logic will not change at all :)

[MasteMind]

Why batch_length? batch_size seems intuitive enough.

[hubcio]

sometimes you might want to batch based on real size in bytes, sometimes you might want to batch based on length (numeric)

[MasteMind]

Numeric won't always be correct at scale. Mostly I might need to limit memory and use it to the fullest, based on my machine configuration.

[hubcio]

It will be option, so you use either one or another.

[MasteMind]

maximum_buffer_size signifies that there is a minimum_buffer_size somewhere too. Do we have plans to allow some range here?

If not, then just buffer_size seems intuitive enough again.

[hubcio]

what about pending_messages_buffer (bytes) and pending_messages_length (length)?

[MasteMind]

I think just buffer_size should be enough.

These can be derived metrics on top of it for monitoring.

[MasteMind]

linger_time might require me to search for the time unit being followed by this configuration.

linger_ms itself speaks for me to use milliseconds only.

[hubcio]

in iggy, we use IggyDuration which supports constructs like:
IggyDuration::from("1ms").unwrap()
or
IggyDuration::from("100us").unwrap()
so this is much more convinient.

[MasteMind]

Understood, makes sense

[MasteMind]

Even if background send is set to false, we might still need buffer_size. As there could be a back-pressure situation or some erroneous client implementation.

In these cases, client would need the capability to ensure these producers don't end up taking all the memory available. Buffer_size or buffer_memory should be set always with may be a default value of say 33 MB.

[hubcio]

I will think about it, good point.

[haze518]

@hubcio, did you get a chance to revisit @MasteMind point on using buffer_size even without background sending? It sounded like a sensible way to safeguard against client-side memory overuse

[hubcio]

for case when background sending is enabled, it soudns good, something like buffer_memory or memory_cap or memory_limit fits the bill (IggyByteSize).

for case when it's disabled, how would it map to actual code?

[haze518]

We could calculate the total size of the messages in the Vec passed to the send method before sending them (whether via background or immediately). If the total size exceeds a predefined threshold, we could return an error. This would also cover the case where a single message(via send_one) is sent synchronously, but it has a large payload — for example, a blob

[MasteMind]

"how errors on send should be handled when buffer is not full"

Errors for any sync send should be immediately communicated. These could be failures or timeouts.

Errors for any async send should have a callback method that can be triggered. Clients might need to log them, or send them to another dead-letter-queue, or may be do something custom with it.

[hubcio]

good point!

[hubcio]

@MasteMind @spetz @numinnex
Today/tomorrow I will create a github issue out of this discussion.

[haze518]

Hey! What do you think about an approach similar to the kafka sarama library https://github.com/IBM/sarama? There, this issue is solved not through method parameters or a separate method, but by having distinct producers: SyncProducer for synchronous sending and AsyncProducer, which works via a channel.

It seems like this approach is more convenient because it's immediately clear how you're sending messages, and there’s no confusion with parameters. Plus, error handling is also more straightforward: with the synchronous producer, errors are returned immediately, while with the asynchronous one, they come through a separate channel

[hubcio]

@haze518 how are you feeling about this feature? do you still want to do it? :)

[hubcio]

the big minus of this approach is that it's a breaking interface. TBH I'd prefer to ahve a single Producer which has two modes...

perhaps I'm wrong?

can you propose a interface for those two cases?

[spetz]

I also think that having an additional method purely responsible for sending the messages in the background (which could be as simple as pushing the records via channel to the actual sender doing all the heavy lifting), could be a better approach.

[haze518]

Makes sense, I’m fine with going this way. Let’s settle on that then! Could you tell me if we already have an issue for this feature, or should I create a new one? :)

[spetz]

This is the one #1600 :)

[haze518]

Aight, great — I’ll assign it to myself then :)

[hubcio]

@MasteMind if you are still here, please if such interface fits your needs:
#1838

[hubcio]

Closing the discussion because feature was merged under #1838.

Good job @haze518!