Not all kafka consumers are getting assigned to partitions

Question

I have 10 consumers and 10 partitions. I take the number of partitions

int partitionCount = getPartitionCount(kafkaUrl);

and I create the same number of consumers with the same group.id.

    public void listen() {
        try {
            String kafkaUrl = getKafkaUrl();
            int partitionCount = getPartitionCount(kafkaUrl);
            Stream.iterate(0, i -> i + 1)
                    .limit(partitionCount)
                    .forEach(index -> executorService.execute(() ->
                            consumerTask.invokeKafkaConsumerTask(prepareConsumerConfig(index, kafkaUrl), INPUT_TOPIC)));
        } catch (Exception exception) {
            logger.error("Cannot receive event from kafka ", exception);
        }


    public void invokeKafkaConsumerTask(Properties properties, String topicName) {
        try(KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties)) {
            consumer.subscribe(Collections.singletonList(topicName));
            logger.info("[KAFKA] consumer created");
            invokeKafkaConsumer(consumer);
        } catch (IllegalArgumentException exception) {
            logger.error("Cannot create kafka consumer ", exception);
        }
    }

    private void invokeKafkaConsumer(KafkaConsumer<String, String> consumer) {
        try {
            while (true) {
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(4));
                if (consumerRecords.count() > 0) {
                    consumeRecords(consumerRecords);
                    consumer.commitSync();
                }
            }
        } catch (Exception e) {
            logger.error("Error while receiving records ", e);
        }
    }

method getPartitionCount return 10 partitions so it's working right

config looks like this

Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaUrl);
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_CLIENT_ID);
properties.put(ConsumerConfig.CLIENT_ID_CONFIG, CONSUMER_CLIENT_ID + index);
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "300000");
properties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "10000");
properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, String.valueOf(Integer.MAX_VALUE));
properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RoundRobinAssignor");

what I see after assigning consumers to the partition

TOPIC      PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CLIENT-ID                                                        
topicName  1          89391           89391           0               consumer0
topicName  3          88777           88777           0               consumer1
topicName  5          89280           89280           0               consumer2
topicName  4          88776           88776           0               consumer2
topicName  0          4670991         4670991         0               consumer0
topicName  9          23307           89343           66036           consumer4
topicName  7          89610           89610           0               consumer3
topicName  8          88167           88167           0               consumer4
topicName  2          89138           89138           0               consumer1
topicName  6          88967           88967           0               consumer3

only half of the consumers have been assigned to the partitions why did this happen?
There should be one consumer per partition acording to documentation.
Am I doing something wrong?
kafka version 2.1.1.

I also find few this logs ->

Setting newly assigned partitions:[empty]

What is the configuration of property auto.leader.rebalance.enable? — Giorgos Myrianthous
I don't see it in the configuration but in logs i can see auto.leader.rebalance.enable = true — michalavis
Can you share the output of kafka-topics --describe --zookeeper my_zookeeper_ip:2181 --topic yourTopicName — Giorgos Myrianthous
the same as in my description -> PARTITION from 0-9 and CONSUMER-ID from 0-4 each consumer has 2 partitions ctxt.io/2/AABABWsYEg — michalavis
My bad. It was meant to be kafka-consumer-groups.sh --bootstrap-server <kafka_brokers> --describe –group <consumer_group_id> — Giorgos Myrianthous

michalavis michalavis · Accepted Answer · 2020-02-18T10:39:48

[solution] interesting case I changed group.id and partition.assignment.strategy, added auto.offset.reset=earliest and it looks like it works...

Not all kafka consumers are getting assigned to partitions

2 Answers