Consuming Messages

1. Creating Consumer Instance

Property	Description	Default Value	Notes
bootstrap.servers	Initial list of brokers to connect to	None	Required. Format: `host1:port1,host2:port2`
group.id	Consumer group ID	None	Required for consumer groups. Unique identifier for the consumer group.
key.deserializer	Deserializer for message keys	`org.apache.kafka.common.serialization.StringDeserializer`	Class name or instance of deserializer
value.deserializer	Deserializer for message values	`org.apache.kafka.common.serialization.StringDeserializer`	Class name or instance of deserializer
enable.auto.commit	Whether to automatically commit offsets	`true`	Set to `false` for manual offset management
auto.offset.reset	What to do when there is no initial offset or the offset is out of range	`latest`	Options: `earliest`, `latest`, `none`

Example: Creating a Basic Consumer Instance

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

2. Setting Consumer Properties

Property	Default Value	Description	Notes
bootstrap.servers	None	List of broker addresses to connect to	Required for consumer to start
group.id	None	Consumer group ID for load balancing	Required for consumer to join a group
key.deserializer	org.apache.kafka.common.serialization.StringDeserializer	Deserializes message keys	Must match the serializer used when producing
value.deserializer	org.apache.kafka.common.serialization.StringDeserializer	Deserializes message values	Must match the serializer used when producing
enable.auto.commit	true	Automatically commit offsets	Set to false for manual offset control
auto.commit.interval.ms	5000	Interval between automatic offset commits	Only used when enable.auto.commit=true
session.timeout.ms	45000	Timeout for consumer session	Used to detect consumer failures
heartbeat.interval.ms	3000	Interval between heartbeat requests	Must be less than session.timeout.ms
max.poll.records	500	Maximum number of records returned in a single poll	Helps control memory usage
fetch.min.bytes	1	Minimum bytes to fetch in a single request	Can be used to reduce network overhead
fetch.max.wait.ms	500	Maximum time to wait for data in a fetch request	Used to balance latency and throughput

Example: Consumer Configuration

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false");
props.put("max.poll.records", "100");

3. Configuring Bootstrap Servers

Property	Default Value	Description
bootstrap.servers	localhost:9092	Comma-separated list of host and port pairs that are the addresses of Kafka brokers to connect to.
bootstrap.servers	localhost:9092	Used to establish initial connection to the Kafka cluster.

Example: Setting Bootstrap Servers

{
  "bootstrap.servers": "broker1:9092,broker2:9092,broker3:9092"
}

Note: The client will use this list to discover all the brokers in the cluster.

4. Configuring Deserializers

Property	Default Value	Description
key.deserializer	`org.apache.kafka.common.serialization.StringDeserializer`	Deserializer for message keys
value.deserializer	`org.apache.kafka.common.serialization.ByteArrayDeserializer`	Deserializer for message values

Example: Custom Deserializer Configuration

Properties props = new Properties();
props.put("key.deserializer", "com.example.CustomKeyDeserializer");
props.put("value.deserializer", "com.example.CustomValueDeserializer");

Note: Custom deserializers must implement org.apache.kafka.common.serialization.Deserializer interface.

5. Setting Consumer Group

Property	Default Value	Description
group.id	`consumer-group`	The unique identifier for the consumer group. All consumers in the same group will share the load of consuming messages from a topic.
session.timeout.ms	`45000`	The timeout used to detect consumer failures. If a consumer does not send a heartbeat within this time, it is considered dead and the group rebalances.
heartbeat.interval.ms	`3000`	The time interval between heartbeats to the consumer group coordinator.
max.poll.interval.ms	`300000`	The maximum time between poll calls. If a consumer takes longer than this, it is considered dead and the group rebalances.

Example: Consumer Group Configuration

const consumer = kafka.consumer({
  groupId: 'my-consumer-group',
  sessionTimeout: 60000,
  heartbeatInterval: 5000,
  maxPollInterval: 300000
});

6. Subscribing to Topics

Method	Description	Parameters	Notes
subscribe()	Subscribe to a list of topics	`topics: string[]`	Can be called multiple times to add more topics
subscribe()	Subscribe to topics using a pattern	`topicPattern: string`	Matches topics based on regex pattern

Example: Subscribing to Topics

kafkaConsumer.subscribe({ topics: ['my-topic', 'another-topic'] });

Example: Subscribing with Pattern

kafkaConsumer.subscribe({ topicPattern: 'my-topic-.*' });

7. Polling Messages

Method	Description	Parameters	Return Type
poll()	Polls for new messages from Kafka topics	`timeoutMs` (optional): Maximum time to wait for messages	`ConsumerRecord[]`
poll(long timeoutMs)	Polls for new messages with a specified timeout	`timeoutMs`: Maximum time to wait for messages in milliseconds	`ConsumerRecord[]`

Example: Polling Messages

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092']
});

const consumer = kafka.consumer({ groupId: 'test-group' });

await consumer.connect();
await consumer.subscribe({ topic: 'test-topic', partitions: [0] });

const messages = await consumer.poll({
  timeout: 1000
});

console.log(messages);

Note: The poll() method returns a list of messages or an empty array if no messages are available within the specified timeout.

8. Processing Records

Property	Description	Default	Notes
key.deserializer	Deserializer for record keys	`org.apache.kafka.common.serialization.StringDeserializer`	Must implement `org.apache.kafka.common.serialization.Deserializer`
value.deserializer	Deserializer for record values	`org.apache.kafka.common.serialization.StringDeserializer`	Must implement `org.apache.kafka.common.serialization.Deserializer`
enable.auto.commit	Automatically commit offsets	`true`	Set to `false` for manual control
auto.commit.interval.ms	Interval for automatic offset commits	`5000`	Only used when `enable.auto.commit=true`
max.poll.records	Maximum number of records returned in a single poll	`500`	Controls batch size for processing

Example: Basic Record Processing

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
}

Example: Manual Offset Committing

props.put("enable.auto.commit", "false");
props.put("max.poll.records", "100");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        // Process the record
        processRecord(record);
    }
    consumer.commitSync(); // Commit all processed records
}

Note: When enable.auto.commit is set to false, you must manually commit offsets to ensure that records are not reprocessed in case of consumer failure.

9. Using Manual Assignment

Example: Manual Partition Assignment

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

public class ManualAssignmentExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("group.id", "manual-group");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        
        // Manually assign partitions
        List<TopicPartition> partitions = Arrays.asList(
            new TopicPartition("my-topic", 0),
            new TopicPartition("my-topic", 1)
        );
        
        consumer.assign(partitions);
        
        // Poll for messages
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", 
                    record.offset(), record.key(), record.value());
            }
        }
    }
}

Method	Description	Notes
assign(Collection<TopicPartition> partitions)	Manually assign specific partitions to the consumer	Overrides automatic partition assignment
subscription()	Returns the current topic subscription	Useful for checking what topics are subscribed to
assignment()	Returns the set of partitions currently assigned	Useful for checking what partitions are assigned to this consumer

Note: When using manual assignment, the consumer does not participate in group rebalancing. You must handle partition reassignment manually if partitions are added or removed.

10. Seeking to Position

Example: Seek to Specific Offset

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

public class SeekExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("group.id", "seek-example-group");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        TopicPartition partition = new TopicPartition("my-topic", 0);

        // Seek to a specific offset
        consumer.seek(partition, 100L);

        // Poll for messages from the new position
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }

        consumer.close();
    }
}

Method	Description	Parameters
seek(TopicPartition partition, long offset)	Sets the fetch position of the consumer to a specific offset	`partition`: TopicPartition object `offset`: The offset to seek to
seekToBeginning(Collection<TopicPartition> partitions)	Seeks to the beginning of each partition	`partitions`: Collection of TopicPartition objects
seekToEnd(Collection<TopicPartition> partitions)	Seeks to the end of each partition	`partitions`: Collection of TopicPartition objects

Note: Seeking to a position does not guarantee that the message at that offset will be available if it has been deleted due to retention policies.

11. Closing Consumer

Method	Description	Notes
close()	Closes the consumer and releases all resources.	Should be called when consumer is no longer needed to free up resources.
close(timeout)	Closes the consumer with a specified timeout.	Allows for graceful shutdown within a given time frame.

Example: Closing Consumer

const consumer = kafka.consumer({ groupId: 'test-group' });

await consumer.connect();
await consumer.subscribe({ topic: 'test-topic', partitions: [0] });

// Consume messages...
await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    console.log(message.value.toString());
  }
});

// Close consumer when done
await consumer.close();

Note: It is important to close the consumer to release resources and avoid memory leaks.