Zookeeper is a prerequisite and should be up and running before starting Kafka.
See this page for more details on how to install Zookeeper: Apache ZooKeeper

Download Apache Kafka: https://kafka.apache.org/downloads

Extract the file "kafka_2.13-3.4.0.tgz" in the folder you want to install Kafka: /opt/kafka_2.13-3.4.0
$ tar -xf ~/Downloads/kafka_2.13-3.4.0.tgz -C /opt/
Note: In the following sections, the environment variable ${KAFKA_HOME} will refer to this location "/opt/kafka_2.13-3.4.0"

$ export KAFKA_HOME=/opt/kafka_2.13-3.4.0

Create the folder "kafka-logs": "${KAFKA_HOME}/kafka-logs"
Create the log folder "logs": "${KAFKA_HOME}/logs"

$ mkdir ${KAFKA_HOME}/kafka-logs $ mkdir ${KAFKA_HOME}/logs
Configure the kafka server properties file server.properties, by adjusting the following properties:
$ vi ${KAFKA_HOME}/config/server.properties broker.id=0 listeners=PLAINTEXT://localhost:9092 log.dirs=/opt/kafka_2.13-3.4.0/kafka-logs zookeeper.connect=localhost:2181/kafka
Notes:

• Leave listener hostname empty to bind to default interface: listeners=PLAINTEXT://:9092
  
• Specify listener hostname as 0.0.0.0 to bind to all interfaces: listeners=PLAINTEXT://0.0.0.0:9092
  
• Specify multiple listeners: listeners=PLAINTEXT://myhost:9092,SSL://:9091,CLIENT://0.0.0.0:9092,REPLICATION://localhost:9093

• Start Kafka server: $ ${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties
  
  • Validate that Kafka is up/running and listening on port "9092":
    $ netstat -lp | grep 9092 tcp6 0 0 localhost:9092 [::]:* LISTEN 31402/java
    
  • Validate that Kafka was registered properly to ZooKeeeper: $ ${ZK_HOME}/bin/zkCli.sh -server localhost:2181 get /kafka/brokers/ids/0 { "listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"}, "endpoints":["PLAINTEXT://localhost:9092"], "jmx_port":-1, "host":"localhost", "timestamp":"1567958157584", "port":9092, "version":4 }
    
• Stop Kafka server: $ ${KAFKA_HOME}/bin/kafka-server-stop.sh

The following examples show how to configure a cluster of Kafka instances on the same host.

Extract the file "kafka_2.13-3.4.0.tgz" in the folders you want to install Kafka:
$ mkdir /opt/kafka-i0 $ mkdir /opt/kafka-i1 $ mkdir /opt/kafka-i2
$ tar -xf ~/Downloads/kafka_2.13-3.4.0.tgz -C /opt/kafka-i0 $ tar -xf ~/Downloads/kafka_2.13-3.4.0.tgz -C /opt/kafka-i1 $ tar -xf ~/Downloads/kafka_2.13-3.4.0.tgz -C /opt/kafka-i2
Create the folders "kafka-logs" and "logs" for each instance:
$ mkdir /opt/kafka-i0/kafka_2.13-3.4.0/kafka-logs $ mkdir /opt/kafka-i0/kafka_2.13-3.4.0/logs
$ mkdir /opt/kafka-i1/kafka_2.13-3.4.0/kafka-logs $ mkdir /opt/kafka-i1/kafka_2.13-3.4.0/logs
$ mkdir /opt/kafka-i2/kafka_2.13-3.4.0/kafka-logs $ mkdir /opt/kafka-i2/kafka_2.13-3.4.0/logs
Configure the kafka server properties file server.properties, by adjusting the following properties for each instance (the broker.id, listeners, log.dirs are specific to each instance):
$ vi /opt/kafka-i0/kafka_2.13-3.4.0/config/server.properties broker.id=0 listeners=PLAINTEXT://localhost:9092 log.dirs=/opt/kafka-i0/kafka_2.13-3.4.0/kafka-logs zookeeper.connect=localhost:2181/kafka
$ vi /opt/kafka-i1/kafka_2.13-3.4.0/config/server.properties broker.id=1 listeners=PLAINTEXT://localhost:9192 log.dirs=/opt/kafka-i1/kafka_2.13-3.4.0/kafka-logs zookeeper.connect=localhost:2181/kafka
$ vi /opt/kafka-i2/kafka_2.13-3.4.0/config/server.properties broker.id=2 listeners=PLAINTEXT://localhost:9292 log.dirs=/opt/kafka-i2/kafka_2.13-3.4.0/kafka-logs zookeeper.connect=localhost:2181/kafka
Start Kafka servers: $ cd /opt/kafka-i0/kafka_2.13-3.4.0 && ./bin/kafka-server-start.sh -daemon ./config/server.properties $ cd /opt/kafka-i1/kafka_2.13-3.4.0 && ./bin/kafka-server-start.sh -daemon ./config/server.properties $ cd /opt/kafka-i2/kafka_2.13-3.4.0 && ./bin/kafka-server-start.sh -daemon ./config/server.properties
Stop Kafka servers: $ cd /opt/kafka-i0/kafka_2.13-3.4.0 && ./bin/kafka-server-stop.sh $ cd /opt/kafka-i1/kafka_2.13-3.4.0 && ./bin/kafka-server-stop.sh $ cd /opt/kafka-i2/kafka_2.13-3.4.0 && ./bin/kafka-server-stop.sh

The following error indicates that the configured broker.id is already in use, or Kafka server was shutdown and restarted but ZooKeeper didn't timeout yet so it appears that Kafka is re-registering. FATAL [Kafka Server 0], Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer) java.lang.RuntimeException: A broker is already registered on the path /brokers/ids/0.
The following error indicates that a process is already running using the same port: FATAL [Kafka Server 0], Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer) kafka.common.KafkaException: Socket server failed to bind to localhost:9092: Address already in use.

$ ${KAFKA_HOME}/bin/kafka-server-stop.sh
$ rm -rf ${KAFKA_HOME}/kafka-logs/* $ rm -rf ${KAFKA_HOME}/logs/*
$ ${ZK_HOME}/bin/zkCli.sh -server localhost:2181 deleteall /kafka

Create a topic with one partition: $ ${KAFKA_HOME}/bin/kafka-topics.sh --create \ --topic topic1 --partitions 1 --replication-factor 1 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" Created topic topic1.
Describe topic1: $ ${KAFKA_HOME}/bin/kafka-topics.sh --describe \ --topic topic1 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" Topic: topic1 | TopicId: qYBDRmzmT8GYae3Kz41WQw | PartitionCount: 1 | ReplicationFactor: 1 | Configs: Topic: topic1 | Partition: 0 | Leader: 0 | Replicas: 0 | Isr: 0
Create a topic with two partitions: $ ${KAFKA_HOME}/bin/kafka-topics.sh --create \ --topic topic2 --partitions 2 --replication-factor 3 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" Created topic topic2.
Describe topic2: $ ${KAFKA_HOME}/bin/kafka-topics.sh --describe \ --topic topic2 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" Topic: topic2 | TopicId: yH9U2HJ1QrWNODyQTzZjCg | PartitionCount: 2 | ReplicationFactor: 3 | Configs: Topic: topic2 | Partition: 0 | Leader: 1 | Replicas: 1,0,2 | Isr: 1,0,2 Topic: topic2 | Partition: 1 | Leader: 0 | Replicas: 0,2,1 | Isr: 0,2,1
Add new partitions to a topic (if partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected): $ ${KAFKA_HOME}/bin/kafka-topics.sh --alter \ --topic topic2 --partitions 3 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292"
Describe topic2: $ ${KAFKA_HOME}/bin/kafka-topics.sh --describe \ --topic topic2 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" Topic: topic2 | TopicId: yH9U2HJ1QrWNODyQTzZjCg | PartitionCount: 3 | ReplicationFactor: 3 | Configs: Topic: topic2 | Partition: 0 | Leader: 1 | Replicas: 1,0,2 | Isr: 1,0,2 Topic: topic2 | Partition: 1 | Leader: 0 | Replicas: 0,2,1 | Isr: 0,2,1 Topic: topic2 | Partition: 2 | Leader: 0 | Replicas: 0,2,1 | Isr: 0,2,1

List topics: $ ${KAFKA_HOME}/bin/kafka-topics.sh --list \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" __consumer_offsets topic1 topic2
Describe a topic (name, partitions count, replication factor, partitions list/configurations): $ ${KAFKA_HOME}/bin/kafka-topics.sh --describe \ --topic topic2 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292"

$ ${KAFKA_HOME}/bin/kafka-topics.sh --delete \ --topic topic1 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292"

Using the console to produce a message to Kafka: $ ${KAFKA_HOME}/bin/kafka-console-producer.sh \ --topic topic2 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" >Hello Kafka World... >Hello Again... >^C
The output when your read the messages with a consumer should looks as following: Hello Kafka World... Hello Again...
Using a file to produce a message to Kafka (each line in the file is sent as a seprate message): $ echo "Message From A File" > /tmp/kafka-messages.txt $ echo "Another message From The Same File" >> /tmp/kafka-messages.txt $ ${KAFKA_HOME}/bin/kafka-console-producer.sh \ --topic topic2 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" \ < /tmp/kafka-messages.txt
The output when your read the messages with a consumer should looks as following (the order of messages might be different): Message From A File Another message From The Same File
Using a file with "key:value" lines (one entry per line) to produce a message to Kafka (each line in the file is sent as a separate message): $ echo "msgkey:Key-Value Message From A File" > /tmp/kafka-messages-key-value.txt $ echo "anothermsgkey:Another Key-Value Message From A File" >> /tmp/kafka-messages-key-value.txt $ ${KAFKA_HOME}/bin/kafka-console-producer.sh \ --topic topic2 --property "parse.key=true" --property "key.separator=:" \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" \ < /tmp/kafka-messages-key-value.txt
The output when your read the messages with a consumer should looks as following (the order of messages might be different): Key-Value Message From A File Another Key-Value Message From A File
Using "key:value" command line to produce a message to Kafka: $ for (( i=0; i<5; i++ )); do echo "key$i:value$i" | \ ${KAFKA_HOME}/bin/kafka-console-producer.sh \ --topic topic2 --property "parse.key=true" --property "key.separator=:" \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" \ ; done;
The output when your read the messages with a consumer should looks as following (the order of messages might be different): value0 value1 value2 value3 value4

You need to configure the property max.request.size of the file ${KAFKA_HOME}/config/producer.properties # the maximum size of a request in bytes max.request.size=987654321
You need to configure the property message.max.bytes of the file ${KAFKA_HOME}/config/zookeeper.properties message.max.bytes=987654321

Consume new messages from a specific topic using a custom "consumer.properties" file:
$ echo "group.id=group-id-1" > /tmp/consumer-group-id-1.properties $ ${KAFKA_HOME}/bin/kafka-console-consumer.sh \ --topic topic2 \ --consumer.config "/tmp/consumer-group-id-1.properties" \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292"
Consume all messages from a specific topic using the option "from-beginning":
$ ${KAFKA_HOME}/bin/kafka-console-consumer.sh \ --topic topic2 --from-beginning \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292"
Consume messages from a specific topic using custom properties:
$ ${KAFKA_HOME}/bin/kafka-console-consumer.sh \ --topic topic2 \ --from-beginning \ --formatter kafka.tools.DefaultMessageFormatter \ --property group.id=group-id-1 \ --property print.key=true \ --property print.value=true \ --property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \ --property value.deserializer=org.apache.kafka.common.serialization.StringDeserializer \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292"
A list of various command lines to consume messages from a specific topic using partitions properties: $ ${KAFKA_HOME}/bin/kafka-console-consumer.sh \ --topic topic2 --offset latest --partition 0 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" $ ${KAFKA_HOME}/bin/kafka-console-consumer.sh \ --topic topic2 --offset 1 --partition 0 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" $ ${KAFKA_HOME}/bin/kafka-console-consumer.sh \ --topic topic2 --consumer-property "num.partitions"=0 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292"

List all consumers groups (this will only show information about consumers that use ZooKeeper, not those using the Java Consumer API): $ ${KAFKA_HOME}/bin/kafka-consumer-groups.sh --list \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" console-consumer-82209 group-id-1
Describe a consumer group (this will only show information about consumers that use ZooKeeper, not those using the Java Consumer API): $ ${KAFKA_HOME}/bin/kafka-consumer-groups.sh --describe \ --group group-id-1 \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292"
Consumer group 'group-id-1' has no active members. GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID group-id-1 topic2 0 9 10 1 - - - group-id-1 topic2 1 8 10 2 - - - group-id-1 topic2 2 4 6 2 - - -

Get a topic configuration: $ ${KAFKA_HOME}/bin/kafka-configs.sh --describe \ --entity-name topic2 --entity-type topics \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292"
Get a broker configuration: $ ${KAFKA_HOME}/bin/kafka-configs.sh --describe \ --entity-name 0 --entity-type brokers \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292"

Add/update new/existing configuration of a specific topic: $ ${KAFKA_HOME}/bin/kafka-configs.sh --alter \ --add-config 'retention.ms=1000' --entity-name topic2 --entity-type topics \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292"
Get the topic configuration: $ ${KAFKA_HOME}/bin/kafka-configs.sh --describe \ --entity-name topic2 --entity-type topics \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292" Dynamic configs for topic topic2 are: retention.ms=1000 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:retention.ms=1000}
Delete a configuration from a specific topic: $ ${KAFKA_HOME}/bin/kafka-configs.sh --alter \ --delete-config retention.ms --entity-name topic2 --entity-type topics \ --bootstrap-server "localhost:9092,localhost:9192,localhost:9292"

Download Kafka Tool: http://www.kafkatool.com/download.html

Copy Kafka Tool in the folder you want to install it: /opt/kafkatool

Run Kafka Tool: $ /opt/kafkatool/kafkatool
Broker (Properties): - id (0) - host (localhost) - Port (9092)
Topic (Properties): - Name (topic2) - Key Content Type (No Key, String, Byte Array) - Message Content Type (String, Byte Array) - Total number of messages
Topic (Data): - Partition (0) - Offset (0) - Key - Message
Topic (Partitions): - id (0) - Leader (localhost:9092)
Consumers (Partitions): - Id (group-id-1) - Consumer Type (kafka) - Offsets stored in (kafka)
Consumers (Offsets): - Topic (topic2) - Partition (0) - Start (offset) - End (offset) - Offset - Lag