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

## 前提
* Kafka 部署需要依赖 Zookeeper 集群。用于元数据管理。 Zookeeper 安装部署请参考：[Zookeeper 安装](../zookeeper/installation-zookeeper.rst)。

### 安装
首先登录到 kafka1，然后切换到 root 用户
``` sh
ssh kafak1
su - root
```

创建一个`kafkahosts`文件，包含 Kafka 集群中所有的机器
``` sh
cat > ${HOME}/kafkahosts << EOF
kafka1
kafka2
kafka3
EOF
```

在 kafka1 节点配置 yum 源，安装 lava 命令行管理工具
```sh
# 从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 免密码登陆和分发配置文件。
```sh
lava ssh-exkeys -f ${HOME}/kafkahosts -p ********
```

分发 repo 文件到其他机器
```sh
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
```sh
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
```sh 
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`
``` sh
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 下

### 分发配置文件到其他机器
```sh
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 服务
```sh
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 节点执行如下命令
```sh
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 服务
``` sh
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

```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 命令（从 lava 6.3.0.0 开始，lava login 支持交互式输入密码）
lava login -u oushu -T {租户id}
lava onprem-register service -s Kafka -f ~/kafka-register.json
```

如果返回值为：
```
Add service by self success
```
则表示注册成功，如果有错误信息，请根据错误信息处理。

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