Developing Event-driven Applications with Apache Kafka and Red Hat AMQ Streams

The previous approach allows different consumers to process records from different partitions within the same application logic. In the previous banking diagram, consider another application (for example, a payments application) that needs to process the transactions independently. This means that its consumers read all transactions in the topic regardless of other applications, so its consumers are independent from another application's consumers.

To achieve several consumers working together independently, Kafka introduces the concept of a consumer group. A consumer group contains several consumers reading together from a topic. Every consumer is assigned several partitions to read from. The following diagram displays a banking diagram with two applications: accounts and payments. Both applications read the records in the topic independently.

Figure 2.10: Consumers in different consumer groups reading from a topic

The main way to scale reading data in Kafka is by adding more consumers to a consumer group. If you include a new consumer, then Kafka redistributes the partitions assigned to each consumer. This process is known as rebalance and happens whenever a new consumer joins or leaves the consumer group. Kafka distributes partitions among consumers in the group based on the provided partition assignment strategy for the group.

Two consumers within the same consumer group cannot read from the same partition. Therefore, if there are more consumers reading from a topic than partitions in that topic, then some consumers are left unassigned and serve as a fallback in the case that any assigned consumer crashes. For example, if a topic has 5 consumers reading but there are only 3 partitions to read from, then Kafka leaves 2 consumers unassigned.

Every consumer group has a designated Kafka broker called the group coordinator. The group coordinator broker is responsible for receiving the heartbeats of every consumer. A heartbeat is a request sent by the consumer application to the group coordinator broker to notify that the consumer is still alive, this request is sent regularly. If the group coordinator broker stops receiving heartbeats for a long enough time, then the consumer is considered dead. In this case, the group coordinator triggers a rebalance. The partitions of the dead consumer are assigned to other consumers in the group.

The partition assignment happens when the consumer joins the group or when another consumer leaves the group. There are several ways in which Kafka can assign partitions to consumers.

Consumers actively request data from the corresponding broker. Consumers receive records in batches. The maximum number of records in the batch can be configured by setting the max.poll.records consumer property. The following snippet shows a typical implementation of a consumer in the Java programming language.

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;

Consumer<Void,Integer> consumer = new KafkaConsumer<>(configureProperties());
consumer.subscribe(Collections.singletonList("topic"));

while (true) { 
    System.out.println("Waiting for events...");

    ConsumerRecords<Void, Integer> records =
        consumer.poll(Duration.ofMillis(10000)); 

    for (ConsumerRecord<Void, Integer> record : records) { 
        System.out.println("Record received: " + record.value());
    }
}

For every partition and every consumer group, Kafka keeps a consumer offset. This offset is used to determine the last record read by a consumer. This partition offset is independent for every consumer group, allowing every consumer group to read records independently. Kafka stores different record offset values for two different consumer groups.

Keeping this offset is useful when a partition rebalance is triggered. For example, consider two consumers, A and B in the same consumer group. A is reading from Partition 0 and B is reading from Partition 1. If a rebalance occurs and A is assigned Partition 1, then the consumer starts reading from the offset of Consumer B for Partition 1. The consumer offset avoids reading duplicate records.

The consumer is responsible for sending an offset commit to Kafka. Committing means acknowledging that the records have been received successfully. The consumer sends the offset of the last record that has been received correctly.

You can send an automatic commit by setting the enable.auto.commit=true property. With this approach, the consumer sends a request to Kafka with the last record offset received periodically. The default time is five seconds.

Using an automatic commit, however, might cause race condition issues. For example, a consumer can take longer than five seconds to process a record. If it throws an exception after six seconds, the record has already been committed. Data duplication and data loss is covered later in this course.

By using the Quarkus Kafka extension, you do not have to take care of creating the Consumer object or declaring the poll loop. Instead, on the start-up of the application, Kafka will:

The following snippet shows a method annotated with @Incoming responsible for processing records of double type.

import org.eclipse.microprofile.reactive.messaging.Incoming;
import javax.enterprise.context.ApplicationScoped;

@ApplicationScoped
public class PriceConsumer {

    @Incoming("prices") 
    public void consume(double price) {
        // Process record
    }
}

A possible configuration for the previous snippet:

kafka.bootstrap.servers=my-cluster-kafka-bootstrap-youruser-kafka-cluster.apps.cluster.example.com:443
kafka.security.protocol=SSL
kafka.ssl.truststore.location=/ABSOLUTE_PATH_TO_YOUR_WORKSPACE_FOLDER/truststore.jks
kafka.ssl.truststore.password=password

mp.messaging.incoming.prices.topic=store-prices 
mp.messaging.incoming.prices.value.deserializer=org.apache.kafka.common.serialization.DoubleDeserializer 
mp.messaging.incoming.prices.group.id=prices