Adding a possibility to notify consumers

[hubcio]

The usual workflow is polling until I get messages, if I want low latency I have to poll very often, but if my consumer wants to sleep because there is no activity, it's not possible, it would be useful to have the option to notify consumers (not sending messages, just a notification in a tcp stream per example), so the consumer can start polling again.

discord discussion started here : discord.com/channels/1144142576266530928/1144142577369628684/1477810937405898883

From: #2851

[Luishfs]

@hubcio How would one start to implement this?

[hubcio]

we don't know yet. that's the reason for starting the discussion.

[hubcio]

Just to add: this functionality will allow to implement actual blocking poll, where poll interval wouldn't really matter - consumers could hang indefinitely on wait_for_notification() and wake up once they get notification from the server, and then they can call actual poll_messages().

[numinnex]

Kinda late contribution, but I would like to keep this discussion alive, as I have idea how this potentially could be implemented in a way where it's extensible by the end-user.

I think we could turn this mechanism into an "general purpose" notification mechanism. The way I imagine this could work is as follows:

• We would expose through some embeddable technology (like WASM), certain structures of our server in readonly mode
• The WASM module would output opaque bytes constructed based on the data structures that we provide
• The user would provide both the WASM module to create those bytes aswell how to deserialize those

This way we don't have to create our selves all of the possible notifications that users would ask for, instead it's on user to implement whatever notification they wish.

[hubcio]

i checked other message streaming platforms. in Kafka they have consumer.poll(Duration), in redis there is XREAD BLOCK, in NATS there is NextMsg(timeout), this is recurring pattern. i think we need to have similar API, not some WASM-like libraries built by users - these are neat, but I cannot imagine how would that look in code, since all traffic has to go through one TCP connection. Also, from users perspective it would be nightmare to use. I think we can do better.

after analyzing our architecture (both the current single node server and the new VSR we're building), i think the right approach is deferred-response polling - essentially what kafka, redis, and NATS all do.

the idea is simple: add wait_timeout_ms to PollMessages. value 0 = current behavior (return immediately). value > 0 = "hold my request until data arrives or timeout expires." one parameter, every SDK already has poll_messages() implemented - this is a single field addition per language. old clients that don't send it get default 0, fully backward compatible.

why not push notifications? push (server sends unsolicited frames to client) would require splitting every connection handler into separate reader/writer tasks, new subscribe/unsubscribe commands, frame type discriminator in the wire protocol (breaking change), and every foreign SDK would need a frame demultiplexer with background reader task. that's weeks of work per SDK for the same wake latency. deferred-response gives sub-millisecond wakeup with zero new concepts for users.

how it works server-side: the connection handler is not the message pump. each TCP connection has its own handler task, the pump processes ShardFrames sequentially via channels. we don't block the pump - we "park" the poll. when poll_messages with wait_timeout > 0 arrives and partition has no new data, the shard stores a lightweight wakeup handle in a per-partition wait registry. handler returns from process_frame immediately, pump is not blocked and continues processing other frames. when SendMessages commits to that partition, the pump sends a wakeup signal to waiting consumers (one atomic CAS per consumer, natural coalescing). the parked poll completes, client receives data. if timeout expires first, client gets an empty response (not an error, this is normal).

this fits the new VSR architecture well - the notification fire point is inside commit_messages() after partition.offset.store(committed_offset). consumers only get woken after data has achieved quorum, never after prepare. view change just lets pending polls expire naturally. the response IS the data - no race between "notification arrives" and "poll for data" that you'd get with push notifications in a consumer group scenario.

as for the server internals, when we build the new connection handler we should design it with split reader/writer tasks from day one - not for push notifications, but because it eliminates per-request channel allocation and the select! overhead in the read path. this makes adding push notification support later (for cluster events, rebalance signals, etc.) trivial - just one more message type on the writer task's channel. but the user-facing API stays poll_messages(wait_timeout_ms) regardless.

thoughts? @numinnex @spetz

[numinnex]

Yeah poll_messages /w timeout can definitely solve this problem (atleast partially), but the point I tried to make is that notifications could be used for different purposes, for example notifying when consumer/consumers fall really far behind and build massive backlog (this could be reacted by the user of the SDK, by creating additional consumers).

I proposed to make it in an extensible way, because there would be more use-cases like this and we wouldn't be able to cover all of those on our own.

Also I kinda feel like your solution doesn't necessary solve the problem at hand, because what they user wanted was a notification, not long polling, this gives the user of the SDK higher degree of freedom as what they want to do with it (in most cases it would be using poll_messages API).