> For the complete documentation index, see [llms.txt](https://advwacloud.gitbook.io/rd-newbie-guide/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://advwacloud.gitbook.io/rd-newbie-guide/kafka/kafka_deploy_note.md).

# Kafka Deploy

## Deploy

建一個namespace, 名子取kafka, namespace不要設quota, quota繼承workspace的 ![](/files/-M4wc0Q3eYYRtegrw9cY)

本地k8s config的namespace指到kafka

去harbor抓kafka chart, 然後解壓縮 (`tar xvf FileName.tar`)

<https://harbor.arfa.wise-paas.com/harbor/projects/68/helm-charts/kafka/versions/1.1.5>

編輯deploy.sh, 下圖圈紅線的地方是已經修改過的

&#x20;![](/files/-M4wc0Q5vTGpGPuyrZkR)

在chart根目錄執行

```
$ ./deploy.sh kafkatest
```

看到下面這個hint, 代表佈成功

![](/files/-M4wc0Q7LxJZPEB32nLM)

接著一直反覆執行 `kubectl get all`, statefulset都佈完成代表所有資源已成功運行

![](/files/-M4wc0Q9W1oYyQoJSkT4)

如果遇到這個錯誤, 代表FIRST\_LISTENER\_PORT已經被占用

```
Error: Service "kafka-0-external" is invalid: spec.ports[0].nodePort: Invalid value: 31695: provided port is already allocated
helm.go:75: [debug] Service "kafka-0-external" is invalid: spec.ports[0].nodePort: Invalid value: 31695: provided port is already allocated
```

此時只要去deploy.sh改成其他數字就好

## 驗證

登入management portal建個topic

平台開的測試環境IP是172.21.92.173, 下kubectl get all找到kafkatest-kafka-manager (NodePort)的port為32556, 接著在瀏覽器訪問172.21.92.173:32556

密碼的話請下指令: helm status kafkatest, 再用提示的指令拿到帳密

```
Kafka Web Manager authentication:
username:
kubectl get secret kafkatest-kafka-manager -n kafka -o jsonpath='{.data.basicAuthUsername}' | base64 -d
password:
kubectl get secret kafkatest-kafka-manager -n kafka -o jsonpath='{.data.basicAuthPassword}' | base64 -d
```

登入portal後, 創個topic ![](/files/-M4wc0QBmCzVzsRy8BPk)

接著開啟Conductor

要先組一個url, 規則為`$KFK_BROKER_POD_NAME.$KFK_BROKER_HEADLESS_SVC_NAME.$NAMESPACE.svc.cluster.local`

以這篇範例來說, 就是

kafkatest-0.kafkatest-kafka-headless.kafka.svc.cluster.local

接著去你電腦的hosts file加上一行

`172.21.92.173 kafkatest-0.kafkatest-kafka-headless.kafka.svc.cluster.local`

![](/files/-M4wc0QDcCVivT8Tb3y2)

接著在Conductor設定zk和kfk broker, external ip請下kubectl get all去找, Additional Properties用下面這段

```
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-secret";
serviceName="Kafka"
```

![](/files/-M4wc0QFX-a894HdbT_B)

接著就可以測試producer/consumer了

![](/files/-M4wc0QHp8gK6NHZz3tR)

## Prometheus Test

要測Prometheus要先請平台的人在站點安裝Prometheus

Prometheus會自動找尋服務, 所以不須其他設定, 就可以看到像下面這張圖 ![](/files/-M4wc0QJElm31r2e_4hm)

接著可以下查詢語法, 取出各項量測指標, 這裡參考Kenny投影片, 查詢特定topic的每秒訊息量統計

`sum(kafka_server_brokertopicmetrics_messagesinpersec_count{job="kafkatest"}) by (topic)`

![](/files/-M4wc0QLGwB2y8rBsV-O)

![](/files/-M4wc0QNiz6IUOHPYuUP)

## 使用Kafka Broker Container內建工具

這裡是補充Kenny投影片裡面`/usr/bin/kafka-console-producer` 跟 `/usr/bin/kafka-console-consumer` 操作細節

#### 進去container

```
kubectl exec -it kafkatest-0 --container kafka-broker -- /bin/bash
```

#### 把container的kafka-run-class複製到本地

```
kubectl cp kafkatest-0:usr/bin/kafka-run-class kafka-run-class --container kafka-broker
```

#### 把本地的修改完的kafka-run-class複製回container

把kafka-run-class下面這三行註解掉

```
# JMX port to use
#if [ $JMX_PORT ]; then
# KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -Dcom.sun.management.jmxremote.port=$JMX_PORT "
#fi
```

```
kubectl cp kafka-run-class kafkatest-0:usr/bin/kafka-run-class --container kafka-broker

## in container
chmod +x /usr/bin/kafka-run-class
```

#### 把producer.properties複製進container

{% code title="producer.properties" %}

```
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
```

{% endcode %}

```
kubectl cp producer.properties kafkatest-0:etc/kafka/producer.properties --container kafka-broker
```

#### 把consumer.properties複製進container

{% code title="consumer.properties" %}

```
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
```

{% endcode %}

```
kubectl cp consumer.properties kafkatest-0:etc/kafka/consumer.properties --container kafka-broker
```

#### 把kafka\_client\_jaas.conf複製進container

{% code title="kafka\_client\_jaas.conf" %}

```
KafkaClient {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="admin-secret";
};
```

{% endcode %}

```
kubectl cp kafka_client_jaas.conf kafkatest-0:etc/kafka/kafka_client_jaas.conf --container kafka-broker
```

#### producer發送

```
export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/kafka/kafka_client_jaas.conf" \
&& ./usr/bin/kafka-console-producer --broker-list 10.233.23.176:9092 --topic roy02 --producer.config /etc/kafka/producer.properties
```

#### consumer接收

```
export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/kafka/kafka_client_jaas.conf" \
&& ./usr/bin/kafka-console-consumer --bootstrap-server 10.233.23.176:9092 --topic roy02 --from-beginning --consumer.config /etc/kafka/consumer.properties
```

#### 補充 - 壓力測試工具

```
export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/kafka/kafka_client_jaas.conf" \
&& ./usr/bin/kafka-producer-perf-test --num-records 10000000 --topic roy02 --throughput 100000 --record-size 100 --producer-props bootstrap.servers=10.233.23.176:9092 --producer.config /etc/kafka/producer.properties
```