Kafka を Docker で動かしてリハビリする

同僚が困っていたので、支援することになって、久々に Docker をやる事になったので、リハビリ兼ねて Kafka をビデオで学んで、最終的に、k8s の上で動かす helm chart を書くところまでしようと思っている。

<img width="1485" alt="Screen Shot 2019-02-24 at 10.44.19 PM.png" src="https://qiita-image-store.s3.amazonaws.com/0/3470/b95b7436-3a79-de1d-f7c1-30c49c9ebfeb.png">


最近C＃のコーディングばかりだったので、しばらく触っていないので、思い出すためにもちょっと動かしてみた。kafka は Zookeeper と連係して、メッセージを捌くメッセージングの仕組みで、Azureでいうと、EventHubなんだけど、Scalaで出来ているっぽく、細かい制御が出来そう。最近EventHubがKafkaのインターフェイスを持つようになったので、最終的に連係させて、Azureのサービスを使いやすくしてみたい。

# Docker を思い出す

最初にイメージを探したが、なんと公式イメージが無い。Spotify とかの野良のイメージはあるけど、公式は無い。今回は、kafka に多大な貢献するをしている、confluent のイメージを使ってみたい。

* [Docker Configuration Parameters](https://docs.confluent.io/current/installation/docker/docs/config-reference.html)
* [cp-docker-images/debian/kafka/Dockerfile](https://github.com/confluentinc/cp-docker-images/blob/5.1.2-post/debian/kafka/Dockerfile)

このイメージは、DockerHubで見るとDockerfileへのリンクがなくて見つけにくかったが、上記のリンクのところにある。パラメータの設定がたくさん必要なので、素直にconfluentのページを見るのが良さげ（上記のリンク）

## Zookeeper を起動する

これは簡単で、次の通り。kafka は Zookeeper に依存している。上記のConfluent のページのまんま。

```
docker run -d \
--net=host \
--name=zookeeper \
-e ZOOKEEPER_CLIENT_PORT=32181 \
-e ZOOKEEPER_TICK_TIME=2000 \
-e ZOOKEEPER_SYNC_LIMIT=2 \
confluentinc/cp-zookeeper:5.1.2
```

設定を確認すると、想像がつくけど、TICK_TIME と SYNC_LIMIT がわからない。TICK_TIME １回のTickの間隔。SYNC_LIMITは、ZooKeeperでフォロワーがリーダーにシンクする間隔。Tickの単位。つまり、4000 milliseconds 詳細はこちら。

* [ZooKeeper Administrator's Guide](https://zookeeper.apache.org/doc/r3.3.3/zookeeperAdmin.html#id_tickTime)

## kafka を起動する

これも、confluent をそのままで、

```
docker run -d \
    --net=host \
    --name=kafka \
    -e KAFKA_ZOOKEEPER_CONNECT=localhost:32181 \
    -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:29092 \
    -e KAFKA_BROKER_ID=2 \
    -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
    confluentinc/cp-kafka:5.1.2
```

これもだいたい想像がつく。ブローカのIDが２で、REPLICATION_FACTOR＝１なのでシングル構成。あとは、リスナーのポートと、Zookeeperのアドレス。

## Topicを作成する

Kafkaは全く知らんので、こちらのチュートリアルにしたがって見る。

* [kafka Quick start](https://kafka.apache.org/quickstart)

まともにインストールすると超めんどくさい。（Scalaとか入れないといけない）だから、Dockerでなんとかしたい。

```
$ docker run -it --network host confluentinc/cp-kafka:5.1.2 /bin/bash
```

インタラクションモードで起動している。 network を host にしているのは、ホストコンピュータのポートをコンテナから参照したいから。上記のサンプルの`docker run` でもそうなっている。まぁ手抜きなのだが、手元で実行して試すためにはいいだろう。ちなみに、network の設定の、bridge(default) と host の違いは、こちらの StackOverflow で丁寧に解説してくれている人がいた。

* [From inside of a Docker container, how do I connect to the localhost of the machine?](https://stackoverflow.com/questions/24319662/from-inside-of-a-docker-container-how-do-i-connect-to-the-localhost-of-the-mach)

次にTopic を作成する。cp-kafka のイメージでは、上記のQuickStartで使われているシェルが、`/usr/bin`の下に配置されてる。それを実行してみよう。レプリケーションファクター１、パーティション、トピック名が test という単純な設定

```
root@linuxkit-025000000001:/# cd /usr/bin
root@linuxkit-025000000001:/usr/bin# ./kafka-topics --create --zookeeper localhost:32181 --replication-factor 1 --partitions 1 --topic test
Created topic "test".
```

## Producer でメッセージを送る

Consumer 本来は、 Java か Scala かで書くのだが、すでにサンプルが作られているので、動かしてみよう。先ほどTopicを作ったコンテナでそのまま実行する。

```
root@linuxkit-025000000001:/usr/bin# ./kafka-console-producer --broker-list localhost:29092 --topic test
>This is message
>This is another message
>
```

これで、トピックにメッセージが２つ送信された。まだ読み込みはされていないので、カーソルは一番最初のまま。ポートは、上記のものに合わせている。

## Consumer を起動する

例によってインストールが面倒なので、Dockerで動かそう。

```
$ docker run -it --network host confluentinc/cp-kafka:5.1.2 /bin/bash
root@linuxkit-025000000001:/#  cd /usr/bin
root@linuxkit-025000000001:/usr/bin# kafka-console-consumer --bootstrap-server localhost:29092 --topic test --from-beginning
This is message
This is another message
```
しっかりとメッセージがコンシュームされた。デフォルトでは、読み込みは、オートコミットモードなので、自動的にコミットされている。

# コードを読む

Producerと、Consumerのコードは多分単純だからみてみよう。コンソールを読みこんで、KafkaProducerのオブジェクトにSendメッセージを送るだけ。単純ね。ちなみに、シャットダウンフックで、クローズされるようになっている。

_ConsoleProducer.scala_

```scala
  def main(args: Array[String]): Unit = {

    try {
        val config = new ProducerConfig(args)
        val reader = Class.forName(config.readerClass).getDeclaredConstructor().newInstance().asInstanceOf[MessageReader]
        reader.init(System.in, getReaderProps(config))

        val producer = new KafkaProducer[Array[Byte], Array[Byte]](producerProps(config))

        Runtime.getRuntime.addShutdownHook(new Thread() {
          override def run() {
            producer.close()
          }
        })

        var record: ProducerRecord[Array[Byte], Array[Byte]] = null
        do {
          record = reader.readMessage()
          if (record != null)
            send(producer, record, config.sync)
        } while (record != null)
    } catch {
      case e: joptsimple.OptionException =>
        System.err.println(e.getMessage)
        Exit.exit(1)
      case e: Exception =>
        e.printStackTrace
        Exit.exit(1)
    }
    Exit.exit(0)
  }

```

* [kafka/core/src/main/scala/kafka/tools/ConsoleProducer.scala](https://github.com/apache/kafka/blob/694da1ac1e77c5b9aafe990def2238e521c0fa1f/core/src/main/scala/kafka/tools/ConsoleProducer.scala#L37-L67)

Consumer 側は、基本的に、Consumerのオブジェクトで、メッセージを受け取るけど、自分でPollingのループを回す必要がある。`receive`メソッドで受け取って、それを出力しているだけですね。ちなみに、このクラスを見るとコンフィグが相当ややこしい感じ。こんな単純なのに、なんでたくさん設定するんだろう。

_ConsoleConsumer.scala_

```scala
  def process(maxMessages: Integer, formatter: MessageFormatter, consumer: ConsumerWrapper, output: PrintStream,
              skipMessageOnError: Boolean) {
    while (messageCount < maxMessages || maxMessages == -1) {
      val msg: ConsumerRecord[Array[Byte], Array[Byte]] = try {
        consumer.receive()
      } catch {
        case _: WakeupException =>
          trace("Caught WakeupException because consumer is shutdown, ignore and terminate.")
          // Consumer will be closed
          return
        case e: Throwable =>
          error("Error processing message, terminating consumer process: ", e)
          // Consumer will be closed
          return
      }
      messageCount += 1
      try {
        formatter.writeTo(new ConsumerRecord(msg.topic, msg.partition, msg.offset, msg.timestamp,
                                             msg.timestampType, 0, 0, 0, msg.key, msg.value, msg.headers), output)
      } catch {
        case e: Throwable =>
          if (skipMessageOnError) {
            error("Error processing message, skipping this message: ", e)
          } else {
            // Consumer will be closed
            throw e
          }
      }
      if (checkErr(output, formatter)) {
        // Consumer will be closed
        return
      }
    }
  }
```

* [kafka/core/src/main/scala/kafka/tools/ConsoleConsumer.scala](https://github.com/apache/kafka/blob/694da1ac1e77c5b9aafe990def2238e521c0fa1f/core/src/main/scala/kafka/tools/ConsoleConsumer.scala#L101-L134)

# まとめ

こんな感じで、簡単にハローワールドレベルを試すことが出来た。次は、AKSで動かして、Helmチャート化やなぁ。