# Handle Slow Consumers (/learn/events/how-to/handle-slow-consumers)



When a subscriber's receive buffer is full, KubeMQ waits up to the **write deadline** before dropping the event. Understanding this behavior is critical for building reliable event-driven systems.

## How Message Drops Happen [#how-message-drops-happen]

<Mermaid
  chart="sequenceDiagram
    participant P as Publisher
    participant K as KubeMQ
    participant S as Slow Subscriber
    P->>K: publish event 1
    K->>S: deliver event 1
    Note over S: Processing slowly...
    P->>K: publish event 2
    K->>S: deliver event 2
    Note over S: Buffer filling up...
    P->>K: publish event 3
    K--xS: buffer full, wait writeDeadline
    Note over K: writeDeadline (2s) expires
    Note over K: Event 3 DROPPED for this subscriber"
/>

*The drop lifecycle: KubeMQ holds an event for the write deadline, then drops it for any subscriber whose buffer never clears.*

### The Write Deadline [#the-write-deadline]

* Default: **2000 milliseconds** (2 seconds)
* When a subscriber's buffer is full, KubeMQ waits this duration for space
* If the buffer does not clear in time, the event is **dropped** for that subscriber
* A warning is logged server-side with the channel, event ID, and metadata
* Other subscribers are not affected

## Mitigation Strategies [#mitigation-strategies]

### Strategy 1: Process Events Quickly [#strategy-1-process-events-quickly]

Offload slow work to a background worker pool. Keep the event callback fast.

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="fast_callback.go"
    workCh := make(chan []byte, 1000)

    go func() {
        for body := range workCh {
            processOrder(body) // slow work happens here
        }
    }()

    sub, err := client.SubscribeToEvents(ctx, "order-events", "",
        kubemq.WithOnEvent(func(event *kubemq.Event) {
            select {
            case workCh <- event.Body:
            default:
                log.Println("Local buffer full, dropping event")
            }
        }),
        kubemq.WithOnError(func(err error) {
            log.Println("Error:", err)
        }),
    )
    ```
  </Tab>

  <Tab value="Python">
    ```python title="fast_callback.py"
    import queue
    import threading

    work_queue = queue.Queue(maxsize=1000)

    def worker():
        while True:
            body = work_queue.get()
            process_order(body)  # slow work happens here
            work_queue.task_done()

    threading.Thread(target=worker, daemon=True).start()

    def on_event(event):
        try:
            work_queue.put_nowait(event.body)
        except queue.Full:
            print("Local buffer full, dropping event")

    client.subscribe_to_events(
        subscription=EventsSubscription(
            channel="order-events",
            on_receive_event_callback=on_event,
            on_error_callback=lambda e: print(f"Error: {e}"),
        ),
        cancel=CancellationToken(),
    )
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript title="fast_callback.js"
    const workQueue = [];
    const MAX_BUFFER = 1000;

    setInterval(() => {
      while (workQueue.length > 0) {
        const body = workQueue.shift();
        processOrder(body); // slow work
      }
    }, 10);

    client.subscribeToEvents({
      channel: "order-events",
      onEvent: (msg) => {
        if (workQueue.length >= MAX_BUFFER) {
          console.warn("Local buffer full, dropping event");
          return;
        }
        workQueue.push(msg.body);
      },
      onError: (err) => console.error("Error:", err.message),
    });
    ```
  </Tab>

  <Tab value="Java">
    ```java title="FastCallback.java"
    ExecutorService executor = Executors.newFixedThreadPool(4);
    BlockingQueue<byte[]> workQueue = new LinkedBlockingQueue<>(1000);

    executor.submit(() -> {
        while (true) {
            byte[] body = workQueue.take();
            processOrder(body); // slow work
        }
    });

    client.subscribeToEvents(EventsSubscription.builder()
        .channel("order-events")
        .onReceiveEventCallback(event -> {
            if (!workQueue.offer(event.getBody())) {
                System.err.println("Local buffer full, dropping event");
            }
        })
        .onErrorCallback(err ->
            System.err.println("Error: " + err.getMessage()))
        .build());
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="FastCallback.cs"
    var workQueue = Channel.CreateBounded<byte[]>(1000);

    _ = Task.Run(async () =>
    {
        await foreach (var body in workQueue.Reader.ReadAllAsync())
        {
            ProcessOrder(body); // slow work
        }
    });

    await foreach (var msg in client.SubscribeToEventsAsync(
        new EventsSubscription { Channel = "order-events" }))
    {
        if (!workQueue.Writer.TryWrite(msg.Body.ToArray()))
        {
            Console.Error.WriteLine("Local buffer full, dropping event");
        }
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="FastCallback.kt"
    val workQueue = LinkedBlockingQueue<ByteArray>(1000)

    thread(isDaemon = true) {
        while (true) {
            val body = workQueue.take()
            processOrder(body) // slow work
        }
    }

    client.subscribeToEvents(
        channel = "order-events",
        onEvent = { event ->
            if (!workQueue.offer(event.body)) {
                System.err.println("Local buffer full, dropping event")
            }
        },
        onError = { err -> System.err.println("Error: ${err.message}") }
    )
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="fast_callback.cpp"
    std::queue<std::string> workQueue;
    std::mutex queueMutex;
    const size_t MAX_BUFFER = 1000;

    std::thread worker([&]() {
        while (true) {
            std::string body;
            {
                std::lock_guard<std::mutex> lock(queueMutex);
                if (workQueue.empty()) continue;
                body = workQueue.front();
                workQueue.pop();
            }
            processOrder(body); // slow work
        }
    });

    client.subscribeToEvents("order-events", "",
        [&](const kubemq::Event& event) {
            std::lock_guard<std::mutex> lock(queueMutex);
            if (workQueue.size() >= MAX_BUFFER) {
                std::cerr << "Local buffer full, dropping event" << std::endl;
                return;
            }
            workQueue.push(event.body);
        },
        [](const std::string& err) {
            std::cerr << "Error: " << err << std::endl;
        }
    );
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="fast_callback.rs"
    use kubemq::prelude::*;
    use kubemq::Subscription;
    use tokio::sync::mpsc;

    #[tokio::main]
    async fn main() -> kubemq::Result<()> {
        let client = KubemqClient::builder()
            .host("localhost")
            .port(50000)
            .build()
            .await?;

        // Bounded local buffer; a background task does the slow work
        let (tx, mut rx) = mpsc::channel::<Vec<u8>>(1000);

        tokio::spawn(async move {
            while let Some(body) = rx.recv().await {
                process_order(body).await; // slow work happens here
            }
        });

        let sub: Subscription = client
            .subscribe_to_events(
                "order-events",
                "",
                move |event| {
                    let tx = tx.clone();
                    Box::pin(async move {
                        // try_send returns immediately; never block the callback
                        if tx.try_send(event.body).is_err() {
                            eprintln!("Local buffer full, dropping event");
                        }
                    })
                },
                None,
            )
            .await?;

        tokio::signal::ctrl_c().await.ok();
        sub.unsubscribe().await;
        client.close().await?;
        Ok(())
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="fast_callback.rb"
    require 'kubemq'

    client = KubeMQ::PubSubClient.new(address: "localhost:50000", client_id: "order-subscriber")
    cancel = KubeMQ::CancellationToken.new

    # Bounded local buffer drained by a background worker thread
    work_queue = SizedQueue.new(1000)

    Thread.new do
      loop do
        body = work_queue.pop
        process_order(body) # slow work happens here
      end
    end

    sub = KubeMQ::PubSub::EventsSubscription.new(channel: "order-events")
    client.subscribe_to_events(sub, cancellation_token: cancel,
                               on_error: ->(e) { warn "Error: #{e.message}" }) do |event|
      # push(..., true) is non-blocking; raises when the buffer is full
      begin
        work_queue.push(event.body, true)
      rescue ThreadError
        warn "Local buffer full, dropping event"
      end
    end

    sleep
    cancel.cancel
    client.close
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="fast_callback.exs"
    {:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "order-subscriber")

    # Spawn a worker process; the callback only hands work off to it
    worker =
      spawn(fn ->
        Stream.repeatedly(fn ->
          receive do
            {:work, body} -> process_order(body) # slow work happens here
          end
        end)
        |> Stream.run()
      end)

    {:ok, sub} =
      KubeMQ.Client.subscribe_to_events(client, "order-events",
        on_event: fn event ->
          # Guard against an unbounded mailbox; drop when the worker falls behind
          {:message_queue_len, len} = Process.info(worker, :message_queue_len)

          if len >= 1000 do
            IO.puts("Local buffer full, dropping event")
          else
            send(worker, {:work, event.body})
          end
        end,
        on_error: fn err -> IO.puts("Error: #{err.message}") end
      )

    Process.sleep(:infinity)
    KubeMQ.Subscription.cancel(sub)
    KubeMQ.Client.close(client)
    ```
  </Tab>
</Tabs>

### Strategy 2: Use Consumer Groups [#strategy-2-use-consumer-groups]

Distribute the load across multiple subscribers so no single consumer is overwhelmed.

<Mermaid
  chart="graph LR
    P[&#x22;Publisher&#x22;]
    EV{{&#x22;Events channel<br/>order-events&#x22;}}
    W1[&#x22;Worker A<br/>group: processors&#x22;]
    W2[&#x22;Worker B<br/>group: processors&#x22;]
    W3[&#x22;Worker C<br/>group: processors&#x22;]

    P -- &#x22;high volume&#x22; --> EV
    EV -- &#x22;1/3&#x22; --> W1
    EV -- &#x22;1/3&#x22; --> W2
    EV -- &#x22;1/3&#x22; --> W3

    class EV events
    class P,W1,W2,W3 client"
/>

*A consumer group spreads the load: each event goes to just one worker, so no single subscriber is overwhelmed.*

See [Consumer Groups](/learn/events/tutorials/consumer-groups) and [Scale Subscribers](/learn/events/how-to/scale-subscribers) for implementation details.

### Strategy 3: Switch to Events Store [#strategy-3-switch-to-events-store]

If losing messages is unacceptable, use [Events Store](/learn/events-store) instead. Events Store provides persistence and replay, ensuring no messages are lost even when consumers are slow or temporarily offline.

## Decision Guide [#decision-guide]

| Situation                                   | Recommendation                   |
| ------------------------------------------- | -------------------------------- |
| Occasional slowdowns, some drops acceptable | Strategy 1: Buffer in app        |
| Consistent high volume                      | Strategy 2: Consumer groups      |
| Zero message loss required                  | Strategy 3: Events Store         |
| Latency-sensitive, best-effort delivery     | Strategy 1 + Strategy 2 combined |

<Callout type="warn">
  Slow consumer drops are **silent** from the subscriber's perspective. Monitor server logs for `writeDeadline` warnings to detect when drops occur.
</Callout>

## Related [#related]

* [Consumer Groups](/learn/events/tutorials/consumer-groups) for load-balanced delivery
* [Scale Subscribers](/learn/events/how-to/scale-subscribers) for horizontal scaling
* [Events Store](/learn/events-store) for guaranteed delivery