CODEWITHSUNDEEP
javaapache-kafkakafka-consumer-api
Luiggi Mendoza
Luiggi Mendoza

Reputation: 85781

Why my Kafka consumers with same group id are not being balanced?

I'm writing a proof of concept application to consume messages from Apache Kafka 0.9.0.0 and see if I can use it instead of a common JMS message broker because of the benefits Kafka provides. This is my base code, using the new consumer API:

public class Main implements Runnable {

    public static final long DEFAULT_POLL_TIME = 300;
    public static final String DEFAULT_GROUP_ID = "ltmjTest";

    volatile boolean keepRunning = true;
    private KafkaConsumer<String, Object> consumer;
    private String servers;
    private String groupId = DEFAULT_GROUP_ID;
    private long pollTime = DEFAULT_POLL_TIME;
    private String[] topics;

    public Main() {
    }

    //getters and setters...

    public void createConsumer() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        configs.put("enable.auto.commit", "true");
        configs.put("auto.commit.interval.ms", "1000");
        configs.put("session.timeout.ms", "30000");

        configs.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        configs.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<>(configs);
        consumer.subscribe(asList(topics));
    }

    public static void main(String[] args) {
        Main main = new Main();
        if (args != null && args.length > 0) {
            for (String arg : args) {
                String[] realArg = arg.trim().split("=", 2);
                String argKey = realArg[0].toLowerCase();
                String argValue = realArg[1];
                switch (argKey) {
                case "polltime":
                    main.setPollTime(Long.parseLong(argValue));
                    break;
                case "groupid":
                    main.setGroupId(argValue);
                    break;
                case "servers":
                    main.setServers(argValue);
                    break;
                case "topics":
                    main.setTopics(argValue.split(","));
                    break;
            }
        }
        main.createConsumer();
        new Thread(main).start();
        try (Scanner scanner = new Scanner(System.in)) {
            while(true) {
                String line = scanner.nextLine();
                if (line.equals("stop")) {
                    main.setKeepRunning(false);
                    break;
                }
            }
        }
    }
}

I've started a kafka server using default settings and a kafka producer using the shell tool kafka-console-producer.sh to write messages to my topic. Then I connect with two consumers using this code, sending the proper server to connect and topic to subscribe, everything else with default values, which means both consumers have the same group id. I notice that only one of my consumers consumes all the data. I've read that the default behaviour should be that the consumers must be balanced by the server, from the official tutorial:

If all the consumer instances have the same consumer group, then this works just like a traditional queue balancing load over the consumers.

How can I fix the consumers to behave like the default? Or maybe I'm missing something?

Upvotes: 3

Views: 2239

Answers (1)

Natalia
Natalia

Reputation: 4532

there is trait kafka.consumer.PartitionAssignor that says how partitions should be assigned per consumers. It has two immplementations: RoundRobinAssignor and RangeAssignor. The default one is RangeAssignor.

Can be changed by setting param "partition.assignment.strategy".

Round Robin documentation:

The roundrobin assignor lays out all the available partitions and all the available consumers. It then proceeds to do a roundrobin assignment from partition to consumer. If the subscriptions of all consumer instances are identical, then the partitions will be uniformly distributed. (i.e., the partition ownership counts will be within a delta of exactly one across all consumers.) For example, suppose there are two consumers C0 and C1, two topics t0 and t1, and each topic has 3 partitions, resulting in partitions t0p0, t0p1, t0p2, t1p0, t1p1, and t1p2. The assignment will be: C0: [t0p0, t0p2, t1p1] C1: [t0p1, t1p0, t1p2]

Range Assignor documentation

The range assignor works on a per-topic basis. For each topic, we lay out the available partitions in numeric order and the consumers in lexicographic order. We then divide the number of partitions by the total number of consumers to determine the number of partitions to assign to each consumer. If it does not evenly divide, then the first few consumers will have one extra partition. For example, suppose there are two consumers C0 and C1, two topics t0 and t1, and each topic has 3 partitions, resulting in partitions t0p0, t0p1, t0p2, t1p0, t1p1, and t1p2. The assignment will be: C0: [t0p0, t0p1, t1p0, t1p1] C1: [t0p2, t1p2]

So, if all our topics have only one partition, only one consumer will work

Upvotes: 5

Related Questions