[Bug]: Large custom producer settings caused stall on consume

[digikata]

What happened
Producing to fluvio with large record batch settings (10 MiB) can cause stalls on consume.

Sample config below:

async fn get_fluvio_producer(fluvio: &Fluvio, topic_name: &str) -> Result<fluvio::TopicProducer> {
    const FACTOR_1M: usize = 1024 * 1024;
    const PRODUCE_BATCH_SIZE_BYTES: usize = 10 * FACTOR_1M;
    const PRODUCE_TIMEOUT_MS: u64 = 5 * 60_000;
    const PRODUCE_LINGER_MS: u64 = 1_000;
    const PRODUCE_BATCH_N_QUEUE: usize = 2;
        
    let pconfig = fluvio::TopicProducerConfigBuilder::default()
        .batch_size(PRODUCE_BATCH_SIZE_BYTES)
        .batch_queue_size(PRODUCE_BATCH_N_QUEUE)
        .compression(Compression::Gzip)
        .linger(time::Duration::from_millis(PRODUCE_LINGER_MS))
        .timeout(time::Duration::from_millis(PRODUCE_TIMEOUT_MS))
        .build()?;
    fluvio.topic_producer_with_config(topic_name, pconfig).await
}

On consume, the output can stall after a few hundred lines:

fluvio consume TOPIC -d -B

-d should detach after all records are returned
-B specifies start from beginning of topic

Expected behavior
Producing records with large record batch settings should consume with stall.

Describe the setup
fluvio version 0.11.9
a local fluvio cluster
37Million records, ~164 bytes each (it's unknown if the number of records plays a part in the stall).

How to reproduce it (as minimally and precisely as possible)

Environment (please complete the following information):
macos
Darwin Kernel Version 23.5.0

Workaround

Lowering batch size to 1M or back to default also works for this case.

[digikata]

Some tooling that might be useful to reproduce this issue https://github.com/infinyon/fluvio-community/tree/main/sql2fluvio

[github-actions]

Stale issue message