命令行安装

如果您希望使用命令行的方式部署 Kafka，请按照本章节步骤安装。

前提

• Kafka 部署需要依赖 Zookeeper 集群。用于元数据管理。 Zookeeper 安装部署请参考：Zookeeper 安装。

安装

首先登录到 kafka1，然后切换到 root 用户

ssh kafak1
su - root

创建一个kafkahosts文件，包含 Kafka 集群中所有的机器

cat > ${HOME}/kafkahosts << EOF
kafka1
kafka2
kafka3
EOF

在 kafka1 节点配置 yum 源，安装 lava 命令行管理工具

# 从yum源所在机器(假设为192.168.1.10)获取repo文件
scp root@192.168.1.10:/etc/yum.repos.d/oushu.repo /etc/yum.repos.d/oushu.repo
# 追加yum源所在机器信息到/etc/hosts文件
# 安装lava命令行管理工具
yum clean all
yum makecache
yum install -y lava

kafka1 节点和集群内其他节点交换公钥，以便 ssh 免密码登陆和分发配置文件。

lava ssh-exkeys -f ${HOME}/kafkahosts -p ********

分发 repo 文件到其他机器

lava scp -f ${HOME}/kafkahosts /etc/yum.repos.d/oushu.repo =:/etc/yum.repos.d
lava ssh -f ${HOME}/kafkahosts -e "sudo yum makecache"

安装

在使用 yum install 安装 Kafka

lava ssh -f ${HOME}/kafkahosts -e "sudo yum install -y kafka"

配置

修改 Kafka 配置文件/usr/local/oushu/conf/kafka/server.properties，注意替换其中的变量

• ${nodeIp}: Kafka 节点的 hostname, 这里的三台节点分别是 kafka1，kafka2，kafka3

• ${nodeIndex}：待安装 Kafka 节点的序号，比如kafka1，kafka2，kafka3 分别对应的 broker.id 为 0，1，2

• ${dataDir}：配置一个目录作为 Kafka 的数据文件存储

• ${zkAddress}：Zookeeper 的访问地址，例如 zk1:2181,zk2:2181,zk3:2181

cat > /usr/local/oushu/conf/kafka/server.properties << EOF
listeners=PLAINTEXT://${nodeIp}:9092
broker.id=${nodeIndex}
offsets.topic.replication.factor=1
transaction.state.log.min.isr=1
group.initial.rebalance.delay.ms=0
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
log.dirs=${dataDir}
num.partitions=1
transaction.state.log.replication.factor=1
num.io.threads=8
socket.request.max.bytes=104857600
log.retention.hours=168
log.segment.bytes=1073741824
zookeeper.connection.timeout.ms=18000
num.network.threads=3
num.recovery.threads.per.data.dir=1
log.retention.check.interval.ms=300000
zookeeper.connect=${zkAddress}
EOF

Kerberos 相关配置（可选）

修改 Kafka 配置文件/usr/local/oushu/conf/kafka/server.properties

cat >> /usr/local/oushu/conf/kafka/server.properties << EOF
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=GSSAPI
sasl.enabled.mechanisms=GSSAPI
sasl.kerberos.service.name=Kafka
advertised.listeners=SASL_PLAINTEXT://${nodeIp}:9092
EOF

准备好 kafka-jaas.conf 文件，放在 /usr/local/oushu/conf/kafka 下

分发配置文件到其他机器

lava scp -f ${HOME}/kafkahosts /usr/local/oushu/conf/kafka/server.properties =:/tmp
lava ssh -f ${HOME}/kafkahosts -e "mv -f /tmp/server.properties /usr/local/oushu/conf/kafka"
lava ssh -f ${HOME}/kafkahosts -e "chown kafka:kafka /usr/local/oushu/conf/kafka/server.properties"
# kerberos
lava scp -f ${HOME}/kafkahosts /usr/local/oushu/conf/kafka/kafka-jaas.conf =:/tmp
lava ssh -f ${HOME}/kafkahosts -e "mv -f /tmp/kafka-jaas.conf /usr/local/oushu/conf/kafka"
lava ssh -f ${HOME}/kafkahosts -e "chown kafka:kafka /usr/local/oushu/conf/kafka/kafka-jaas.conf"

之后需要到 kafka2，kafka3 节点手动修改 /usr/local/oushu/conf/kafka/server.properties 中的 broker.id，listeners，advertised.listeners 等属性的 ${nodeIndex}，${nodeIp}

启动

在 kafka1 节点, 执行以下操作以启动各节点的 Kafka 服务

lava ssh -f ${HOME}/kafkahosts -e "sudo -u kafka /usr/local/oushu/kafka/bin/kafka-server-start.sh -daemon /usr/local/oushu/conf/kafka/server.properties"

带 Kerberos 启动时需要增加一些属性，在每个 Kafka 节点执行如下命令

sudo su kafka sh -c "export KAFKA_OPTS=\"-Djava.security.krb5.conf=/usr/local/oushu/kerberos/Kafka/krb5.conf -Djava.security.auth.login.config=/usr/local/oushu/conf/kafka/kafka-jaas.conf\";/usr/local/oushu/kafka/bin/kafka-server-start.sh -daemon /usr/local/oushu/conf/kafka/server.properties"

检查状态

在各节点通过 jps 查看进程：

node	process
kafka1	Kafka
kafka2	Kafka
kafka3	Kafka

常用命令

停止 Kafka 服务

sudo -u kafka /usr/local/oushu/kafka/bin/kafka-server-start.sh -daemon /usr/local/oushu/conf/kafka/server.properties

注册到Skylab（可选）

在 kafka1 节点修改 lava 命令行工具配置中 skylab 的节点 ip

vi /usr/local/oushu/lava/conf/server.json

编写注册 request 到一个文件，例如 ~/kafka-register.json

{
  "data": {
    "name": "KafkaCluster",
    "group_roles": [
      {
        "role": "kafka.broker",
        "cluster_name": "kafka_cluster",
        "group_name": "broker1",
        "machines": [
          {
            "id": 1,
            "name": "kafka1",
            "subnet": "lava",
            "data_ip": "${kafka1}",
            "manage_ip": "${kafka1}",
            "assist_port": 1622,
            "ssh_port": 22
          },
          {
            "id": 2,
            "name": "kafka2",
            "subnet": "lava",
            "data_ip": "${kafka2ip}",
            "manage_ip": "${kafka2ip}",
            "assist_port": 1622,
            "ssh_port": 22
          },
          {
            "id": 3,
            "name": "kafka3",
            "subnet": "lava",
            "data_ip": "${kafka3ip}",
            "manage_ip": "${kafka3ip}",
            "assist_port": 1622,
            "ssh_port": 22
          }
        ]
      }
    ],
    "config": {
      "server.properties": [
        {
          "key":"num.network.threads",
          "value":"3"
        },
        {
          "key":"num.io.threads",
          "value":"8"
        },
        {
          "key":"log.dirs",
          "value":"${dataDir}"
        },
        {
          "key":"num.partitions",
          "value":"1"
        },
        {
          "key":"num.recovery.threads.per.data.dir",
          "value":"1"
        },
        {
          "key":"offsets.topic.replication.factor",
          "value":"1"
        },
        {
          "key":"log.retention.hours",
          "value":"168"
        },
        {
          "key":"log.segment.bytes",
          "value":"1073741824"
        },
        {
          "key":"log.retention.check.interval.ms",
          "value":"300000"
        },
        {
          "key":"zookeeper.connect",
          "value":"localhost:2181"
        },
        {
          "key":"zookeeper.connection.timeout.ms",
          "value":"18000"
        },
        // kerberos 相关配置
        {
          "key": "security.inter.broker.protocol",
          "value": "SASL_PLAINTEXT"
        },
        {
          "key": "sasl.mechanism.inter.broker.protocol",
          "value": "GSSAPI"
        },
        {
          "key": "sasl.enabled.mechanisms",
          "value": "GSSAPI"
        },
        {
          "key": "sasl.kerberos.service.name",
          "value": "Kafka"
        }
      ]
    }
  }
}

上述配置文件中，需要根据实际情况修改 machines 数组中的机器信息，通过如下方式查看，在平台基础组件 lava 所安装的机器执行：

psql lavaadmin -p 4432 -U oushu -c "select m.id,m.name,s.name as subnet,m.private_ip as data_ip,m.public_ip as manage_ip,m.assist_port,m.ssh_port from machine as m,subnet as s where m.subnet_id=s.id;"

获取到所需的机器信息，根据服务角色对应的节点，将机器信息添加到 machines 数组中。

调用 lava 命令注册集群：

lava login -u oushu -p ********
lava onprem-register service -s Kafka -f ~/kafka-register.json

如果返回值为：

Add service by self success

则表示注册成功，如果有错误信息，请根据错误信息处理。

同时，从页面登录后，在自动部署模块对应服务中可以查看到新添加的集群,同时列表中会实时监控 Kafka 进程在机器上的状态。