Problem when batch processing kafka messages

[andrerigon]

Hi,

I'm trying to process batches of kafka messages using quarkus/smallrye. From time to time, I get:

 SRMSG00201: Error caught while processing a message: io.smallrye.mutiny.subscription.BackPressureFailure: Cannot emit item due to lack of requests
bidder-athena-69b47f97fb-58w6c bidder-athena    at io.smallrye.mutiny.operators.multi.MultiBufferWithTimeoutOp$MultiBufferWithTimeoutProcessor.flushCallback(MultiBufferWithTimeoutOp.java:162)
bidder-athena-69b47f97fb-58w6c bidder-athena    at io.smallrye.mutiny.operators.multi.MultiBufferWithTimeoutOp$MultiBufferWithTimeoutProcessor.onItem(MultiBufferWithTimeoutOp.java:193)

We have a consumer that receives a Multi<Message> similar to:

@Broadcast
@Incoming(FORECAST_UPDATES_IN)
@Outgoing(FORECAST_UPDATES_ACK)
@OnOverflow(value = OnOverflow.Strategy.DROP)
fun persistEvents(messages: Multi<Message<ForecastUpdate>>): Multi<Message<ForecastUpdate>> = messages.groupBy(batchSize, maxDelay)
    .emitOn(executor)
    .onOverflow()
    .drop()
    .flatMap { list ->
        // slow blocking operation
        Multi.createFrom().items(list.stream())
    }

And a consumer to ack those messages:

@Incoming(FORECAST_UPDATES_ACK)
@OnOverflow(value = OnOverflow.Strategy.DROP)
fun messageAcknowledging(message: Message<ForecastUpdate>): Uni<Void> {
    log.debug("Ack for message: ${message.payload}")
    return Uni.createFrom().completionStage(message.ack())
}

Any suggestion?

Best,

Andre

[quarkus-bot[bot]]

/cc @cescoffier, @evanchooly, @ozangunalp

[ozangunalp]

I can imagine this could happen during a shutdown. Can you correlate this to a scale down or restart in dev mode?

[andrerigon]

Hi @ozangunalp. Sorry ffor the delay. We endedup changing the feature and I can test it with the original scenario anymore.