docker-composeでKafkaの動作確認をする


Kafka に入門するために docker-compose を使い動作確認環境を作成しました。  
要件は次のとおりです。

- ローカルからアクセスできるようにします
  - port を expose します。もしアクセスするアプリケーションなども docker 上で動かす場合は設定が変わってくると思います
- 個人的に検証したいことがあったため、broker を 2 つ立ち上げます
- [kafka-ui](https://github.com/kafbat/kafka-ui)という Kafka を管理できる Web UI を使います
- 動作に必須もしくは変更する必要のあるもの以外の設定値は混乱を避けるためになるべく設定しないようにします

## 完成品

検索したヒットしたものをいろいろツギハギして次のような YAML を作成しました。

```yaml
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.9.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka0:
    image: confluentinc/cp-kafka:7.9.0
    container_name: kafka0
    ports:
      - "9092:9092"
    depends_on:
      - zookeeper
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://kafka0:29092
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT_INTERNAL

  kafka1:
    image: confluentinc/cp-kafka:7.9.0
    container_name: kafka1
    ports:
      - "9093:9093"
    depends_on:
      - zookeeper
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9093,PLAINTEXT_INTERNAL://kafka1:29093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT_INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  kafka-ui:
    container_name: kafka-ui
    image: ghcr.io/kafbat/kafka-ui:v1.2.0
    ports:
      - "8080:8080"
    depends_on:
      - kafka0
    restart: always
    environment:
      KAFKA_CLUSTERS_0_NAME: local-cluster
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka0:29092
```

## 設定値について

### 使用イメージ

各イメージは記事執筆時点で最新でした。

### 環境変数

#### cp-zookeeper

##### ZOOKEEPER_CLIENT_PORT

Kafka からのアクセスをどのポートでリッスンするかを設定します。Kafka クラスターで使う場合、zookeeper ではこの環境変数のみ起動に必須でした。

#### cp-kafka

`cp-kafka`の環境変数についてのドキュメントは

https://docs.confluent.io/platform/current/installation/docker/config-reference.html

にあります。しかし、このページにはどの環境変数がどのような意味かについてあまり書かれていません。  
このページには、`cp-kafka`の`KAFKA_ZOOKEEPER_CONNECT`のような環境変数は`KAFKA_`を取って`_`を`.`に読み替えるということが書いてあります。  
つまり`KAFKA_ZOOKEEPER_CONNECT`は`zookeeper.connect`になるわけですが、この値で検索すれば設定値の意味を調べられます。

Kafka の設定値全般は例えば、

https://docs.confluent.io/platform/current/installation/configuration/broker-configs.html#ak-broker-and-controller-configurations-for-cp

や

https://mogile.web.fc2.com/kafka/documentation_3.html#configuration

にまとまっています。環境変数からの変換方式を把握した上で、このドキュメントを読めば様々な設定が Docker の環境変数経由で可能です。  
Docker でない（本来の？）Kafka はプロパティファイルから設定を取得するわけですが、そちらでは`zookeeper.connect`形式が設定値のキーになっています。

##### KAFKA_ZOOKEEPER_CONNECT

Kafka ブローカーが zookeeper と通信するための接続文字列を指定します。  
Kafka ブローカーが zookeeper という名前のサービス（Docker 内）の 2181 ポートを介して zookeeper に接続することを意味します。

##### KAFKA_LISTENER_SECURITY_PROTOCOL_MAP

Kafka のリスナーとセキュリティプロトコルのマッピングを指定します。

https://access.redhat.com/documentation/ja-jp/red_hat_amq/2021.q2/html/using_amq_streams_on_rhel/con-kafka-listener-configuration-str

によるとリスナーとは、

> 各リスナーは異なるポートまたはネットワークインターフェースをリッスンするために使用され

とあるので、雑ですが、Kafka が通信を受信する場所、のように理解しています。  
セキュリティプロトコルは

https://mogile.web.fc2.com/kafka/documentation_3.html#producerconfigs_security.protocol

によると、

> ブローカーと通信するために使われるプロトコル。有効な値は: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL。

とあるので SSL などを指定して暗号化することもできるようですが、開発環境なので、あまりそのあたりの設定の理解はせず PLAINTEXT を利用しています。

`PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT`とすると PLAINTEXT というリスナーが平文（PLAINTEXT）で通信し、PLAINTEXT_INTERNAL というリスナーも平文（PLAINTEXT）で通信する、という設定になると思います。

##### KAFKA_ADVERTISED_LISTENERS

Kafka ブローカーがクライアントや他のブローカーに対して自分自身をアドバタイズ（公開）するためのアドレスを設定します。  
リスナーは`<listenerName>://<hostname>:_<port>_`のように表記されます。  
つまり、`KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9093,PLAINTEXT_INTERNAL://kafka1:29093`という環境変数で、

- localhost:9093 で接続できる PLAINTEXT というリスナー
- kafka1:29093 で接続できる PLAINTEXT_INTERNAL というリスナー

の 2 つのリスナーがクライアントに公開されるということになります。  
なぜ 2 つ必要かというとローカルと Docker ネットワーク内（kafka-ui、zookeeper、他の broker）からの両方からのアクセスができなければならないからです。  
ローカルからは localhost でアクセスできますが、例えば kafka-ui コンテナからは localhost では他のコンテナにある broker との通信ができません。他のコンテナからアクセスしたい場合は kafka0 などコンテナ名をホスト名に指定してアクセスする必要があります。  
これが advertised.listeners を複数設定している理由だと理解しています。

advertised.listeners は個人的には理解が難しかったです。以下のページなどが参考になるかもしれません。

https://www.confluent.io/blog/kafka-listeners-explained/

https://udomomo.hatenablog.com/entry/2019/03/29/102552

##### KAFKA_INTER_BROKER_LISTENER_NAME

Kafka はブローカー同士でも通信するため、Docker ネットワーク内の他のブローカーにどのリスナーに向けて通信すれば良いかを伝えます。

##### KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR

Kafka が内部的に利用しているトピックである offsets topic の replication factor （どの程度冗長化するか）の設定です。ブローカー数以下である必要があるため、1にしています。

#### kafka-ui

[kafka-ui](https://github.com/kafbat/kafka-ui)とは Kafka を Web UI で確認、操作（トピック作成、メッセージ送受信、メッセージ削除など）できるツールです。  
環境変数による設定は

https://ui.docs.kafbat.io/configuration/misc-configuration-properties

に記載されています。

##### KAFKA_CLUSTERS_0_NAME

クラスター名です。UI上に表示されます。

![kafa-ui ダッシュボード](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273742/db96b65e-3d51-44fc-add3-a803e31f57de.png)

##### KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS

Kafka クラスターにどのアドレスでアクセスすればよいかを設定します。

### 動作確認

Kafka へのメッセージ送受信をできる[kafkacat](https://github.com/edenhill/kcat)を使って動作確認してみます。  
kafkacat は Debian，Ubuntu では

```bash
apt install kafkacat
```

でインストールできますが、Mac では

```bash
brew install kcat
```

でインストールできるようです。私は Ubuntu で動作確認しましたが、Mac の場合は以下の記述の`kafkacat`を`kcat`に置き換える必要があると思います。

#### メッセージ送信

```bash
kafkacat -b localhost:9092 -t new_topic -P
```

このコマンドで対話モードになるため、適当な文字列を入力して Enter するとメッセージが送信されます。なお、Ctrl+D で抜けます。  
`-b`で Bootstrap broker を指定します。Kafka クライアントが Kafka クライアントと通信するときに最初にアクセスするサーバーです。ブローカーのうちの 1 つを指定すればよいです。つまり、

```bash
kafkacat -b localhost:9093 -t new_topic -P
```

としても良いです。Kafka と通信する際に実際に接続するブローカーは最初の接続時のブローカーとは変わる可能性もあります。実際の Kafka クラスターにおいてブローカーは多数あることもありますが、それを全て把握、列挙しなくてもどれがを指定すればよしなに Kafka との通信ができることになります。

また、`-t`でトピックを指定します。デフォルトの設定では事前にトピックを作成しなくても自動で作成されます。  
`-P`で kafkacat をプロデューサーモードで起動します。

メッセージが Kafka に送信できたことは`http://localhost:8080`から kafka-ui にアクセスすると下の画像のように確認できます。

![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/273742/ec28a2d1-764e-ce63-58c7-418fe11b3618.png)

#### メッセージ受信

画面でもメッセージを確認できましたが、kafkacat の場合は、

```bash
kafkacat -b localhost:9092 -t new_topic -C
```

でメッセージを受信できます。  
`-C`でコンシューマーモードを指定します。  
これでトピックのメッセージを待ち受けることができます。