Using Console Tools

1. Producing Messages

Example: Produce a message to a Kafka topic

kafka-console-producer.sh \
  --broker-list localhost:9092 \
  --topic test-topic \
  --property parse.key=true \
  --property key.separator=:

Option	Values	Description
--broker-list	`localhost:9092`	Comma-separated list of broker addresses
--topic	`test-topic`	Name of the topic to produce messages to
--property parse.key=true	`true`	Enable parsing of message keys
--property key.separator=:	`:`	Separator used between key and value in input

Note: The console producer reads messages from standard input. Each line is treated as a separate message.

2. Consuming Messages

Command	Description	Options	Example
kafka-console-consumer.sh	Consumes messages from a Kafka topic	`--bootstrap-server`, `--topic`, `--from-beginning`	`kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning`

Example: Consuming messages from a topic

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic test-topic \
  --from-beginning \
  --value-only

Example: Consuming messages with custom deserializer

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic json-topic \
  --key-deserializer org.apache.kafka.common.serialization.StringDeserializer \
  --value-deserializer org.apache.kafka.common.serialization.StringDeserializer

Note: The --from-beginning flag ensures that all messages in the topic are consumed, not just new ones.

3. Reading from Beginning

Example: Read from Beginning

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic my-topic \
  --from-beginning \
  --value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

Option	Values	Description
--bootstrap-server	`localhost:9092`	Address of the Kafka broker to connect to
--topic	`my-topic`	Name of the topic to consume messages from
--from-beginning		Start reading from the earliest available message
--value-deserializer	`org.apache.kafka.common.serialization.StringDeserializer`	Deserializes message values as strings

Note: The --from-beginning flag ensures that all messages in the topic are read, not just new ones.

4. Specifying Consumer Group

Property	Value	Description
group.id	`my-consumer-group`	Unique identifier for the consumer group
session.timeout.ms	`45000`	Timeout for consumer session
heartbeat.interval.ms	`3000`	Interval between heartbeat requests

Example: Consumer Group Configuration

{
  "group.id": "my-consumer-group",
  "session.timeout.ms": 45000,
  "heartbeat.interval.ms": 3000
}

Note: The group.id must be unique per consumer group to ensure proper message consumption.

5. Setting Message Format

Format	Description	Usage
json	JSON formatted messages	Use for structured data with key-value pairs
avro	Avro binary format	Use for schema evolution and efficient serialization
protobuf	Protocol Buffers format	Use for high-performance binary serialization
string	Plain text messages	Use for human-readable content or simple payloads

Example: Setting JSON format for message consumption

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic my-topic \
  --from-beginning \
  --value-format json

Example: Setting Avro format for message production

kafka-console-producer.sh \
  --broker-list localhost:9092 \
  --topic my-topic \
  --value-format avro \
  --property schema.registry.url=http://localhost:8081

6. Using Key Deserializer

Property	Values	Description
key.deserializer	`org.apache.kafka.common.serialization.StringDeserializer`	Serializer class for key bytes
key.deserializer	`org.apache.kafka.common.serialization.ByteArrayDeserializer`	Serializer class for key bytes
key.deserializer	`org.apache.kafka.common.serialization.IntegerDeserializer`	Serializer class for key bytes

Example: Configuring Key Deserializer

Properties props = new Properties();
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

Note: Ensure that the deserializer matches the serializer used when producing messages.

7. Setting Partition

Property	Values	Description
partition.count	`1`	Number of partitions for the topic
replication.factor	`1`	Replication factor for the topic
topic.name	`my-topic`	Name of the Kafka topic to create

Example: Create topic with 3 partitions

bin/kafka-topics.sh --create --topic my-topic --partitions 3 --replication-factor 1 --bootstrap-server localhost:9092

Note: Ensure the replication factor does not exceed the number of available brokers.

8. Reading Specific Offset

Property	Value	Description
--offset	`0`	Start reading from a specific offset
--partition	`0`	Read from a specific partition
--topic	`my-topic`	Name of the topic to read from

Example: Reading from a specific offset

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic my-topic \
  --from-beginning \
  --offset 100

Note: The --from-beginning flag is used to read from the beginning of the topic, but you can specify a specific offset using --offset.

9. Displaying Message Timestamp

Property	Description	Default Value	Notes
timestamp.type	Specifies the type of timestamp to display	`CreateTime`	Can be `CreateTime`, `LogAppendTime`, or `Nothing`
timestamp.format	Format for displaying the timestamp	`iso`	Can be `iso`, `unix`, or `custom`

Example: Displaying Message Timestamps in Console Consumer

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic my-topic \
  --from-beginning \
  --value-deserializer org.apache.kafka.common.serialization.StringDeserializer \
  --timestamp.type CreateTime \
  --timestamp.format iso

Note: The --timestamp.type and --timestamp.format options are available in Kafka 2.0+.

10. Using Timeout

Option	Values	Description
--timeout	`30000`	Timeout in milliseconds for the console consumer to wait for messages
--max.poll.interval.ms	`300000`	Maximum time between consumer polls before rebalancing

Example: Setting Timeout for Console Consumer

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic my-topic \
  --from-beginning \
  --timeout 5000

Note: The --timeout parameter controls how long the consumer waits for new messages before exiting. If no messages are received within this time, the consumer will exit.