Skip to main content

Kafka Cluster Deployment

NodeServer IPHost RoleNode ID
Node01192.168.10.7Kafka Broker and Controller1
Node02192.168.10.8Kafka Broker and Controller2
Node03192.168.10.9Kafka Broker and Controller3

Install Kafka​

Execution servers: Node01, Node02, and Node03 (run on all three servers).

  1. Download the JDK installation package

    wget https://pdpublic.mingdao.com/private-deployment/offline/common/OpenJDK21U-jdk_x64_linux_hotspot_21.0.8_9.tar.gz

  2. Extract the JDK and configure the Java symbolic link

    tar -zxvf OpenJDK21U-jdk_x64_linux_hotspot_21.0.8_9.tar.gz -C /usr/local
    mv /usr/local/jdk-21.0.8+9 /usr/local/openjdk-21
    ln -s /usr/local/openjdk-21/bin/java /bin/java
    java -version

  3. Download the Kafka installation package

    wget https://pdpublic.mingdao.com/private-deployment/offline/common/kafka_2.13-4.3.1.tgz

  4. Install Kafka and create the data directory

    tar -zxvf kafka_2.13-4.3.1.tgz -C /usr/local
    mv /usr/local/kafka_2.13-4.3.1 /usr/local/kafka
    mkdir -p /data/kafka/kafka-logs

  5. Set the Kafka JVM heap size to 4 GB

    sed -i ':a;N;$!ba;s/Xm[xs]1G/Xmx4G/1' /usr/local/kafka/bin/kafka-server-start.sh
    sed -i ':a;N;$!ba;s/Xm[xs]1G/Xms4G/1' /usr/local/kafka/bin/kafka-server-start.sh

  6. Create the Kafka service account

    useradd -M -s /sbin/nologin kafka

Configure Each Node​

The following sections provide the server.properties configuration for Node01, Node02, and Node03. Before writing each file, confirm the node IP addresses and ports for your environment, and update the following properties as needed:

  • node.id: Set this property to 1, 2, and 3 on Node01, Node02, and Node03, respectively. Node IDs must be unique.

  • controller.quorum.bootstrap.servers: Use the same value on all three nodes and include the IP address and port 9093 of all three Controller nodes.

  • advertised.listeners: Enter the current node's IP address and Broker port 9092. Applications use this address to connect to Kafka.

Node01​

Execution server: Node01.

cat > /usr/local/kafka/config/server.properties <<'EOF'
process.roles=broker,controller
node.id=1
controller.quorum.bootstrap.servers=192.168.10.7:9093,192.168.10.8:9093,192.168.10.9:9093
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
advertised.listeners=PLAINTEXT://192.168.10.7:9092
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
inter.broker.listener.name=PLAINTEXT
controller.listener.names=CONTROLLER
log.dirs=/data/kafka/kafka-logs
num.partitions=10
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
default.replication.factor=3
min.insync.replicas=2
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
group.initial.rebalance.delay.ms=3000
message.max.bytes=10485760
replica.fetch.max.bytes=10485760
EOF

Node02​

Execution server: Node02.

cat > /usr/local/kafka/config/server.properties <<'EOF'
process.roles=broker,controller
node.id=2
controller.quorum.bootstrap.servers=192.168.10.7:9093,192.168.10.8:9093,192.168.10.9:9093
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
advertised.listeners=PLAINTEXT://192.168.10.8:9092
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
inter.broker.listener.name=PLAINTEXT
controller.listener.names=CONTROLLER
log.dirs=/data/kafka/kafka-logs
num.partitions=10
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
default.replication.factor=3
min.insync.replicas=2
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
group.initial.rebalance.delay.ms=3000
message.max.bytes=10485760
replica.fetch.max.bytes=10485760
EOF

Node03​

Execution server: Node03.

cat > /usr/local/kafka/config/server.properties <<'EOF'
process.roles=broker,controller
node.id=3
controller.quorum.bootstrap.servers=192.168.10.7:9093,192.168.10.8:9093,192.168.10.9:9093
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
advertised.listeners=PLAINTEXT://192.168.10.9:9092
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
inter.broker.listener.name=PLAINTEXT
controller.listener.names=CONTROLLER
log.dirs=/data/kafka/kafka-logs
num.partitions=10
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
default.replication.factor=3
min.insync.replicas=2
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
group.initial.rebalance.delay.ms=3000
message.max.bytes=10485760
replica.fetch.max.bytes=10485760
EOF

Initialize and Start the Cluster​

  1. Generate and save the cluster ID and Controller directory IDs

    Execution server: Node01.

    Run the following commands on Node01 to generate one cluster ID and three Controller directory IDs. Record all four values and use the same values when formatting the three nodes.

    KAFKA_CLUSTER_ID=$(/usr/local/kafka/bin/kafka-storage.sh random-uuid)
    CONTROLLER_1_UUID=$(/usr/local/kafka/bin/kafka-storage.sh random-uuid)
    CONTROLLER_2_UUID=$(/usr/local/kafka/bin/kafka-storage.sh random-uuid)
    CONTROLLER_3_UUID=$(/usr/local/kafka/bin/kafka-storage.sh random-uuid)
    echo "KAFKA_CLUSTER_ID=$KAFKA_CLUSTER_ID"
    echo "CONTROLLER_1_UUID=$CONTROLLER_1_UUID"
    echo "CONTROLLER_2_UUID=$CONTROLLER_2_UUID"
    echo "CONTROLLER_3_UUID=$CONTROLLER_3_UUID"

  2. Format the data directories on all three nodes

    Execution servers: Node01, Node02, and Node03 (run on all three servers).

    Run the following commands on each Kafka server. Replace the variables with the values recorded in the previous step. Also replace the IP addresses in --initial-controllers with the actual Controller IP addresses.

    KAFKA_CLUSTER_ID=REPLACE_WITH_ACTUAL_CLUSTER_ID
    CONTROLLER_1_UUID=REPLACE_WITH_ACTUAL_CONTROLLER_DIRECTORY_ID
    CONTROLLER_2_UUID=REPLACE_WITH_ACTUAL_CONTROLLER_DIRECTORY_ID
    CONTROLLER_3_UUID=REPLACE_WITH_ACTUAL_CONTROLLER_DIRECTORY_ID
    /usr/local/kafka/bin/kafka-storage.sh format \
      --cluster-id "$KAFKA_CLUSTER_ID" \
      --initial-controllers "1@192.168.10.7:9093:${CONTROLLER_1_UUID},2@192.168.10.8:9093:${CONTROLLER_2_UUID},3@192.168.10.9:9093:${CONTROLLER_3_UUID}" \
      --config /usr/local/kafka/config/server.properties
    • Use this operation only when initializing empty data directories for the first time. Never reformat a node that contains data or has already joined the cluster. Doing so removes its local metadata and prevents it from rejoining the cluster correctly.

  3. Grant permissions to the Kafka directories

    Execution servers: Node01, Node02, and Node03 (run on all three servers).

    After formatting the storage directories, reset the directory ownership on all three servers so that the Kafka service can read and write the program and data directories.

    chown -R kafka:kafka /usr/local/kafka /data/kafka

  4. Configure the systemd service

    Execution servers: Node01, Node02, and Node03 (run on all three servers).

    cat > /etc/systemd/system/kafka.service <<'EOF'
    [Unit]
    Description=Kafka KRaft
    After=network.target
    [Service]
    User=kafka
    Group=kafka
    LimitNOFILE=102400
    LimitNPROC=102400
    ExecStart=/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties
    ExecStop=/usr/bin/kill $MAINPID
    Restart=on-failure
    [Install]
    WantedBy=multi-user.target
    EOF
    systemctl daemon-reload

  5. Start the Kafka cluster

    Execution servers: Node01, Node02, and Node03 (run on all three servers).

    systemctl enable kafka
    systemctl start kafka
    systemctl status kafka

Verify the Cluster​

  1. Verify the Controller quorum status

    Run the following command on any node. Confirm that a Controller Leader has been elected and that all three nodes are members of the quorum.

    /usr/local/kafka/bin/kafka-metadata-quorum.sh --bootstrap-server 127.0.0.1:9092 describe --status

  2. Create and inspect a three-replica Topic

    Run the following commands on any node.

    /usr/local/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic test01 --partitions 3 --replication-factor 3
    /usr/local/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --describe --topic test01

    Confirm that each partition has three replicas and that the Isr list contains all three node IDs.

  3. Start a producer on Node01

    /usr/local/kafka/bin/kafka-console-producer.sh --bootstrap-server 127.0.0.1:9092 --topic test01

  4. Start a consumer on Node02 or Node03

    /usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic test01 --from-beginning

    If the consumer can receive the messages sent by the producer, the Kafka cluster message flow is working correctly.