Enable back pressure for KafkaConsumer

[blindspotbounty]

We performed some basic tests for consumer but with big number of records >11M

for try await record in consumer.messages {
    i = record.offset
    ctr += 1

    if ctr % 1000 == 0 {
        print("read up to \(i), ctr: \(ctr)")
    }
}

Unfortunately, without back pressure it leads to memory growth. For simplicity attaching screenshot:

At some point, when all entries are read from kafka, memory usage is declining but would be nice if it would be possible to limit it with configuration (e.g. NIO HighLow watermark backpressure strategy).

PS: it seems that back pressure was removed around here #66

[blindspotbounty]

Just to compare that with even simple back pressure implementation (nio HighLowWatermark), memory go high:

[FranzBusch]

Definitely something we have to look into. We initially started with backpressure around poll but poll expects to be called on the configured interval which kinda defeats backpressure. I still haven't found the right APIs on rdkafka that we can use so that we can poll for events that are not messages on the interval but only poll for messages when we are not backpressured

[felixschlegel]

Since librdkafka is maintaining its own queue with fetched messages and we just read from that, applying back pressure to the NIOAsyncSequence in our code will not stop librdkafka from fetching more messages.

However, what we can do is find an elegant way to wrap queued.min.messages and queued.max.messages.kbytes (see CONFIGURATION.md to allow for some back pressure.

Edit: this does not really seem to solve the problem since it seems like large amounts of data somewhere else. Also: fetch.max.bytes might be interesting to look into

[FranzBusch]

We should figure out what data is getting queued here and where it is queued. Is it inside rdkafka or inside SwiftKafka? My assumption was that the messages are actually buffered in our AsyncSequence and that's why rdkafka keeps on producing more.

[blindspotbounty]

Hi @felixschlegel and @FranzBusch, sorry for making you wait for the answer.
This case is related to swift-kafka-client and particularly to unbound AsyncSequence. Though, librdkafka have its own queue, it will stop requesting new data when it is full. Since it is configurable, it is possible to limit this queue.
Sorry that I missed providing the code, but basically I've used NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark(lowWatermark: 50, highWatermark: 100) as a backpressure strategy. Plus added increased interval between polls (up to max.poll.interval.ms) in case queue is full. Since NIOAsyncSequence allows to produce even if it returns .stopProducing but with up to 5 minutes interval (default value of max.poll.interval.ms).
For sure, that is not perfect backpressure as we cannot avoid polling time to time but still make a lot of sense at application recovery when a lot of messages should be read.

[FranzBusch]

Okay I think I got to the bottom of this and now what to do now. So the problem is that we are calling rd_kafka_poll_set_consumer which directs the main poll queue to the consumer poll queue. The main poll queue is normally used for events such as rebalance, statistics, etc. That queue must be called on regular intervals to serve the events. However those events are not really driven by the consumer.
On the other hand, the consumer queue is driven by the consumer and must exert back pressure. Since we are interleaving the two right now we cannot exert back pressure anymore.

The solution to this is to not redirect the main queue to the consumer queue anymore and then tie polling the consumer queue to the backpressure from our NIOAsyncSequenceProducer. When this producer exerts back pressure we still need to continue to poll the other queue. @felixschlegel Do you have time to look into this?

[blindspotbounty]

@FranzBusch thank you for the brief excerpt. Sorry that I didn't provide it that way.
That sounds absolutely correct and dividing those queues should solve most of the problems except that anyway we should poll at least once per max.poll.interval.ms in consumer queue at least for one message to avoid silent client stop or provide some exception/error to downstream application (e.g. related to #110).

[felixschlegel]

Working on it!