Connect to Kafka with Confluent Parallel Consumer

[bojanv55]

Description

Parallel Kafka Connector for https://github.com/confluentinc/parallel-consumer

In order to be able to consume messages in parallel from single partition (with possibility to lock on a key level), Parallel Consumer allows just that.

Repository name

quarkus-parallel-kafka

Short description

Parallel Kafka Connector

Repository Homepage URL

https://docs.quarkiverse.io/<REPOSITORY_NAME>/dev/

Repository Topics

• quarkus-extension

...

Team Members

Additional context

No response

[quarkus-bot]

/cc @alesj (kafka), @cescoffier (kafka), @ozangunalp (kafka)

[ozangunalp]

The Quarkus Messaging Kafka incoming channels do allow parallel processing using @Blocking(ordered = false) or @RunOnVirtualThread annotations.

AFAIK, the only bit missing from it is the key-level ordering. I'd very much like to add this feature to the connector.
It should not be very tough to add this.

[bojanv55]

@ozangunalp I am trying now to implement this, since I need key-level ordering. If I use the setup you proposed, and add some code to do the locking based on a key-level, how would kafka currently ack the records? (eg. if I'm processing 1,2,3,4,5 and I finished processing 5th but no 1,2,3,4 would it ack 5 and then I would potentially lose 1,2,3,4 records as unprocessed?) I guess acking part is missing if I would like to implement this in the user-code side? It has to go in the framework?

P.S. are there any plans about feature timeline or nothing like that exists?

[ozangunalp]

The "throttled" commit-strategy, which is the default in Quarkus Messaging, already handles messages that are acked out of order.

As for the key-level ordering, I've done some tests. I can have something functional with minimal overhead using decorators and pausable channels. But it isn't quite feature-complete. If you want to give it a try, I pushed my test here for you to check.

[bojanv55]

@ozangunalp thanks, I will try also your tests to check how it works. In the meanwhile I also created https://github.com/bojanv55/parallel-kafka-connector that is connector for parallel-consumer. It is basically 90% of current code copy-pasted with some changes to support parallel-consumer.
I tested it a bit seems working. Could I ask you also to check these 2 lines if they provide optimal integration with quarkus pipelines:

• is this the way to ack/nack messages in smallrye pipeline https://github.com/bojanv55/parallel-kafka-connector/blob/3c1bfa023ed16f0b9ff46c202e5cd5f03c87a18f/src/main/java/me/vukas/parallel/ParallelKafkaSource.java#L110
• is it OK to always return the same publisher for multiple "concurrency" consumers https://github.com/bojanv55/parallel-kafka-connector/blob/3c1bfa023ed16f0b9ff46c202e5cd5f03c87a18f/src/main/java/me/vukas/parallel/ParallelKafkaConnector.java#L77

[bojanv55]

@ozangunalp Just playing with the test you pushed. Seems I'm not able to block it on the key-level (probably doing something wrong in my code?)

This is the code that receives the messages

@io.smallrye.reactive.messaging.annotations.Blocking(ordered = false/*, value = "my-pool"*/)
    @Incoming("some-topic")
    //@Blocking
    public Uni<Void> sink(String word, IncomingKafkaRecordMetadata<String, String> metadata) {
        Log.info("Received user: " + word);
        return myEntityRepository.upsertCurrentProducer("Asd", "sdf", word);
    }

This is the code that saves to db eventually

Uni<Void> upsertCurrentProducerInternal(
            String x, String y, String producer) {

        String sql =
                "INSERT INTO MyEntity (id, field)\n" +
                        "SELECT 1 AS id, '"+producer+"' AS field\n" +
                        "FROM DUAL\n" +
                        "WHERE SLEEP(5) = 0\n" +
                        "ON DUPLICATE KEY UPDATE field = 'fff';";

        return sessionFactory.withTransaction(
                (session, tx) ->
                        session
                                .createNativeQuery(sql)
                                .executeUpdate()
                                .replaceWithVoid());
    }

I would expect that when I send 5 messages with the same key at the same time - since the key is same and SQL is sleeping for 5 seconds - it should not process other messages with same key until 5 seconds passed. But it automatically consumes all other messages.

This is my config

smallrye.messaging.worker.my-pool.max-concurrency=50
mp.messaging.incoming.some-topic.connector=smallrye-kafka
mp.messaging.incoming.some-topic.commit-strategy=throttled
mp.messaging.incoming.some-topic.throttled.processing-order=ordered_by_key
mp.messaging.incoming.some-topic.auto.offset.reset=latest
mp.messaging.incoming.some-topic.group.id=grupaTrotled
mp.messaging.incoming.some-topic.key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
mp.messaging.incoming.some-topic.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

Do you have any advice here? Thanks.

[bojanv55]

@ozangunalp After checking the code a bit more - I guess this is only implemented and working in your tests?

[ozangunalp]

I guess this is only implemented and working in your tests?

That's true. Let me push something that can work OOTB with your code.

[ozangunalp]

You can check it out again.

[bojanv55]

@ozangunalp you sure you pushed? I cannot find anything.

[ozangunalp]

Here : https://github.com/ozangunalp/smallrye-reactive-messaging/commits/kafka_throttled_concurrency_ordered/

Note that it needs some more work. And I am thinking we could even generalize this to other connectors.

[bojanv55]

@ozangunalp Locking seems working now with my example. Will test it a bit more. For now I can see only that when I emit a bit more messages in sequence, I get this exception

2025-10-29 18:08:50,922 WARN  [io.sma.rea.mes.kafka] (vert.x-eventloop-thread-3) SRMSG18231: The record 132 from topic-partition 'some-0' has waited for 62 seconds to be acknowledged. This waiting time is greater than the configured threshold (60000 ms). At the moment 423 messages from this partition are awaiting acknowledgement. The last committed offset for this partition was 131. This error is due to a potential issue in the application which does not acknowledged the records in a timely fashion. The connector cannot commit as a record processing has not completed.
2025-10-29 18:08:50,922 WARN  [io.sma.rea.mes.kafka] (vert.x-eventloop-thread-3) SRMSG18228: A failure has been reported for Kafka topics '[some]': io.smallrye.reactive.messaging.kafka.commit.KafkaThrottledLatestProcessedCommit$TooManyMessagesWithoutAckException: The record 132 from topic/partition 'some-0' has waited for 62 seconds to be acknowledged. At the moment 423 messages from this partition are awaiting acknowledgement. The last committed offset for this partition was 131.

2025-10-29 18:08:50,925 INFO  [io.sma.rea.mes.kafka] (smallrye-kafka-consumer-thread-0) SRMSG18224: Executing consumer revoked re-balance listener for group 'grupaTrotled'

Just watched this in wireshark - when I emit 50 elements (starting from 5) it ACKs only 3 (5, 48, 55) but it acks all that it needs to and it seems to do that pretty slow, but at least it doesn't timeout. When I emit 500 elements (starting from 55) it ACKs only 3 and it doesn't ACK all needed ones (should be up to 555) again (57, 107, 132) and then it gets stuck and times-out.