Micronaut Kafka: multi threaded listeners not working as documented

[krickert]

Disclaimer: I believe I'm setting this up per the directions - but I'm posting it as a discussion to first validate that what I'm doing is setup as per the directions before calling it a bug. I hope it's not a bug, and I could easily just be doing something wrong.

Now here's my problem, and why I chose Micronaut Kafka listeners to do it:

problem
I want to create a app that can download a lot of files simultaneously. The order does not matter. The application gets a list of files it knows it needs, and there's a space where it should be saved to.

my solution
Send off all the file download requests to kafka - and have 3 threads of listeners running to download 3 at a time.

It works like this:
Producer:
creates 65 download requests, successfully go into kafka

Consumer:
take in a download request, and download 1 at a time

How I attempted to solve this with micronaut:

1. setup kafka
   I followed the instructions to install the confluent server with avro schema registry (this works fine, so does the serialization)

2. create a CLI producer app to send off the requests
   This also works fine. I go out to a website to get a list of all the files with their SHA1 sums, and then I send off all the messages to kafka server.

The topic is auto-created. It has one partition (I know I can split them off to three to solve this, more on that in a sec).

I produce the messages in a straightforward way:

@KafkaClient(id = "download-request-producer")
public interface DownloadRequestProducer {
    @Topic("download-request")
    void sendDownloadRequest(@KafkaKey UUID key, DownloadFileRequest request);

}

This creates the request without any issues and I can see the messages just fine in the front end of the confluent platform.

3. Create a messaging app and consume the messages
   Next, I found this little tidbit on the kafka documentation which got me excited, since spring has this feature and quarkus does not. Here it is:

Link to documentation
From the documentation, they suggest that I:

1. Define a fixed thread executor of size 25 in the config (see below)
2. Ensure that the listener I create is tagged as a @prototype instead of a @singleton, since each thread created will make their own instance
3. Mark the listener as having the number of desired threads

So here's my configured class for a listener:

package com.krickert.search.download.dump;

import com.krickert.search.download.dump.util.FileDownloader;
import com.krickert.search.model.DownloadFileRequest;
import io.micronaut.configuration.kafka.annotation.*;
import io.micronaut.context.annotation.Prototype;
import io.micronaut.context.annotation.Value;
import jakarta.inject.Inject;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.UUID;

@Slf4j
@KafkaListener(threads = 3,
        groupId = "download-request-listener")
@Prototype
public class DownloadDumpFileListener {
    @Value("${wikipedia.download-location}")
    String downloadLocation;

    @Inject
    FileDownloader fileDownloader;


    @Topic("download-request")
    public void receive(@KafkaKey UUID uuid,
                        DownloadFileRequest request,
                        long offset,
                        int partition,
                        String topic,
                        long timestamp) {
        log.debug("Got the request {} with UUID {}", request, uuid.toString());
        log.info("this {} was sent {} ago from partition {} from the {} topic at {}",
                request.getURL(), offset, partition, topic, timestamp);

        //now we're going to download the file
        downloadFileToDestination(request);
    }

    private void downloadFileToDestination(DownloadFileRequest request) {
        try {
            fileDownloader.download(
                    new URL(request.getURL()),
                    new File(this.downloadLocation + "/" + request.getFileName()),
                    request.getErrorcheck());
        } catch (MalformedURLException e) {
            log.error("message {} was sent ok but did not have a valid URL", request);
        }
    }
}

As you can see, it follows all the steps per the directions.

Now, this is what happens:

1. Start the listener. Three threads are created (hurray!)
2. Produce the messages for consumption
3. Only one consumer is processing the messages serially

I'm positive there's got to be a quick fix around this. I'll be glad to debug the micronaut code and get deeper - but first I want to ensure that "I'm doing it right."

Additional points
(most of) my application.yml:

micronaut:
  executors:
    consumer:
      type: fixed
      nThreads: 25
  application:
    name: downloadRequestProcessor
kafka:
  key:
    serializer: org.apache.kafka.common.serialization.UUIDSerializer
    deserializer: org.apache.kafka.common.serialization.UUIDDeserializer
  value:
    serializer: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer
    deserializer: io.confluent.kafka.streams.serdes.avro.SpecificAvroDeserializer

  enabled: true
  schema.registry.url: http://localhost:8081

Then my resulting listener configuration:

[main] INFO org.apache.kafka.clients.consumer.ConsumerConfig - ConsumerConfig values: 
	allow.auto.create.topics = true
	auto.commit.interval.ms = 5000
	auto.offset.reset = latest
	bootstrap.servers = [localhost:9092]
	check.crcs = true
	client.dns.lookup = use_all_dns_ips
	client.id = download-request-processor-download-dump-file-listener-11
	client.rack = 
	connections.max.idle.ms = 540000
	default.api.timeout.ms = 60000
	enable.auto.commit = true
	exclude.internal.topics = true
	fetch.max.bytes = 52428800
	fetch.max.wait.ms = 500
	fetch.min.bytes = 1
	group.id = download-request-listener
	group.instance.id = null
	heartbeat.interval.ms = 3000
	interceptor.classes = []
	internal.leave.group.on.close = true
	internal.throw.on.fetch.stable.offset.unsupported = false
	isolation.level = read_uncommitted
	key.deserializer = class org.apache.kafka.common.serialization.UUIDDeserializer
	max.partition.fetch.bytes = 1048576
	max.poll.interval.ms = 300000
	max.poll.records = 500
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
	receive.buffer.bytes = 65536
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = GSSAPI
	security.protocol = PLAINTEXT
	security.providers = null
	send.buffer.bytes = 131072
	session.timeout.ms = 10000
	socket.connection.setup.timeout.max.ms = 30000
	socket.connection.setup.timeout.ms = 10000
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.3
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	value.deserializer = class io.confluent.kafka.streams.serdes.avro.SpecificAvroDeserializer

(repeated 3x, since it is successfully creating 3 listeners as we expect...)

Now, I start running the producers and only one if them is working.

I'm 100% convinced that I have one setting that's just a little off - or a few.

Can anyone let me know what more I can do to test this out?

[krickert]

Well, so much for that adventure. I figured out a way around it (although I'd rather just use a single partition - since some files might take a LONG time and others very short. That means I'm going to have hanging threads that are just listening to nothing)

I found two ways around it - and for now I'll run with partitions because it's more straightforward.

1. Create the topic with 3 partitions
2. when sending the messages, I did this:

        int partitionOffset = 0;
        int numOfPartitions = 3;
        for(DownloadFileRequest request : sendMe) {
            producer.sendDownloadRequest(UUID.randomUUID(), (partitionOffset++%numOfPartitions), request);
        }

Which appropriately hashes to the 3 partition buckets.

Then I don't have to do a think with the threads- Kafka automagically assigned a partition to each one.

So I realized: the way it's setup now, I'm not sure how to have multiple threads of consumers read from a single partition. This would be really cool to do - it would prevent a situation like this with two partitions:

File 1 - 1MB
File 2 - 10GB
File 3 - 1K
File 4 - 1GB

using the partition strategy above, the following partitions would be created:
Partition 1: File 1, File 3
Partition 2: File 2, File 4

Consumers:
Thread-1: partition 1
Thread-2: partition 2

Result: Thread 2 takes FOREVER to complete (with my ISP at least) and Thread 1 is done before I refresh this page.

I did fine a strategy to have each thread point to a block of offsets:
How to have multiple consumers read from a single kafka partition

But it feels like I'm doing kafka gymnastics if I do this.

So just to see if anyone agrees: kafka might not be the best messaging queue for this situation, right? Due to it's in-process nature (which is awesome), trying to consume messages without a care for the ordering seems like it isn't the best product.

Or am I wrong? Does kafka has this capability and if so, does Micronaut's helpers provide it? I suppose underneath the rug, I can look at the code that does this. I'm willing to get my hands dirty to add it if someone can talk to me a little more about it. Or am I trying to fit a square peg in a round hole?

[graemerocher]

I suspect you have more consumers than partitions so the consumers running in other threads are simply idle. You can implement the ConsumerRebalanceListeners interface to manually assign to a specific partition. See https://micronaut-projects.github.io/micronaut-kafka/latest/guide/index.html#kafkaOffsets for an example

[krickert]

Thanks so much! Yeah I went into a deep dive last night and split the processing up into 8 partitions instead, and the balance of big/small aren't that big of a difference. But I wouldn't mind if I can just take it off a single partition instead, because the data is all independent and the ordering won't matter.

I'll look into this more. It's really a cool product. Just feels less "kruft" than doing spring stuffs.