Kafka consumer polls return empty results even with long timeouts

Clients Java Clients
1

Hi! I am querying a kafka topic with a newly created consumer group. I chose the following props:

properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getCanonicalName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class.getCanonicalName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "someRandomGroup2");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString());
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

My polling logic is like so:

try (KafkaConsumer<String, JsonNode> consumer = new KafkaConsumer<>(properties)) {
            consumer.subscribe(List.of(topic));
            ConsumerRecords<String, JsonNode> records;
            records = consumer.poll(Duration.ofMillis(KAFKA_CONSUMER_TIMEOUT_MS));
            System.out.println("Length of recs: " + records.count());
}..

The issue is similar to here: Consumer.poll() returns empty ConsumerRecords
Setting KAFKA_CONSUMER_TIMEOUT_MS = 1000 (ie 1s) makes results empty very often (4/5 times)
Setting KAFKA_CONSUMER_TIMEOUT_MS = 10000 (ie 10s) gets me at best 1 record from 1 partition.

When I change AUTO_OFFSET_RESET_CONFIG to ‘latest’ and set a new consumer group id = “latestConsumerGroup”, even with 10s timeout I get 0 results.

Why? This feels so weird and buggy. Am I doing something incorrect here?

Also, how does the confluent cloud UI list of messages on a partition work? My hope was to replicate similar functionality (ie fetch N messages from start or end - anything will do).

Screenshot 2024-08-24 at 11.20.00 AM
Screenshot 2024-08-24 at 11.20.00 AM1448×1548 192 KB

2

This question was cross-posted on the community slack. From the context provided there, it seems that using assign() instead of subscribe() might be the way to go.

3

Regarding the behavior of the cloud UI, here are some notes that might help you (keep in mind that the UI uses the Kafka REST APIs but the same general principles would likely apply across all clients).

In the case of fetching the “Latest” records as pictured. If there are not records being actively produced to the topic then you will see zero records initially. The Cloud UI tries to work around this “cold start” experience by doing a look-back to grab whatever the “Max records” value is set to.
So if the initial result set comes back empty we would use the offset values for each partition and subtract some amount (current offset - x records to look back). It looks like the Consumer provides a seek() method that might be useful in this specific use-case.