Apache Kafka Brokerを構築してみる


# はじめに

本ドキュメントでは、試験環境での利用を目的としたApache Kafka Brokerのセットアップ手順を記します。

# 前提条件

- AMD64のコンテナ実行環境
- docker compose
- セットアップするコンテナとは別のJava実行環境 (動作確認で使います)

# Kafka brokerの設定

## セットアップ

Kafka brokerとしてbitnamiが提供しているコンテナイメージを利用する。environmentに指定する項目についてはdocker hubの[bitnami/kafka Overview](https://hub.docker.com/r/bitnami/kafka)ページに詳細の記載がある。

下記は、コンテナ内部からの通信用として9092、コンテナ外部からの通信用として9093を指定している。またSecurity ProtocolとしてPLAINTEXTを使っている。暗号化、パスワード認証等の設定を行なっていないため、試験環境での利用目的としてご利用下さい。

IPアドレスの箇所は適宜ホストのIPアドレスとして読み替えて実行して下さい。

```yaml
version: "2"

services:
  zookeeper:
    image: docker.io/bitnami/zookeeper:3.8
    ports:
      - "2181:2181"
    volumes:
      - "zookeeper_data:/bitnami"
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
  kafka:
    image: docker.io/bitnami/kafka:3.4
    ports:
      - "9092:9092"
      - "9093:9093"
    volumes:
      - "kafka_data:/bitnami"
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      - KAFKA_CFG_LISTENERS=INTERNAL://:9092,EXTERNAL://:9093
      - KAFKA_CFG_ADVERTISED_LISTENERS=INTERNAL://192.168.1.3:9092,EXTERNAL://192.168.1.3:9093
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=INTERNAL
    depends_on:
      - zookeeper

volumes:
  zookeeper_data:
    driver: local
  kafka_data:
    driver: local
```

## コンテナ実行状態の確認

```sh
docker ps -a
CONTAINER ID   IMAGE                                COMMAND                  CREATED          STATUS          PORTS                                                                     NAMES
37b52d5c953a   bitnami/kafka:3.4                    "/opt/bitnami/script…"   31 seconds ago   Up 29 seconds   0.0.0.0:9092-9093->9092-9093/tcp, :::9092-9093->9092-9093/tcp             kafka-kafka-1
95b691c7f378   bitnami/zookeeper:3.8                "/opt/bitnami/script…"   33 minutes ago   Up 33 minutes   2888/tcp, 3888/tcp, 0.0.0.0:2181->2181/tcp, :::2181->2181/tcp, 8080/tcp   kafka-zookeeper-1
```

## kafka clientを使った通信テスト

二つのターミナルからkafka brokerコンテナにログインし、それぞれProducerとConsumerを起動し動作を確認する。

Producerの起動。`>`が表示された後に文字列をタイプしてEnterを押下すると、Consumerにメッセージが送信されます。`--broker-list`には、kafka brokerのIPアドレスとポートを指定します。Listとなっているように、複数台のbrokerを指定可能です。

```sh
docker exec -it {bitnami/kafka Container ID} bash
/opt/bitnami/kafka/bin/kafka-console-producer.sh --broker-list 192.168.1.3:9092 --topic test
```

Consumerの起動。下記の例では、`--from-beginning`をつけているので、ストアされている既存のメッセージから取得します。`--from-beginning`をつけない場合は、起動したタイミング以降のメッセージを受信します。

```sh
docker exec -it {bitnami/kafka Container ID} bash
/opt/bitnami/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.3:9092 --topic test  --from-beginning
```

## コンテナ外からのメッセージの送受信

### Macの場合
コンテナとは別のJava実行環境にて [apach kafka (Binary)](https://kafka.apache.org/downloads) をダウンロードし、解凍後、ディレクトリへ移動する。

```sh
tar xvof kafka_2.13-3.4.0.tgz
cd kafka_2.13-3.4.0
```

Consumerの起動。`--topic`では受信したいトピック名を指定すること。コンテナ内のProducerからメッセージ送信し、別のJava実行環境 (私の場合はm1 mac) にてメッセージを受信できることを確認します。

```sh
bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.3:9093 --topic test
```

### Windowsの場合
コンテナとは別のJava実行環境にて [apach kafka (Binary)](https://kafka.apache.org/downloads) をダウンロードし、解凍後、Cドライブ直下などの出来るだけ浅い階層にフォルダごと移動する。
あわせて、フォルダ名も `kafka_2.13-3.4.0` から `kafka` にリネームしておく。

Consumerの起動。`--topic`では受信したいトピック名を指定すること。コンテナ内のProducerからメッセージ送信し、別のJava実行環境にてメッセージを受信できることを確認します。

```powershell
cd C:\kafka
bin\windows\kafka-console-producer.bat --topic test --bootstrap-server 192.168.1.3:9093
```


## Vantiqとkafka brokerの接続

Vantiq kafka sourceのproducerとしての設定。ネットワーク的に疎通可能であることが条件となります。クラウド環境にkafka brokerを構築している場合、インターネットから9093でインバウンドアクセスできるようにセキュリティ設定を行う必要があります。

```json
{
    "bootstrap.servers": "http://192.168.1.3:9093",
    "contentType": "application/json"
}
```

Vantiq kafka sourceのconsumerとしての設定。セキュリティ設定の条件は、producerと同じです。`--from-beginning`が付いた状態での接続となります。またVantiq kafka sourceはjsonでの受信を想定して構成されています。`hello world`のようなメッセージを送信したtopicに接続するとエラーが出力されることになりますのでご注意下さい。

```json
{
    "consumer.topics": [
        "test"
    ],
    "bootstrap.servers": "sb://192.168.1.3:9093",
    "contentType": "application/json"
}
```

# 参考情報

[Kafka Listeners - Explained](https://rmoff.net/2018/08/02/kafka-listeners-explained/)